Python 列表数据到Django ORM查询

Question

比方说，我有一个列表数据，例如：

data = [
    {'id': 1, 'name': 'brad', 'color': 'red', 'tags': [], 'author': {'name': 'admin'}},
    {'id': 2, 'name': 'sylvia', 'color': 'blue', 'tags': [], 'author': {'name': 'user'}},
    {'id': 3, 'name': 'sylwia', 'color': 'green', 'tags': [], 'author': {'name': 'admin'}},
    {'id': 4, 'name': 'shane', 'color': 'red', 'tags': [], 'author': {'name': 'admin'}},
    {'id': 5, 'name': 'shane', 'color': 'red', 'tags': ['python', 'django'], 'author': {'name': 'user'}}
]

我想让它成为 ORM'able，比如 Django 所做的：

ModelName.objects.filter(color__icontains="gree")

这就是我所做的；

import operator
from collections import namedtuple
from django.core.exceptions import MultipleObjectsReturned, ObjectDoesNotExist


class DataQuerySet:
    """
    Custom ORM for List dict data,
    
    """
    allowed_operations = {
        'gt': operator.gt,
        'lt': operator.lt,
        'eq': operator.eq,
        'icontains': operator.contains
    }

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

    def all(self):
        return self.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

    def get(self, **kwargs):
        """
        >>> DataQuerySet().get(id=3)
        [{'id': 3, 'name': 'sylwia', 'color': 'green'}]
        """
        operation = namedtuple('Q', 'op key value')

        def parse_get(item):
            key, *op = item[0].split('__')
            return operation(self.allowed_operations['eq'], key, item[1])

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

        if len(filtered_data) > 1:
            raise MultipleObjectsReturned(filtered_data)
        elif len(filtered_data) < 1:
            raise ObjectDoesNotExist(kwargs)
        return filtered_data[0]

并使用它：

class DataModel:

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

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


data = [
  {'id': 1, 'name': 'brad', 'color': 'red', 'tags': [], 'author': {'name': 'admin'}},
  {'id': 2, 'name': 'sylvia', 'color': 'blue', 'tags': [], 'author': {'name': 'user'}},
  {'id': 3, 'name': 'sylwia', 'color': 'green', 'tags': [], 'author': {'name': 'admin'}},
  {'id': 4, 'name': 'shane', 'color': 'red', 'tags': [], 'author': {'name': 'admin'}},
  {'id': 5, 'name': 'shane', 'color': 'red', 'tags': ['python', 'django'], 'author': {'name': 'user'}}
]
d = DataModel(data)
print(d.objects.filter(id__gt=2))
print(d.objects.filter(color='green'))
print(d.objects.filter(color__icontains='gree'))
print(d.objects.get(id=1))

以上测试工作正常，但当我们想做更多时似乎遇到了问题：

print(d.objects.filter(tags__in=['python']))
print(d.objects.filter(author__name='admin'))
print(d.objects.filter(author__name__icontains='use'))

Answer 1

最后，我找到了一个很好的模块来处理这种情况，它用 reobject 调用，这是测试：

from reobject.models import Model, Field
from reobject.query.parser import Q as Query

data = [
  {'name': 'brad', 'color': 'red', 'tags': [], 'author': {'name': 'admin'}},
  {'name': 'sylvia', 'color': 'blue', 'tags': [], 'author': {'name': 'user'}},
  {'name': 'sylwia', 'color': 'green', 'tags': [], 'author': {'name': 'admin'}},
  {'name': 'shane', 'color': 'red', 'tags': [], 'author': {'name': 'admin'}},
  {'name': 'shane', 'color': 'red', 'tags': ['python', 'django'], 'author': {'name': 'user'}}
]

class Book(Model):
    name = Field()
    color = Field()
    tags = Field()
    author = Field()

for item in data:
    Book(**item)


Book.objects.all()
Book.objects.get(name='brad')
Book.objects.filter(name='brad')
Book.objects.filter(author__name='admin')
Book.objects.filter(tags__contains='python')
Book.objects.filter(Query(author__name='admin') | Query(author__name='user'))

Meanwhile, it still doesn't support with id or pk fields.
Mybe because it already taken.

Python 列表数据到Django ORM查询

Python List data to Django ORM query

python

django

django-orm