Выполнение шаблона Class.objects.filter (...) в python - PullRequest
0 голосов
/ 09 октября 2019

Я хочу использовать шаблон, используемый в моделях django Model.objects.filter(...) для построения фильтров по данным. Это, вероятно, было бы хорошим вариантом использования панд, но я больше заинтересован в улучшении своего питона (сначала) перед тем, как попробовать.

Если у меня есть следующие данные:

DATA = [
    {'id': 1, 'name': 'brad', 'color':'red'},
    {'id': 2, 'name': 'sylvia', 'color':'blue'},
]

Я хотел бы построить что-то похожее на следующее:

class MyData:
    objects = <something>

И установить objects эквивалент «ModelManager», а затем выполнить фильтрацию оттуда, чтобы я мог вызвать:

MyData.objects.filter(id>1)

И получить:

[
    {'id': 2, 'name': 'sylvia', 'color':'blue'}
]

Конечно, я могу сделать что-то простое:

res = [_ for _ in DATA if _['id'] > 1]

Но меня больше интересует разработка самого шаблона -Тривиальная природа примера просто предназначена для того, чтобы показать, чего я хочу достичь.

Что было бы хорошим, базовым способом сделать это правильно? Вот соответствующий класс в django для него: https://github.com/django/django/blob/master/django/db/models/query.py#L185.

Ответы [ 4 ]

2 голосов
/ 12 октября 2019

Если вам нужен полный опыт django Model, то есть:

  • создайте новый вектор объектов или ввод данных с datapoint = MyData(name='johndoe', color='green', ...), как в django: например, new_user=User(username='johndoe', email='jd@jd.com');
  • используйте MyData.objects для управления объектами, например MyData.objects.filter(color__eq='yellow');

, здесь описывается подход к логике.

Сначала вам нужно в основном наивноObjectManager class:

import collections
import operator
import inspect

class ObjectManager(collections.MutableSet):
    def __init__(self):
        # this will hold a list of all attributes from your custom class, once 
        # initiated
        self._object_attributes = None
        self._theset = set()
    def add(self, item):
        self._theset.add(item)
    def discard(self, item):
        self._theset.discard(item)
    def __iter__(self):
        return iter(self._theset)
    def __len__(self):
        return len(self._theset)
    def __contains__(self, item):
        try:
            return item in self._theset
        except AttributeError:
            return False

    def set_attributes(self, an_object):
        self._object_attributes = [
            a[0] for a in  inspect.getmembers(
                an_object, lambda a:not(inspect.isroutine(a))
            ) if not(a[0].startswith('__') and a[0].endswith('__'))
            ]

    def filter(self, **kwargs):
        """Filters your objects according to one or several conditions

        If several filtering conditions are present you can set the 
        combination mode to either 'and' or 'or'.
        """
        mode = kwargs.pop('mode', 'or')
        ok_objects = set()
        for kw in kwargs:
            if '__' in kw:
                _kw, op = kw.split('__')
                # only allow valid operators
                assert op in ('lt', 'le', 'eq', 'ne', 'ge', 'gt')
            else:
                op = 'eq'
                _kw = kw
            _oper = getattr(operator, op)
            # only allow access to valid object attributes
            assert _kw in self._object_attributes
            n_objects = (
                obj for obj in self 
                if _oper(getattr(obj, _kw), kwargs[kw])
                )
            if mode == 'and':
                if n_objects:
                    ok_objects = ok_objects.intersection(n_objects)\
                        if ok_objects else set(n_objects)
                else:
                    return set()

            else:
                ok_objects.update(n_objects)
        return ok_objects

    # feel free to add a `get_or_create`, `create`, etc. 

Теперь вы присоединяете экземпляр этого класса в качестве атрибута к вашему MyData классу и убедитесь, что все новые объекты добавлены в него:

class MyData:
    # initiate the object manager
    objects = ObjectManager()

    def __init__(self, uid, name, color):
        self.uid = uid
        self.name = name
        self.color = color

        # populate the list of query-able attributes on creation
        # of the first instance
        if not len(self.objects):
            self.objects.set_attributes(self)
        # add any new instance to the object manager
        self.objects.add(self)

