Я хочу фильтровать книги по некоторым критериям. Поскольку логика фильтрации примерно равна «сравнить значение с column_in_db», я решил создать разные типы фильтры для фильтрации значений. Затем я использую эти фильтры для создания модели Filter. Затем я буду использовать FastAPI, чтобы позволить пользователям моего API заполнять объект фильтра и получать отфильтрованные результаты.
Что работает?
# filtering.pyfrom pydantic import BaseModel, Field
from my_sqlalchemy_models import Book, User
class EqualityFilter(BaseModel):
value: int
_column: Any | None = PrivateAttr(default=None)
def __init__(self, **data):
super().__init__(**data)
self._column = data.get("_column")
@property
def column(self):
return self._column
class MinMaxFilter:
...
class ContainsFilter:
...
class Filter(BaseModel):
author_id: EqualityFilter | None = Field(default=None, column=Book.author_id)
owner_id: EqualityFilter | None = Field(default=None, column=User.id)
owner_name: ContainsFilter | None = Field(default=None, column=User.name)
price: MinMaxFilter | None = Field(default=None, column=Book.price)
@model_validator(mode="before")
@classmethod
def add_columns(cls, data: Any):
for field_name, value in cls.model_fields.items():
schema_extra = value.json_schema_extra
if field_name in data and schema_extra and schema_extra.get("column") is not None:
data[field_name]["_column"] = value.json_schema_extra["column"]
return data
Приведенная выше настройка работает отлично, и я могу создать свой объект фильтра следующим образом:
>>> my_filter = Filter(author_id={"value": 1}, owner_name={"value": "John"}, price={"min": 10})
>>> my_filter.author_id.column
Теперь я могу создать условие для оператора .where в запросе sqlalchemy:
>>> my_filter.author_id.column == my_filter.author_id.value
Что не работает?
Проблема возникает, когда я пытаюсь использовать фильтр в своем приложении FastAPI. FastAPI пытается сгенерировать openapi.json для /docs, но ему не удается сериализовать аргументы столбца в полях.File "***/.venv/lib/python3.12/site-packages/pydantic/_internal/_generate_schema.py", line 2250, in json_schema_update_func
add_json_schema_extra(json_schema, json_schema_extra)
File "***/.venv/lib/python3.12/site-packages/pydantic/_internal/_generate_schema.py", line 2260, in add_json_schema_extra
json_schema.update(to_jsonable_python(json_schema_extra))
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
pydantic_core._pydantic_core.PydanticSerializationError: Unable to serialize unknown type:
Мне не нужна сериализация. Я использую столбец для внутренних целей.
Поэтому мне интересно, есть ли способ игнорировать сериализацию для определенных аргументов в поле. Если то, что я пытаюсь сделать, не имеет особого смысла или невозможно, есть ли другой способ сделать это?
Полный минимальный воспроизводимый пример
Запустите приведенный ниже код с помощью python app.py и перейдите по адресу http://127.0.0.1:8000/docs, и вы увидите ошибку. Для простоты я заменил ссылки на свои модели sqlalchemy на int, который также не подлежит сериализации. Объяснение ошибки выглядит немного иначе, но причина та же.from typing import Any
from fastapi import FastAPI
from pydantic import BaseModel, Field, PrivateAttr, model_validator
class EqualityFilter(BaseModel):
value: int
_column: Any | None = PrivateAttr(default=None)
def __init__(self, **data):
super().__init__(**data)
self._column = data.get("_column")
@property
def column(self):
return self._column
class Filter(BaseModel):
author_id: EqualityFilter | None = Field(default=None, column=int)
owner_id: EqualityFilter | None = Field(default=None, column=int)
@model_validator(mode="before")
@classmethod
def add_columns(cls, data: Any):
for field_name, value in cls.model_fields.items():
schema_extra = value.json_schema_extra
if field_name in data and schema_extra and schema_extra.get("column") is not None:
data[field_name]["_column"] = value.json_schema_extra["column"]
return data
app = FastAPI()
@app.post("/books")
async def get_books(filter: Filter):
# do the filtering
result = []
return result
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)
Подробнее здесь: https://stackoverflow.com/questions/787 ... n-pydantic