# FastAPI Querysets
---
**Documentation**: <a href="https://fastapi-querysets.readthedocs.io" target="_blank">https://fastapi-querysets.readthedocs.io</a>
**Source Code**: <a href="https://github.com/Nikakto/fastapi-querysets" target="_blank">https://github.com/Nikakto/fastapi-querysets</a>
---
## Why to use?
While you are developing FastAPI applications you are using databases with ORM. Most of the endpoints are view of database tables and require restrict queryset by filtering, pagination, ordering. This project is generic and reusable way to create restricted querysets for your endpoints.
## Supported ORM
- <a href="https://github.com/tortoise/tortoise-orm" target="_blank">Tortoise ORM</a>
## Requirements
- python >=3.8,<4.0
- fastAPI >= 0.78
- tortoise-orm >= 0.18.1
## Installation
```
pip install fastapi-querysets
```
# Quick tutorial
---
---
## Tortoise model
Let’s start with our model
```python
# models/tortoise.py
import datetime
from typing import Optional
from tortoise import Model
from tortoise import fields
class Task(Model):
id: int = fields.IntField(pk=True)
approved: Optional[bool] = fields.BooleanField(default=False, null=True)
code: str = fields.CharField(max_length=6)
created_at: datetime.datetime = fields.DatetimeField(default=datetime.datetime.now)
```
---
## Pydantic model
Create database representation model
```python
# models/pydantic.py
from tortoise.contrib.pydantic import pydantic_model_creator
from myproject.models.tortoise import Task
TaskModelOut = pydantic_model_creator(
Task,
name="TaskModelOut",
include=(
"id",
"approved",
"code",
"created_at"
),
)
```
---
## RouterQuerySet
### Filters
We have a number of fields we want to let our users filter based on them. We create a RouterQuerySetFilter for this. Filter class is argument for `FastAPI.Depends` at endpoint.
You require to define ORM filter keyword, type of data and setup source of data (`Query`, `Path`, `Body`, etc).
```python
# querysets_filters.py
import dataclasses
import datetime
from typing import Optional
from fastapi import Query
@dataclasses.dataclass
class RouterQuerySetFilter:
id__in: Optional[list[int]] = Query(None, alias='id[]')
approved: Optional[bool] = Query(None)
approved__isnull: Optional[bool] = Query(None)
code: Optional[str] = Query(None)
created_at__lte: Optional[datetime.datetime] = Query(None)
created_at__gte: Optional[datetime.datetime] = Query(None)
```
### Model Queryset
Configure `RouterQueryset` properties
```python
# querysets.py
from fastapi_querysets.mixins.filters import FilterMixin
from fastapi_querysets.mixins.filters import FilterNegationMixin
from fastapi_querysets.mixins.ordering import OrderingMixin
from fastapi_querysets.mixins.pagination import PaginationMixin
from fastapi_querysets.mixins.pagination import RouterPagination
from fastapi_querysets.queryset import RouterQuerySet
from myproject.models.tortoise import Task
from myproject.querysets_filters import RouterQuerySetFilter
class TasksRouterQuerySet(FilterMixin, FilterNegationMixin, OrderingMixin, PaginationMixin, RouterQuerySet):
filter_class = RouterQuerySetFilter
ordering_default = "id"
ordering_fields = (
"id",
"approved",
"code",
"created_at",
)
pagination_class = RouterPagination
model = Task
```
---
## Application
Create application, register list, list with pagination and retrieve endpoints.
```python
# app.py
from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise
from tortoise.queryset import QuerySet
from myproject.models.pydantic import TaskModelOut
from myproject.models.tortoise import Task
from myproject.querysets import TasksRouterQuerySet
app = FastAPI()
register_tortoise(
app,
db_url="sqlite://:memory:",
modules={"models": ["myproject.models.tortoise"]},
generate_schemas=True,
add_exception_handlers=True,
)
@app.get("tasks/", response_model=list[TaskModelOut])
async def tasks_list_paginated(queryset: QuerySet[Task] = TasksRouterQuerySet()) -> list[TaskModelOut]:
return await TaskModelOut.from_queryset(queryset)
@app.get("tasks/paginated", response_model=list[TaskModelOut])
async def tasks_list_paginated(queryset: QuerySet[Task] = TasksRouterQuerySet().paginated) -> list[TaskModelOut]:
return await TaskModelOut.from_queryset(queryset)
@app.get("tasks/{instance_id}", response_model=list[TaskModelOut])
async def tasks_retrieve(task: QuerySet[Task] = TasksRouterQuerySet().instance) -> list[TaskModelOut]:
return TaskModelOut.from_orm(task)
```
---
## Requests
### List
On request effective queryset will be filtered and ordered by query params.
For example, user has requested endpoint with some query params
```json
{
"created_at__lte": "2023-01-01T00:00:00",
"approved": false,
"ordering[]": "created_at",
}
```
Request URL looks like
```http://localhost:8000/tasks/?created_at__lte=2023-01-01T00:00:00&approved=false&ordering[]=created_at```
Effective queryset at the endpoint method will be
```python
(
Task
.filter(created_at__lte=datetime.datetime(2023, 1, 1, 0, 0, 0), approved=False)
.order_by("created_at")
)
```
### List paginated
Like not paginated endpoint at this queryset will be filtered, ordering and additional paginated.
For example, user has requested endpoint with some query params
```json
{
"page": 2,
"per_page": 10,
"created_at__lte": "2023-01-01T00:00:00",
"approved": false,
"ordering[]": "created_at",
}
```
Request URL looks like
```http://localhost:8000/tasks/?page=2&per_page=10&created_at__lte=2023-01-01T00:00:00&approved=false&ordering[]=created_at```
Effective queryset at endpoint method will be
```python
(
Task
.filter(created_at__lte=datetime.datetime(2023, 1, 1, 0, 0, 0), approved=False)
.order_by("created_at")
.offset(10)
.limit(10)
)
```
As well as to `Response` will be added pagination information. Pagination information always matches effective queryset
```json
{
"x-page": "2",
"x-pages": "4",
"x-per-page": "10",
"x-total": "32"
}
```
### Retrieve
Request URL looks like
```http://localhost:8000/tasks/10/```
Endpoint method will get `Task` with `id == 10` as argument `task`.
If `Task` with `id == 10` does not exist then endpoint return `Response(404)`
Raw data
{
"_id": null,
"home_page": "https://github.com/Nikakto/fastapi-querysets",
"name": "fastapi-querysets",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8,<4.0",
"maintainer_email": "",
"keywords": "fastapi,filters,filtering,pagination,order,ordering,queryset,Tortoise,orm,database,db",
"author": "Nikakto",
"author_email": "black-z@bk.ru",
"download_url": "https://files.pythonhosted.org/packages/32/25/30c5e1dc330c838141a428dffddd38653c80ff4c2f8e04d00167ea5449ba/fastapi_querysets-0.1.3.tar.gz",
"platform": null,
"description": "# FastAPI Querysets\n\n\n\n---\n**Documentation**: <a href=\"https://fastapi-querysets.readthedocs.io\" target=\"_blank\">https://fastapi-querysets.readthedocs.io</a>\n\n**Source Code**: <a href=\"https://github.com/Nikakto/fastapi-querysets\" target=\"_blank\">https://github.com/Nikakto/fastapi-querysets</a>\n\n---\n\n## Why to use?\n\nWhile you are developing FastAPI applications you are using databases with ORM. Most of the endpoints are view of database tables and require restrict queryset by filtering, pagination, ordering. This project is generic and reusable way to create restricted querysets for your endpoints.\n\n## Supported ORM\n- <a href=\"https://github.com/tortoise/tortoise-orm\" target=\"_blank\">Tortoise ORM</a>\n\n## Requirements\n- python >=3.8,<4.0\n- fastAPI >= 0.78\n- tortoise-orm >= 0.18.1\n\n## Installation\n\n```\npip install fastapi-querysets\n```\n\n# Quick tutorial\n\n---\n\n---\n\n## Tortoise model\n\nLet\u2019s start with our model\n\n```python\n# models/tortoise.py\n\nimport datetime\nfrom typing import Optional\n\nfrom tortoise import Model\nfrom tortoise import fields\n\n\nclass Task(Model):\n id: int = fields.IntField(pk=True)\n approved: Optional[bool] = fields.BooleanField(default=False, null=True)\n code: str = fields.CharField(max_length=6)\n created_at: datetime.datetime = fields.DatetimeField(default=datetime.datetime.now)\n```\n---\n\n## Pydantic model\n\nCreate database representation model\n\n```python\n# models/pydantic.py\n\nfrom tortoise.contrib.pydantic import pydantic_model_creator\n\nfrom myproject.models.tortoise import Task\n\n\nTaskModelOut = pydantic_model_creator(\n Task,\n name=\"TaskModelOut\",\n include=(\n \"id\",\n \"approved\",\n \"code\",\n \"created_at\"\n ),\n)\n```\n\n---\n\n## RouterQuerySet\n\n### Filters\n\nWe have a number of fields we want to let our users filter based on them. We create a RouterQuerySetFilter for this. Filter class is argument for `FastAPI.Depends` at endpoint. \n\nYou require to define ORM filter keyword, type of data and setup source of data (`Query`, `Path`, `Body`, etc).\n\n```python\n# querysets_filters.py\n\nimport dataclasses\nimport datetime\nfrom typing import Optional\n\nfrom fastapi import Query\n\n\n@dataclasses.dataclass\nclass RouterQuerySetFilter:\n id__in: Optional[list[int]] = Query(None, alias='id[]')\n approved: Optional[bool] = Query(None)\n approved__isnull: Optional[bool] = Query(None)\n code: Optional[str] = Query(None)\n created_at__lte: Optional[datetime.datetime] = Query(None)\n created_at__gte: Optional[datetime.datetime] = Query(None)\n```\n\n### Model Queryset\n\nConfigure `RouterQueryset` properties\n\n```python\n# querysets.py\n\nfrom fastapi_querysets.mixins.filters import FilterMixin\nfrom fastapi_querysets.mixins.filters import FilterNegationMixin\nfrom fastapi_querysets.mixins.ordering import OrderingMixin\nfrom fastapi_querysets.mixins.pagination import PaginationMixin\nfrom fastapi_querysets.mixins.pagination import RouterPagination\nfrom fastapi_querysets.queryset import RouterQuerySet\n\nfrom myproject.models.tortoise import Task\nfrom myproject.querysets_filters import RouterQuerySetFilter\n\n\nclass TasksRouterQuerySet(FilterMixin, FilterNegationMixin, OrderingMixin, PaginationMixin, RouterQuerySet):\n filter_class = RouterQuerySetFilter\n ordering_default = \"id\"\n ordering_fields = (\n \"id\",\n \"approved\",\n \"code\",\n \"created_at\",\n )\n pagination_class = RouterPagination\n model = Task\n```\n\n---\n\n## Application\n\nCreate application, register list, list with pagination and retrieve endpoints.\n```python\n# app.py\n\nfrom fastapi import FastAPI\nfrom tortoise.contrib.fastapi import register_tortoise\nfrom tortoise.queryset import QuerySet\n\nfrom myproject.models.pydantic import TaskModelOut\nfrom myproject.models.tortoise import Task\nfrom myproject.querysets import TasksRouterQuerySet\n\n\napp = FastAPI()\n\n\nregister_tortoise(\n app,\n db_url=\"sqlite://:memory:\",\n modules={\"models\": [\"myproject.models.tortoise\"]},\n generate_schemas=True,\n add_exception_handlers=True,\n)\n\n\n@app.get(\"tasks/\", response_model=list[TaskModelOut])\nasync def tasks_list_paginated(queryset: QuerySet[Task] = TasksRouterQuerySet()) -> list[TaskModelOut]:\n return await TaskModelOut.from_queryset(queryset)\n\n\n@app.get(\"tasks/paginated\", response_model=list[TaskModelOut])\nasync def tasks_list_paginated(queryset: QuerySet[Task] = TasksRouterQuerySet().paginated) -> list[TaskModelOut]:\n return await TaskModelOut.from_queryset(queryset)\n\n\n@app.get(\"tasks/{instance_id}\", response_model=list[TaskModelOut])\nasync def tasks_retrieve(task: QuerySet[Task] = TasksRouterQuerySet().instance) -> list[TaskModelOut]:\n return TaskModelOut.from_orm(task)\n```\n\n---\n\n## Requests\n\n### List\n\nOn request effective queryset will be filtered and ordered by query params.\n\nFor example, user has requested endpoint with some query params\n```json\n{\n \"created_at__lte\": \"2023-01-01T00:00:00\",\n \"approved\": false,\n \"ordering[]\": \"created_at\",\n}\n```\n\nRequest URL looks like \n```http://localhost:8000/tasks/?created_at__lte=2023-01-01T00:00:00&approved=false&ordering[]=created_at```\n\nEffective queryset at the endpoint method will be\n```python\n(\n Task\n .filter(created_at__lte=datetime.datetime(2023, 1, 1, 0, 0, 0), approved=False)\n .order_by(\"created_at\")\n)\n```\n\n### List paginated\n\nLike not paginated endpoint at this queryset will be filtered, ordering and additional paginated.\n\nFor example, user has requested endpoint with some query params\n```json\n{\n \"page\": 2,\n \"per_page\": 10,\n \"created_at__lte\": \"2023-01-01T00:00:00\",\n \"approved\": false,\n \"ordering[]\": \"created_at\",\n}\n```\n\nRequest URL looks like \n```http://localhost:8000/tasks/?page=2&per_page=10&created_at__lte=2023-01-01T00:00:00&approved=false&ordering[]=created_at```\n\n\nEffective queryset at endpoint method will be\n```python\n(\n Task\n .filter(created_at__lte=datetime.datetime(2023, 1, 1, 0, 0, 0), approved=False)\n .order_by(\"created_at\")\n .offset(10)\n .limit(10)\n)\n```\n\nAs well as to `Response` will be added pagination information. Pagination information always matches effective queryset\n```json\n{\n \"x-page\": \"2\",\n \"x-pages\": \"4\",\n \"x-per-page\": \"10\",\n \"x-total\": \"32\"\n}\n```\n\n### Retrieve\n\nRequest URL looks like \n```http://localhost:8000/tasks/10/```\n\nEndpoint method will get `Task` with `id == 10` as argument `task`. \nIf `Task` with `id == 10` does not exist then endpoint return `Response(404)` \n",
"bugtrack_url": null,
"license": "MIT",
"summary": "",
"version": "0.1.3",
"project_urls": {
"Documentation": "https://fastapi-querysets.readthedocs.io",
"Homepage": "https://github.com/Nikakto/fastapi-querysets",
"Repository": "https://github.com/Nikakto/fastapi-querysets"
},
"split_keywords": [
"fastapi",
"filters",
"filtering",
"pagination",
"order",
"ordering",
"queryset",
"tortoise",
"orm",
"database",
"db"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "d4a65b00ee7529becc0653356350b663b93e4410c2741a14f947e96cbf95b904",
"md5": "7bee8b4bb5703b32c1c49caf64b67106",
"sha256": "4a11cf64431e0f8841c742ac5bd4157fba302e66d923ed69790f16c411cefc52"
},
"downloads": -1,
"filename": "fastapi_querysets-0.1.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "7bee8b4bb5703b32c1c49caf64b67106",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8,<4.0",
"size": 8262,
"upload_time": "2023-05-10T15:44:17",
"upload_time_iso_8601": "2023-05-10T15:44:17.639467Z",
"url": "https://files.pythonhosted.org/packages/d4/a6/5b00ee7529becc0653356350b663b93e4410c2741a14f947e96cbf95b904/fastapi_querysets-0.1.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "322530c5e1dc330c838141a428dffddd38653c80ff4c2f8e04d00167ea5449ba",
"md5": "644ab97d2d29a61f3c0a7c79150cb4f8",
"sha256": "0a7e4011d129a98332f2c72cb45c116a546a1862141ccb45e64af432416604a0"
},
"downloads": -1,
"filename": "fastapi_querysets-0.1.3.tar.gz",
"has_sig": false,
"md5_digest": "644ab97d2d29a61f3c0a7c79150cb4f8",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8,<4.0",
"size": 6501,
"upload_time": "2023-05-10T15:44:19",
"upload_time_iso_8601": "2023-05-10T15:44:19.376132Z",
"url": "https://files.pythonhosted.org/packages/32/25/30c5e1dc330c838141a428dffddd38653c80ff4c2f8e04d00167ea5449ba/fastapi_querysets-0.1.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-05-10 15:44:19",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Nikakto",
"github_project": "fastapi-querysets",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "fastapi-querysets"
}
JSON