Теперь вы можете импортировать вектор объектов:

DATA = [
    {'uid': 1, 'name': 'brad', 'color':'red'},
    {'uid': 2, 'name': 'sylvia', 'color':'blue'},
]
for dat in DATA:
    myData(**dat)

или создавать новые экземпляры:

d1 = MyData(uid=10, name='john', color='yellow')

и использовать менеджер для фильтрации ваших объектов:

print([md.name for md in MyData.objects.filter(uid__ge=10)])
# > ['john']
print([md.name for md in MyData.objects.filter(mode='and',uid__ge=1,name__eq='john')])
# > ['john']
print([md.name for md in MyData.objects.filter(mode='or',uid__le=4,name__eq='john')])
# > ['john', 'brad', 'sylvia']


Если вы не можете или не хотите менять класс, для которого вам нужен диспетчер объектов, и вы готовы подлатать патчи (обратите внимание, что я не рекламирую это!), Вы можетедаже создать ObjectManager, который можно подключить к произвольному классу (хотя встроенные типы не будут работать) после определения или дажеинициация некоторых экземпляров.

Идея состоит в том, чтобы обезьяньим патчем __init__ целевого класса и добавить атрибут objects при инициализации экземпляра вашего ObjectManager:

import gc
import inspect
import collections
import operator
import wrapt  # not standard lib > pip install wrapt

class ObjectManager(collections.MutableSet):
    def __init__(self, attach_to):
        self._object_attributes = None
        # add self as class attribute
        attach_to.objects = self
        # monkey patch __init__ of your target class
        @wrapt.patch_function_wrapper(attach_to, '__init__')
        def n_init(wrapped, instance, args, kwargs):
            wrapped(*args, **kwargs)
            c_objects = instance.__class__.objects
            if not c_objects:
                c_objects.set_attributes(instance)
            c_objects.add(instance)
        # make sure to be up to date with the existing instances
        self._theset = set(obj for obj in gc.get_objects() if isinstance(obj, attach_to))
        # already fetch the attributes if instances exist
        if self._theset:
            self.set_attributes(next(iter(self._theset)))
        ...
        # the rest is identical to the version above

Итак, вот как вы бы это использовали:

class MyData:

    def __init__(self, uid, name, color):
        self.uid = uid
        self.name = name
        self.color = color

# create some instances
DATA = [
    {'uid': 1, 'name': 'brad', 'color':'red'},
    {'uid': 2, 'name': 'sylvia', 'color':'blue'},
]
my_datas = []
for dat in DATA:
    my_datas.append(myData(**dat))  # appending them just to have a reference
# say that ONLY NOW you decide you want to use an object manager
# Simply do:
ObjectManager(MyData)
# and you are done:
print([md.name for md in MyData.objects.filter(mode='or',uid__le=4,name__eq='john')])
# > ['brad', 'sylvia']
# also any object you create from now on is included:
d1 = MyData(uid=10, name='john', color='yellow')
print([md.name for md in MyData.objects.filter(mode='or',uid__le=4,name__eq='john')])
# > ['brad', 'sylvia', 'john']
1 голос
/ 17 октября 2019

ОП хочет сделать это MyData.objects.filter(id>1).

Посмотрим правде в глаза.

Проблема в том, что Python жадный (охотно оценивает выражения), а не ленивый, как Haskell.
Смотреть Дэвид Бизли - Лямбда-исчисление с нуля - PyCon 2019 для умопомрачительной вещи.

Python оценивает id > 1 перед вызовом filter. Если мы можем остановить оценку на данный момент, мы можем передать выражение без оценки в функцию filter.

Но мы можем отложить оценку выражения до тех пор, пока мы не вложим выражение в функцию. Это идея.

Интерфейс функции будет filter(lambda: id > 1), если мы сможем его реализовать. Этот интерфейс будет очень универсальным, потому что любое выражение Python может быть передано и использовано неправильно.

