queryset.py

#

from dataclasses import dataclass

#

TODO: construct QuerySet from data (filters) + iterable, like: QuerySet( iterable, filters=[ ( “AND”, (“age”, “lte”, 30), (“name”, “startswith”, “A”), ), ], ) TODO: create QuerySet class from data (metaclass to create filter methods from JSON) TODO: QuerySet(iterable).filter(RowClass.field_name >= 10) <=> QuerySet(iterable).filter(field_name__gte=10)

import re

#

class FilterExpression:
    slug = None

#

    def __init__(self, value):

#

TODO: should save field name?

        self.value = value
        self.validate()

#

    def validate(self):
        pass

#

    def __str__(self):
        return "<FilterExpression {} {}>".format(self.slug, repr(self.value))

#

    def __repr__(self):
        return "<FilterExpression {} {}>".format(self.slug, repr(self.value))

#

class IsNull(FilterExpression):
    slug = "isnull"

#

    def execute(self, value):
        return value is None

#

class IsNotNull(FilterExpression):
    slug = "isnotnull"

#

    def execute(self, value):
        return value is not None

#

class Equals(FilterExpression):
    slug = "eq"

#

    def execute(self, value):
        return value == self.value

#

class NotEquals(FilterExpression):
    slug = "neq"

#

    def execute(self, value):
        return value != self.value

#

class GreaterThan(FilterExpression):
    slug = "gt"

#

    def execute(self, value):
        return value > self.value

#

class GreaterThanOrEquals(FilterExpression):
    slug = "gte"

#

    def execute(self, value):
        return value >= self.value

#

class LittleThan(FilterExpression):
    slug = "lt"

#

    def execute(self, value):
        return value < self.value

#

class LittleThanOrEquals(FilterExpression):
    slug = "lte"

#

    def execute(self, value):
        return value <= self.value

#

class StartsWith(FilterExpression):
    slug = "startswith"

#

    def validate(self):
        if not isinstance(self.value, str):
            raise ValueError("Value is not str")

#

    def execute(self, value):
        if not isinstance(value, str):
            raise ValueError("Value is not str")
        return value.startswith(self.value)

#

class EndsWith(FilterExpression):
    slug = "endswith"

#

    def validate(self):
        if not isinstance(self.value, str):
            raise ValueError("Value is not str")

#

    def execute(self, value):
        if not isinstance(value, str):
            raise ValueError("Value is not str")
        return value.endswith(self.value)

#

class Regexp(FilterExpression):
    slug = "regexp"

#

    def validate(self):
        self.regexp = re.compile(self.value)

#

    def execute(self, value):
        return bool(self.value.match(value))

#

class QuerySet:

#

    def __init__(self, iterable, filters=None):
        self.__iterable = iterable
        self.__filters = filters or {}

#

TODO: filters may be a list of expressions

#

    def filter(self, **filters):
        new_filters = self.__filters.copy()
        new_filters.update(filters)
        return QuerySet(iterable=self.iterable, filters=new_filters)

#

    def exclude(self, **filters):

#

TODO: implement

        pass

#

    def matches(self, row):
        for filter_name, filter_value in self.__filters.items():
            print(f"test if {row} matches {filter_name}, {filter_value}"

#

TODO: implement

        return True

#

    def __iter__(self):
        return self

#

    def __getitem__(self):
        for row in self.iterable:
            if self.matches(row):
                yield row

#

isnull startswith endswith gt gte lt lte in Q

#

@dataclass
class Expression:
    field_name = None
    operation = None
    value = None


class Not(Expression):

    @classmethod
    def execute(self, op1):
        return not bool(op1)


class And(Expression):

    @classmethod
    def execute(self, op1, op2):
        return bool(op1) and bool(op2)


class Or(Expression):

    @classmethod
    def execute(self, op1, op2):
        return bool(op1) or bool(op2)