I want to filter books based on some criteria. Since the filtering logic is roughly equal to "compare value with column_in_db", I decided to create different types of filters for filtering values. I am then using these filters to create my Filter model. I will then use FastAPI to let users of my api populate the filter object and get filtered results.
# filtering.py
from 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
The setup above works perfectly and I can create my filter object like:
>>> my_filter = Filter(author_id={"value": 1}, owner_name={"value": "John"}, price={"min": 10})
>>> my_filter.author_id.column
<sqlalchemy.orm.attributes.InstrumentedAttribute object at 0x105933420>
Now I can create a condition for .where statement in sqlalchemy query:
>>> my_filter.author_id.column == my_filter.author_id.value
<sqlalchemy.sql.elements.BinaryExpression object at 0x102bfc860>
The problem happens when I try to use the Filter in my FastAPI app. FastAPI tries to generate openapi.json for the /docs and it fails to serialize the column arguments in Field's.
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: <class 'sqlalchemy.orm.attributes.InstrumentedAttribute'>
I don't need it to be serialized. I am using the column internally.
So I wonder if there is way to ignore serialization for certain arguments in a field. If what I am trying to do doesn't make much sense or it is not possible, is there another way to do it?
Run the below code with python app.py and go to http://127.0.0.1:8000/docs and you will see the error. To keep things simple, I've replaced references to my sqlalchemy models to int which is also non-serializable. The error explanation looks slightly different but it's caused by the same thing.
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)
This is mentioned in pydantic docs. You can attach any metadata to your field's with the following code:
from typing import Annotated, Any
from fastapi import FastAPI
from pydantic import BaseModel, GetCoreSchemaHandler
from pydantic_core import CoreSchema
class Metadata(BaseModel):
column: Any
@classmethod
def __get_pydantic_core_schema__(cls, source_type: type[BaseModel], handler: GetCoreSchemaHandler) -> CoreSchema:
if cls is not source_type:
return handler(source_type)
return super().__get_pydantic_core_schema__(source_type, handler)
class EqualityFilter(BaseModel):
value: int
class Filter(BaseModel):
author_id: Annotated[EqualityFilter, Metadata(column=str)]
owner_id: Annotated[EqualityFilter, Metadata(column=int)]
app = FastAPI()
@app.post("/books")
async def get_books(filter: Filter):
for field, val in filter.model_fields.items():
print(f"{field=}, {getattr(filter, field)}, metadata={getattr(val, "metadata")}")
result = []
return result
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="127.0.0.1", port=8000)
If you go to docs, you will not see any error. And when you perform a request, all the required information will be available:
field='author_id', value=1, metadata=[Metadata(column=<class 'str'>)]
field='owner_id', value=0, metadata=[Metadata(column=<class 'int'>)]