Реализация;

, если мы вызываем лямбду или любую другую функцию с выражением id > 1, Python ищетимя id в локальной, включающей, глобальной области или builtins в зависимости от контекста, в котором вызывается функция.

Если мы можем представить объект с именем id где-нибудь вдо того, как Python найдет id в builtins, мы можем переопределить семантику выражения.

Я собираюсь сделать это с eval, который оценивает выражения в данном контексте.

DATA = [
    {'id': 1, 'name': 'brad', 'color':'red'},
    {'id': 2, 'name': 'sylvia', 'color':'blue'},
]

def myfilter(a_lambda):
    return filter(lambda obj: eval(a_lambda.__code__, obj.copy()),
    DATA)

Я передаю dict.copy в eval, потому что eval изменяет его globals object.

Просмотр его в действии в контексте Model class

In [1]: class Data(Model):
   ...:     name = str()
   ...:     id = int()
   ...:     color = str()
   ...: 

In [2]: Data.objects.create(**{"id": 1, "name": "brad", "color": "red"})

In [3]:     Data.objects.create(**{"id": 2, "name": "sylvia", "color": "blue"})

In [4]:     Data.objects.create(**{"id": 3, "name": "paul", "color": "red"})

In [5]:     Data.objects.create(**{"id": 4, "name": "brandon", "color": "yello"})

In [6]:     Data.objects.create(**{"id": 5, "name": "martin", "color": "green"})

In [7]:     Data.objects.create(**{"id": 6, "name": "annie", "color": "gray"})

In [8]: pprint([vars(obj) for obj in Data.objects.filter(lambda: id == 1)])
[{'color': 'red', 'id': 1, 'name': 'brad'}]

In [9]: pprint([vars(obj) for obj in Data.objects.filter(lambda: 1 <= id <= 2)])
[{'color': 'red', 'id': 1, 'name': 'brad'},
 {'color': 'blue', 'id': 2, 'name': 'sylvia'}]

In [10]: pprint([vars(obj) for obj in Data.objects.filter(lambda: color == "blue")])
[{'color': 'blue', 'id': 2, 'name': 'sylvia'}]

In [11]: pprint([vars(obj) for obj in Data.objects.filter(lambda: "e" in color and (name is "brad" or name is "sylvia"))])
[{'color': 'red', 'id': 1, 'name': 'brad'},
 {'color': 'blue', 'id': 2, 'name': 'sylvia'}]

In [12]: pprint([vars(obj) for obj in Data.objects.filter(lambda: id % 2 == 1)])
[{'color': 'red', 'id': 1, 'name': 'brad'},
 {'color': 'red', 'id': 3, 'name': 'paul'},
 {'color': 'green', 'id': 5, 'name': 'martin'}]

Класс Data наследуется от Model. Model дает Data метод __init__ и атрибут класса с именем objects, который указывает на экземпляр MetaManager, который является дескриптором.

MetaManager возвращает экземпляр Manager для подклассов Model при доступе к атрибуту objects из подкласса. MetaManger идентифицирует класс доступа и передает его в экземпляр Manager. Manager обрабатывает создание, сохранение и выборку объектов.

БД реализована как атрибут класса Manager для простоты.

Чтобы остановить злоупотребление глобальными объектами через функции, функция filter вызывает исключение, если лямбда не передается.

from collections import defaultdict
from collections.abc import Callable


class MetaManager:
    def __get__(self, obj, objtype):
        if obj is None:
            return Manager(objtype)
        else:
            raise AttributeError(
                "Manger isn't accessible via {} instances".format(objtype)
            )


class Manager:
    _store = defaultdict(list)

    def __init__(self, client):
        self._client = client
        self._client_name = "{}.{}".format(client.__module__, client.__qualname__)

    def create(self, **kwargs):
        self._store[self._client_name].append(self._client(**kwargs))

    def all(self):
        return (obj for obj in self._store[self._client_name])

    def filter(self, a_lambda):
        if a_lambda.__code__.co_name != "<lambda>":
            raise ValueError("a lambda required")

        return (
            obj
            for obj in self._store[self._client_name]

            if eval(a_lambda.__code__, vars(obj).copy())
        )


