PyQueryableList: LINQ’s Queryable List for Python

Introduction

PyQueryableList is a quick attempt to duplicate some of Microsoft’s LINQ IQueryable natively in Python.

The main class is QueryableList which extends UserList, and accepts data in the constructor.

QueryableList makes use of a decorator returns_new to enable method chaining.

Code

__author__ = "Edward J. Stembler "
__copyright__ = "Copyright (c) 2009, Edward J. Stembler"
__license__ = "MIT"
__date__ = "$Apr 6, 2009 11:48 PM$"
__module_name__ = "Queryable List"
__version__ = "1.0"

from itertools import ifilter
from itertools import imap
from itertools import islice
import UserList


def returns_new(func):
    def inner(self, args):
        return QueryableList(list(func(self, args)))

    return inner


class QueryableList(UserList.UserList):

    def __init__(self, data):
        super(QueryableList, self).__init__()
        self.data = data

    @returns_new
    def take(self, count):
        return islice(self.data, count)

    @returns_new
    def skip(self, count):
        return islice(self.data, count, None)

    @returns_new
    def join(self, stream):
        return ((left, right) for left in self.data for right in stream)

    @returns_new
    def select(self, func):
        return imap(func, self.data)

    @returns_new
    def where(self, func):
        return ifilter(func, self.data)

    def single_or_default(self, func=None):
        try:
            return self.data[0] if func is None else self.where(func).data[0]
        except IndexError:
            return None

    def single(self, func=None):
        result = self.single_or_default(func)
        if result is None or result == []:
            raise ValueError
        return result

    def order_by(self, func):
        self.data.sort(func)
        return QueryableList(self.data)

    def order_by_descending(self):
        return self.order_by(lambda x, y: y - x)

    def contains(self, item):
        return self.where(lambda x: x == item) != []

    def count(self, func=None):
        return len(self.data) if func is None else self.where(func).count()

    def sum(self, func=None):
        return sum(self.data) if func is None else self.where(func).sum()

    def average(self, func=None):
        return self.sum() / len(self.data) if func is None else self.where(func).average()
queryable_list.py

Usage

a = QueryableList(range(20)) # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19]

b = a.take(10).where(lambda x: x < 5).order_by_descending() # Method chaining example

print b # [4, 3, 2, 1, 0]
print b.count() # 5
print b.count(lambda x: x < 2) # 2

print b.contains(4) # True
print b.contains(5) # False

print b.single() # 4
print b.single(lambda x: x < 3) # 2
print b.single_or_default(lambda x: x == 9) # None
try:
    print b.single(lambda x: x == 9) # ValueError
except ValueError:
    print "ValueError"

print b.sum() # 10
print b.sum(lambda x: x > 2) # 7

print b.average() # 2

Source

ejstembler/PyQueryableList

Sliding Sidebar