I'm having a bit of trouble getting certain results as intended, using FastApi, Sqlalchemy, and Pydantic (v2) schemas.
Here's code I have running, abstracted for the purpose of this question:
database.py
class Base(DeclarativeBase, MappedAsDataclass):
pass
model.py
class User(Base):
__tablename__ = 'users'
id: Mapped[uuid_pkg.UUID] = mapped_column(default_factory=uuid_pkg.uuid4, primary_key=True, unique=True, index=True)
....
bank: Mapped["Bank"] = relationship("Bank", uselist=False, back_populates="user", default=None, lazy='selectin')
class Bank(Base):
__tablename__ = "bank"
id: Mapped[uuid_pkg.UUID] = mapped_column(default_factory=uuid_pkg.uuid4, primary_key=True, unique=True, index=True)
account_number: Mapped[str] = mapped_column(String(10), nullable=False, default=None)
...
user_id: Mapped[uuid_pkg.UUID] = mapped_column(types.Uuid, ForeignKey("users.id"), default=None)
user = relationship("User", back_populates="bank")
Schema.py
class UserRead(BaseModel):
id: uuid_pkg.UUID
bank: Optional[BankBase]
model_config = ConfigDict(from_attributes=True)
class BankBase(BaseModel):
name: Annotated[str, Field(min_length=2, examples=["bank name"])]
account_number: Annotated[str, Field(min_length=2, max_length=10, examples=["account number"])]
account_name: Annotated[str, Field(min_length=2, max_length=50, examples=["account name"])]
model_config = ConfigDict(from_attributes=True)
Expected Result, even as indicated as the appropriate response under Responses for 200 status code:
{
"data": [
{
"id": "3fa85f64-5717-4562-b3fc-xxxxxx",
"username": "gojos",
"email": "gojosatoru@jjk.com",
....
"created_at": "2024-05-17T00:59:58.682Z",
"updated_at": "2024-05-17T00:59:58.682Z",
"bank": {
"name": "bank name",
"account_number": "account number",
"account_name": "account name"
}
}
],
"total_count": 0,
"has_more": true,
"page": 0,
"items_per_page": 0
}
The paginated data is coming through Fastcrud.
NOTE: When I run the /user/me endpoint, that only returns one schema, and not a list: everything works as intended:
{
"id": "25f1a7e3-a47e-4493-84f2-xxxxx",
"username": "gojos",
"email": "gojosaturo@jjk.com",
....
"created_at": "2024-05-16T22:06:17.617345Z",
"updated_at": "2024-05-17T00:19:26.472494Z",
"tier_id": null,
"bank": {
"name": "JJK Bank",
"account_number": "0102939929",
"account_name": "Gojo Satoru",
},
}
Actual Result (ERROR: Internal Server Error
):
raise ResponseValidationError(
fastapi.exceptions.ResponseValidationError: 2 validation errors (2 because I have two users in DB):
{'type': 'model_attributes_type', 'loc': ('response', 'data', 0, 'bank'), 'msg': 'Input should be a valid dictionary or object to extract fields from', 'input': False}
{'type': 'model_attributes_type', 'loc': ('response', 'data', 1, 'bank'), 'msg': 'Input should be a valid dictionary or object to extract fields from', 'input': True}
If I adapt the UserRead
Bank to be bool
such as below, it works by returning true for the user with bank details (created through a ForeignKey) otherwise false:
{
"data": [
{
"id": "3fa85f64-5717-4562-b3fc-xxxxxx",
"username": "gojos",
"email": "gojosatoru@jjk.com",
....
"created_at": "2024-05-17T00:59:58.682Z",
"updated_at": "2024-05-17T00:59:58.682Z",
"bank": true
},
{
"id": "3fa85f64-xxxx-4562-xxxxx-xxxxxx",
"username": "gojosx",
"email": "gojosatorux@jjkx.com",
....
"created_at": "2024-05-17T00:59:58.682Z",
"updated_at": "2024-05-17T00:59:58.682Z",
"bank": false
},
],
"total_count": 2,
"has_more": false,
"page": 0,
"items_per_page": 0
}
The users/all endpoint uses:
users_data = await crud_users.get_multi(
db=db,
offset=compute_offset(page, items_per_page),
limit=items_per_page,
schema_to_select=UserRead,
is_deleted=False,
)
#the below is not working as well
# users_data = await crud_users.get_multi_joined(
# db=db,
# joins_config=[JoinConfig(
# model= Bank,
# join_on= User.id == Bank.user_id,
# schema_to_select= UserRead,
# # join_prefix= 'ba_',
# join_type= 'left',
# )],
# nest_joins=True,
# return_as_model=True,
# schema_to_select=UserRead,
# # join_schema_to_select=BankBase,
# )
response = paginated_response(crud_data=users_data, page=page, items_per_page=items_per_page)
return response
What can I do to get the full bank details instead of a boolean when using the users/all endpoint that returns a list of users in the DB?
users_data = await crud_users.get_multi_joined(
db=db,
joins_config=[JoinConfig(
model=Bank,
join_on=User.id == Bank.user_id,
join_prefix='bank_',
join_type='left',
)],
nest_joins=True,
schema_to_select=UserRead,
offset=compute_offset(page, items_per_page),
limit=items_per_page
)
response: dict[str, Any] = paginated_response(crud_data=users_data, page=page, items_per_page=items_per_page)
A proper joining method resulted to the intended result. I had to go back to the FastCrud documentation (https://igorbenav.github.io/fastcrud/usage/crud/#2-get-joined) and read thoroughly. Surprisingly easy. And before that I had abandoned it to use sql join query. All good now.