class Model:
    objects = MetaManager()

    def __init__(self, **kwargs):
        if type(self) is Model:
            raise NotImplementedError

        class_attrs = self.__get_class_attributes(type(self))

        self.__init_instance(class_attrs, kwargs)

    def __get_class_attributes(self, cls):
        attrs = vars(cls)
        if "objects" in attrs:
            raise AttributeError(
                'class {} has an attribute named "objects" of type "{}"'.format(
                    type(self), type(attrs["objects"])
                )
            )
        attrs = {
            attr: obj
            for attr, obj in vars(cls).items()
            if not attr.startswith("_") and not isinstance(obj, Callable)
        }
        return attrs

    def __init_instance(self, attrs, kwargs_dict):
        for key, item in kwargs_dict.items():
            if key not in attrs:
                raise TypeError('Got an unexpected key word argument "{}"'.format(key))
            if isinstance(item, type(attrs[key])):
                setattr(self, key, item)
            else:
                raise TypeError(
                    "Expected type {}, got {}".format(type(attrs[key]), type(item))
                )


if __name__ == "__main__":
    from pprint import pprint

    class Data(Model):
        name = str()
        id = int()
        color = str()

    Data.objects.create(**{"id": 1, "name": "brad", "color": "red"})
    Data.objects.create(**{"id": 2, "name": "sylvia", "color": "blue"})
    Data.objects.create(**{"id": 3, "name": "paul", "color": "red"})
    Data.objects.create(**{"id": 4, "name": "brandon", "color": "yello"})
    Data.objects.create(**{"id": 5, "name": "martin", "color": "green"})
    Data.objects.create(**{"id": 6, "name": "annie", "color": "gray"})

    pprint([vars(obj) for obj in Data.objects.filter(lambda: id == 1)])
    pprint([vars(obj) for obj in Data.objects.filter(lambda: 1 <= id <= 2)])
    pprint([vars(obj) for obj in Data.objects.filter(lambda: color == "blue")])
    pprint(
        [
            vars(obj)
            for obj in Data.objects.filter(
                lambda: "e" in color and (name is "brad" or name is "sylvia")
            )
        ]
    )
    pprint([vars(obj) for obj in Data.objects.filter(lambda: id % 2 == 1)])
1 голос
/ 12 октября 2019

Ниже приведен пример, где я создаю новый класс NoteQuerySet, который наследуется от django.db.models.QuerySet. После этого я использую метод as_manager, благодаря которому менеджер объектов переопределяется, сохраняя все операции, которые должен иметь менеджер.

Итак, чтобы получить желаемые результатыЯ создал новый метод custom_filter, который работает над NoteQuerySet.data и использует словарь для отслеживания и упрощения добавления новых фильтров.

Как видите, я создаюновый custom_filter вместо переопределения objects.filter;это сделано намеренно, поэтому вы не потеряете встроенную фильтрацию. Также обратите внимание на встроенный модуль operator для простого сопоставления строк с операциями.

models.py

import operator

from collections import namedtuple

from django.db import models


class NoteQuerySet(models.QuerySet):
    data =  [
        {'id': 1, 'name': 'brad', 'color':'red'},
        {'id': 2, 'name': 'sylvia', 'color':'blue'},
        {'id': 3, 'name': 'sylwia', 'color':'green'},
        {'id': 4, 'name': 'shane', 'color':'red'},
    ]
    allowed_operations = {'gt': operator.gt, 'lt': operator.lt, 'eq': operator.eq}

    def custom_filter(self, **kwargs):
        """
        >>> kwargs = {'name': 'sylwia', 'id__gt': 1}
        dict_items([('name', 'sylwia'), ('id__gt', 1)])
        """
        operation = namedtuple('Q', 'op key value')
        def parse_filter(item):
            """item is expected to be a tuple with exactly two elements
            >>> parse_filter(('id__gt', 2))
            Q(op=<built-in function gt>, key='id', value=2)
            """
            key, *op = item[0].split('__')
            # no value after __ means exact value query, e.g. name='sylvia'
            op = op or ['eq']
            return operation(self.allowed_operations[op[0]], key, item[1])

        filtered_data = self.data.copy()
        for item in map(parse_filter, kwargs.items()):
            filtered_data = [
                entry for entry in filtered_data if item.op(entry[item.key], item.value)
            ]
        return filtered_data


