queryish

一个遵循Django的QuerySet API在任意数据源上构造查询的Python库。

动机

Django的QuerySet API是构造数据库查询的强大工具。它允许您逐步组合查询，只有在需要结果时才会执行查询

books = Book.objects.all()
python_books = books.filter(topic='python')
latest_python_books = python_books.order_by('-publication_date')[:5]
print(latest_python_books)  # Query is executed here

这种模式非常适合构建数据列表的Web界面，因为它允许将过滤、排序和分页作为独立的步骤处理。

我们可能需要实现类似接口来处理来自数据库之外的数据源，如REST API或搜索引擎。在这些情况下，我们希望有一个类似的丰富API来构建查询这些数据源。更好的是尽可能遵循QuerySet API，这样我们就可以利用专门为此API设计的现成工具，如 Django的通用基于类的视图。

queryish 是一个库，用于围绕数据源构建包装器以复制QuerySet API，让您可以像使用查询集和模型一样处理数据。

安装

使用pip安装

pip install queryish

用法 - REST API

queryish 提供了一个基类 queryish.rest.APIModel，用于包装REST API。默认情况下，它遵循由 Django REST Framework 提供的现成结构，但提供了各种选项来自定义此结构。

from queryish.rest import APIModel

class Party(APIModel):
    class Meta:
        base_url = "https://demozoo.org/api/v1/parties/"
        fields = ["id", "name", "start_date", "end_date", "location", "country_code"]
        pagination_style = "page-number"
        page_size = 100

    def __str__(self):
        return self.name

生成的类具有一个objects属性，它支持从Django的QuerySet API中熟悉的常规过滤、排序和切片操作，尽管这些可能受到所访问的REST API功能的限制。

>>> Party.objects.count()
4623
>>> Party.objects.filter(country_code="GB")[:10]
<PartyQuerySet [<Party: 16 Bit Show 1991>, <Party: Acorn User Show 1991>, <Party: Anarchy Easter Party 1992>, <Party: Anarchy Winter Conference 1991>, <Party: Atari Preservation Party 2007>, <Party: Commodore Computer Club UK 1st Meet>, <Party: Commodore Show 1987>, <Party: Commodore Show 1988>, <Party: Deja Vu 1998>, <Party: Deja Vu 1999>]>
>>> Party.objects.get(name="Nova 2023")
<Party: Nova 2023>

支持的方法包括all、count、filter、order_by、get、first和in_bulk。结果集可以在任意索引处进行切片——这些索引不必与底层API支持的分页匹配。APIModel将自动根据需要多次发出API请求。

在APIModel.Meta上可用的以下属性：

• base_url：可以从中获取结果的API的基本URL。
• pk_field_name：主键字段的名称。默认为"id"。对"pk"字段名的查找将映射到该字段。
• detail_url：单个对象URL的字符串模板，例如"https://demozoo.org/api/v1/parties/%s/"。如果指定了此值，则对主键和其他字段的查找将指向此URL，而不是base_url。
• fields：API响应中定义的字段名称列表，将复制到返回对象的属性中。
• pagination_style：API使用的分页样式。可识别的值是"page-number"和"offset-limit"；所有其他值（包括默认值None）表示没有分页。
• page_size：如果pagination_style为"page-number"则必需——API返回的每页结果数。
• page_query_param：用于指定页码的URL查询参数名称。默认为"page"。
• offset_query_param：用于指定偏移量的URL查询参数名称。默认为"offset"。
• limit_query_param：用于指定限制的URL查询参数名称。默认为"limit"。
• ordering_query_param：用于指定排序的URL查询参数名称。默认为"ordering"。

为了适应返回的JSON与预期的模型属性不完美映射的API，可以覆盖APIModel上的类方法from_query_data和from_individual_data

class Pokemon(APIModel):
    class Meta:
        base_url = "https://pokeapi.co/api/v2/pokemon/"
        detail_url = "https://pokeapi.co/api/v2/pokemon/%s/"
        fields = ["id", "name"]
        pagination_style = "offset-limit"
        verbose_name_plural = "pokemon"

    @classmethod
    def from_query_data(cls, data):
        """
        Given a record returned from the listing endpoint (base_url), return an instance of the model.
        """
        # Records within the listing endpoint return a `url` field, from which we want to extract the ID
        return cls(
            id=int(re.match(r'https://pokeapi.co/api/v2/pokemon/(\d+)/', data['url']).group(1)),
            name=data['name'],
        )

    @classmethod
    def from_individual_data(cls, data):
        """
        Given a record returned from the detail endpoint (detail_url), return an instance of the model.
        """
        return cls(
            id=data['id'],
            name=data['name'],
        )

    def __str__(self):
        return self.name