class Note(models.Model):
    text = models.CharField(max_length=250)

    objects = NoteQuerySet.as_manager()

Вся логика до сих пор реализована вmodels модуль. Далее возможный вариант использования показан в ListView.

views.py

from django.views.generic import ListView

from .models import Note


class ResultsApplicationView(ListView):
    model = Note
    template_name = 'results.html'

    def get_context_data(self, **kwargs):
        kwargs = super().get_context_data(**kwargs)
        if 'extra' not in kwargs:
            kwargs['extra'] = self.model.objects.custom_filter(id__lt=3, color='red')
        return kwargs

results.html

  <h1>Notes</h1>
  {% for note in object_list %}
    {{note}}
  {% endfor %}

  {{ extra }}

ОБНОВЛЕНИЕ: Реализация без Django:

import operator

from collections import namedtuple


class DataQuerySet:
    allowed_operations = {'gt': operator.gt, 'lt': operator.lt, 'eq': operator.eq}

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

    def filter(self, **kwargs):
        """
        >>> kwargs = {'name': 'sylwia', 'id__gt': 1}
        >>> DataQuerySet().filter(**kwargs)
        [{'id': 3, 'name': 'sylwia', 'color': 'green'}]
        """
        operation = namedtuple('Q', 'op key value')
        def parse_filter(item):
            """item is expected to be a tuple with exactly two elements
            >>> parse_filter(('id__gt', 2))
            Q(op=<built-in function gt>, key='id', value=2)
            >>> parse_filter(('id__  ', 2))
            Q(op=<built-in function eq>, key='id', value=2)
            >>> parse_filter(('color__bad', 'red'))
            Traceback (most recent call last):
             ...
            AssertionError: 'bad' operation is not allowed
            """
            key, *op = item[0].split('__')
            # no value after __ means exact value query, e.g. name='sylvia'
            op = ''.join(op).strip() or 'eq'
            assert op in self.allowed_operations, f'{repr(op)} operation is not allowed'
            return operation(self.allowed_operations[op], key, item[1])

        filtered_data = self.data.copy()
        for item in map(parse_filter, kwargs.items()):
            filtered_data = [
                entry for entry in filtered_data if item.op(entry[item.key], item.value)
            ]
        return filtered_data


class Data:

    def __init__(self, data):
        self._data = DataQuerySet(data)

    @property
    def objects(self):
        return self._data


if __name__ == '__main__':
    data = [
        {'id': 1, 'name': 'brad', 'color':'red'},
        {'id': 2, 'name': 'sylvia', 'color':'blue'},
        {'id': 3, 'name': 'sylwia', 'color':'green'},
        {'id': 4, 'name': 'shane', 'color':'red'},
    ]
    d = Data(data)
    print(d.objects.filter(id__gt=2))
    print(d.objects.filter(color='green'))
1 голос
/ 12 октября 2019

Это то, что вы имеете в виду?

Это решение не зависит от внешней библиотеки и использует ** kwargs, генераторы / замыкания и декоратор @property. Так что с точки зрения обучения это может быть интересно.

Если вам удастся использовать Django для чтения данных, которые есть в вашем списке, то это, вероятно, будет гораздо лучше в отношении совместимости с Django в качестве моего кода. Все зависит от вашей цели. (Идеальная имитация фильтров django) или (изучение того, как сделать не очень идеальную имитацию, но иметь весь исходный код без зависимостей)