自定义REST API查询集类

APIModel子类的objects属性是queryish.rest.APIQuerySet的一个实例，它最初包含完整的记录集。与Django的QuerySet一样，filter等方法返回一个新实例。

可能需要子类化APIQuerySet并覆盖方法以支持某些API响应。例如，基本实现期望未分页的API端点返回作为顶级JSON对象的列表，分页的API端点返回包含results项的字典。如果您正在处理的API返回不同的结构，您可以覆盖get_results_from_response方法以从响应中提取结果列表。

from queryish.rest import APIQuerySet

class TreeQuerySet(APIQuerySet):
    base_url = "https://api.data.amsterdam.nl/v1/bomen/stamgegevens/"
    pagination_style = "page-number"
    page_size = 20
    http_headers = {"Accept": "application/hal+json"}

    def get_results_from_response(self, response):
        return response["_embedded"]["stamgegevens"]

APIQuerySet子类可以独立于APIModel实例化，但结果将返回为纯JSON值。

>>> TreeQuerySet().filter(jaarVanAanleg=1986).first()
{'_links': {'schema': 'https://schemas.data.amsterdam.nl/datasets/bomen/dataset#stamgegevens', 'self': {'href': 'https://api.data.amsterdam.nl/v1/bomen/stamgegevens/1101570/', 'title': '1101570', 'id': 1101570}, 'gbdBuurt': {'href': 'https://api.data.amsterdam.nl/v1/gebieden/buurten/03630980000211/', 'title': '03630980000211', 'identificatie': '03630980000211'}}, 'id': 1101570, 'gbdBuurtId': '03630980000211', 'geometrie': {'type': 'Point', 'coordinates': [115162.72, 485972.68]}, 'boomhoogteklasseActueel': 'c. 9 tot 12 m.', 'jaarVanAanleg': 1986, 'soortnaam': "Salix alba 'Chermesina'", 'stamdiameterklasse': '0,5 tot 1 m.', 'typeObject': 'Gekandelaberde boom', 'typeSoortnaam': 'Bomen', 'soortnaamKort': 'Salix', 'soortnaamTop': 'Wilg (Salix)'}

这可以通过在查询集上定义model属性或覆盖get_instance / get_individual_instance方法来覆盖。要使用自定义查询集与APIModel一起使用，请在模型类上定义base_query_class属性。

class Tree(APIModel):
    base_query_class = TreeQuerySet
    class Meta:
        fields = ["id", "geometrie", "boomhoogteklasseActueel", "jaarVanAanleg", "soortnaam", "soortnaamKort"]

# >>> Tree.objects.filter(jaarVanAanleg=1986).first()
# <Tree: Tree object (1101570)>

其他数据源

queryish不仅限于REST API——基类queryish.Queryish可以用来围绕任何数据源构建类似QuerySet的API。至少，这需要定义一个返回一个可迭代的记录集的run_query方法，该记录集根据查询集的属性进行过滤、排序和切片。例如，一个从简单的内存对象列表中工作的查询集实现可能看起来像这样

from queryish import Queryish

class CountryQuerySet(Queryish):
    def run_query(self):
        countries = [
            {"code": "nl", "name": "Netherlands"},
            {"code": "de", "name": "Germany"},
            {"code": "fr", "name": "France"},
            {"code": "gb", "name": "United Kingdom"},
            {"code": "us", "name": "United States"},
        ]

        # Filter the list of countries by `self.filters` - a list of (key, value) tuples
        for (key, val) in self.filters:
            countries = [c for c in countries if c[key] == val]

        # Sort the list of countries by `self.ordering` - a tuple of field names
        countries.sort(key=lambda c: [c.get(field, None) for field in self.ordering])

        # Slice the list of countries by `self.offset` and `self.limit`. `offset` is always numeric
        # and defaults to 0 for an unsliced list; `limit` is either numeric or None (denoting no limit).
        return countries[self.offset : self.offset + self.limit if self.limit else None]

子类通常还会重写 run_count 方法，该方法返回考虑任何过滤和切片后的查询集中的记录数。如果没有重写，默认实现将调用 run_query 并计算结果。