DATA = [
    {'id': 1, 'name': 'brad',    'color':'red'},
    {'id': 2, 'name': 'sylvia',  'color':'blue'},
    {'id': 3, 'name': 'paul',    'color':'red'},
    {'id': 4, 'name': 'brandon', 'color':'yello'},
    {'id': 5, 'name': 'martin',  'color':'green'},
    {'id': 6, 'name': 'annie',  'color':'gray'},
]

class UnknownOperator(Exception):
    """ custom exception """

class FilterData:
    def __init__(self, data):
        self.data = data

    def _filter_step(self, key, value, data):
        if not "__" in key:
            return (entry for entry in data if entry[key] == value)
        else:
            key, operator = key.split("__")
            if operator == "gt":  # greater than
                return (entry for entry in data if entry[key] > value)
            elif operator == "lt":  # less than
                return (entry for entry in data if entry[key] < value)
            elif operator == "startswith":  # starts with
                return (entry for entry in data if entry[key].startswith(value))
            elif operator == "in":  # starts with
                return (entry for entry in data if entry[key] in value)
            else:
                raise UnknownOperator("operator %s is unknown" % operator)

    def _exclude_step(self, key, value, data):
        if not "__" in key:
            return (entry for entry in data if entry[key] != value)
        else:
            key, operator = key.split("__")
            if operator == "gt":  # greater than
                return (entry for entry in data if entry[key] <= value)
            elif operator == "lt":  # less than
                return (entry for entry in data if entry[key] >= value)
            elif operator == "startswith":  # starts with
                return (entry for entry in data if not entry[key].startswith(value))
            elif operator == "in":  # starts with
                return (entry for entry in data if entry[key] not in value)
            else:
                raise UnknownOperator("operator %s is unknown" % operator)


    def filter(self, **kwargs):
        data = (entry for entry in self.data)
        for key, value in kwargs.items():
            data = self._filter_step(key, value, data)

        return FilterData(data)

    def exclude(self, **kwargs):
        data = (entry for entry in self.data)
        for key, value in kwargs.items():
            data = self._exclude_step(key, value, data)

        return FilterData(data)

    def all(self):
        return FilterData(self.data)

    def count(self):
        cnt = 0
        for cnt, entry in enumerate(self.data, 1):
            pass
        return cnt

    def __iter__(self):
        for entry in self.data:
            yield entry

# make it even more look like django managers / filters
class DataManager:
    def __init__(self, data):
        self.data = data
    @property
    def objects(self):
        return FilterData(self.data)


fdata = FilterData(DATA)

assert [v["id"] for v in fdata.filter(name="paul")] == [3]
assert [v["id"] for v in fdata.filter(color="red")] == [1, 3]
assert [v["id"] for v in fdata.filter(id__gt=2)] == [3, 4, 5, 6]
assert [v["id"] for v in fdata.filter(color__startswith="gr")] == [5, 6]

fmgr = DataManager(DATA)

assert [v["id"] for v in fmgr.objects.filter(name="paul")] == [3]
assert [v["id"] for v in fmgr.objects.filter(color="red")] == [1, 3]
assert [v["id"] for v in fmgr.objects.filter(id__gt=2)] == [3, 4, 5, 6]
assert [v["id"] for v in fmgr.objects.filter(color__startswith="gr")] == [5, 6]
assert [v["id"] for v in fmgr.objects.filter(color__startswith="gr", id__lt=6)] == [5]
assert [v["id"] for v in fmgr.objects.filter(color__startswith="gr", id__lt=6)] == [5]

assert [v["id"] for v in fmgr.objects.filter(color__startswith="gr").filter(id__lt=6)] == [5]

assert fmgr.objects.filter(color__startswith="gr").filter(id__lt=6).count() == 1
assert fmgr.objects.filter(id__gt=2).count() == 4
assert fmgr.objects.count() == 6
assert [v["id"] for v in fmgr.objects.all()] == list(range(1, 7))
Добро пожаловать на сайт PullRequest, где вы можете задавать вопросы и получать ответы от других членов сообщества.
...