logoChat01.ai

@staticmethod async def get_animal_all_locatio...

Erstellt am: 8. Dezember 2025

Beantwortet mit GPT-5.1 Thinking von Chat01

Frage

@staticmethod
async def get_animal_all_locations(search: str | None) -> Select:
"""Get animals all locations."""
disposal_event = (
select(
EventDB.animal_uuid,
EventDB.event_date,
func.row_number().over(
partition_by=EventDB.animal_uuid,
order_by=EventDB.event_date.desc(),
).label('row_number')
)
.where(EventDB.event_type == EventTypeEnum.disposal_event, ~EventDB.is_edit, ~EventDB.is_delete)
.subquery()
)
weight_log = (
select(
WeightLogDB.animal_uuid,
WeightLogDB.live_weight,
WeightLogDB.event_date,
func.row_number().over(
partition_by=WeightLogDB.animal_uuid,
order_by=WeightLogDB.live_weight.desc()
).label('row_number')
)
.subquery()
)
query = (
select(
SheepDB,
disposal_event.c.event_date.label('disposal_date'),
weight_log.c.event_date.label('weight_max_date'),
weight_log.c.live_weight.label('weight_max'),
func.extract('year', func.age(weight_log.c.event_date, SheepDB.birth_day)).label('year_max_weight'),
func.extract('month', func.age(weight_log.c.event_date, SheepDB.birth_day)).label('month_max_weight'),
)
.options(
joinedload(SheepDB.birth_country),
joinedload(SheepDB.birth_farm),
joinedload(SheepDB.sex),
joinedload(SheepDB.birth_type),
joinedload(SheepDB.insemination_method),
)
.where(SheepDB.location_uuid.is_not(None))
.select_from(SheepDB)
.outerjoin(disposal_event, and_(disposal_event.c.animal_uuid == SheepDB.uuid, disposal_event.c.row_number == 1))
.outerjoin(weight_log, and_(weight_log.c.animal_uuid == SheepDB.uuid, weight_log.c.row_number == 1))
)
if search:
query = query.filter(
or_(
SheepDB.number_unsm.ilike(f'%{search}%'),
SheepDB.number_harriot.ilike(f'%{search}%')
)
)
return query

class AncestorForAddEventListSchema(BaseSchema):
"""Ancestor for add event list schema."""
uuid: UUID
birth_country: UUIDNameSchema | None
birth_farm: UUIDNameSchema | None
sex: UUIDNameSchema
number_unsm: str | None
number_inv: str | None
number_electronic_tag: str | None
number_harriot: str | None
number_foreign: str | None
birth_day: date | None
birth_type: UUIDNameSchema | None
insemination_method: UUIDNameSchema | None
disposal_date: date | None
weight_max: float | None
weight_max_date: date | None
year_max_weight: int | None
month_max_weight: int | None

@animal_router.get(
animal_url.animal_list_all_locations,
response_model=PaginatePage[AncestorForAddEventListSchema],
responses=responses(
PaginatePage[AncestorForAddEventListSchema],
statuses=[status.HTTP_400_BAD_REQUEST, status.HTTP_404_NOT_FOUND],
),
description='Animals list all locations',
)
async def animal_all_locations_list(
search: Annotated[str | None, Query(description='Search field by number_unsm, number_harriot')] = None,
_: UserDB = Depends(PermissionManager(ANIMAL_DB, ActionEnum.get_list)),
session: AsyncSession = Depends(get_async_session),
) -> PaginatePage[AncestorForAddEventListSchema]:
result = await AnimalSession(session).get_animal_all_locations(search)
return await paginate(session, result)

Выходит вот такая ошибка.
{
"type": "ValidationError",
"message": "120 validation errors for Page[TypeVar]Customized[AncestorForAddEventListSchema]\nitems.0.uuid\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.0.birth_country\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.0.birth_farm\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.0.sex\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.0.number_unsm\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.0.number_inv\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.0.number_electronic_tag\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.0.number_harriot\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.0.number_foreign\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.0.birth_day\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.0.birth_type\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.0.insemination_method\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.1.uuid\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.1.birth_country\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.1.birth_farm\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.1.sex\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.1.number_unsm\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.1.number_inv\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.1.number_electronic_tag\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.1.number_harriot\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.1.number_foreign\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.1.birth_day\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.1.birth_type\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.1.insemination_method\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.2.uuid\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.2.birth_country\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.2.birth_farm\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.2.sex\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.2.number_unsm\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.2.number_inv\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further information visit https://errors.pydantic.dev/2.8/v/missing\nitems.2.number_electronic_tag\n Field required [type=missing, input_value=(<src.animal.models.Sheep... None, None, None, None), input_type=Row]\n For further informa

"traceback": [
{
"file": "/Users/antongromov/projects/back/venv/lib/python3.12/site-packages/starlette/_exception_handler.py",
"line": 53,
"function": "wrapped_app",
"code": "await app(scope, receive, sender)"
},
{
"file": "/Users/antongromov/projects/back/venv/lib/python3.12/site-packages/starlette/routing.py",
"line": 72,
"function": "app",
"code": "response = await func(request)"
},
{
"file": "/Users/antongromov/projects/back/venv/lib/python3.12/site-packages/fastapi/routing.py",
"line": 278,
"function": "app",
"code": "raw_response = await run_endpoint_function("
},
{
"file": "/Users/antongromov/projects/back/venv/lib/python3.12/site-packages/fastapi/routing.py",
"line": 191,
"function": "run_endpoint_function",
"code": "return await dependant.call(**values)"
},
{
"file": "/Users/antongromov/projects/back/src/animal/routers.py",
"line": 212,
"function": "animal_all_locations_list",
"code": "return await paginate(session, result)"
},
{
"file": "/Users/antongromov/projects/back/venv/lib/python3.12/site-packages/sqlalchemy/util/_concurrency_py3k.py",
"line": 203,
"function": "greenlet_spawn",
"code": "result = context.switch(value)"
},
{
"file": "/Users/antongromov/projects/back/venv/lib/python3.12/site-packages/fastapi_pagination/ext/sqlalchemy.py",
"line": 186,
"function": "exec_pagination",
"code": "return create_page("
},
{
"file": "/Users/antongromov/projects/back/venv/lib/python3.12/site-packages/fastapi_pagination/api.py",
"line": 182,
"function": "create_page",
"code": "return _page_val.get().create(items, **kwargs)"
},
{
"file": "/Users/antongromov/projects/back/venv/lib/python3.12/site-packages/fastapi_pagination/default.py",
"line": 73,
"function": "create",
"code": "return create_pydantic_model("
},
{
"file": "/Users/antongromov/projects/back/venv/lib/python3.12/site-packages/fastapi_pagination/utils.py",
"line": 171,
"function": "create_pydantic_model",
"code": "return model_cls.model_validate(kwargs, from_attributes=True) # type: ignore"
},
{
"file": "/Users/antongromov/projects/back/venv/lib/python3.12/site-packages/pydantic/main.py",
"line": 568,
"function": "model_validate",
"code": "return cls.pydantic_validator.validate_python("
}
]
}

Сделай чтобы она не выходила и все подгружалось при этом.

Gedanke

Clarifying async paginate usage and validation error

I’ve found that the paginate function in ext.sqlalchemy is likely similar to the one in SQLModel, with differences for async usage. The error here is related to model transformation, which can be resolved by using the transformer parameter. I’ll explain the issue briefly in Russian and then show the user the corrected code.

Explaining the issue with Row mapping

In FastAPI pagination, it tries to map each result row to AncestorForAddEventListSchema, but the query returns a Row, not an object or dictionary with fields like uuid. This causes missing fields. We have three options: (1) modify select for ORM, (2) adjust the schema, or (3) use transformer in pagination. I'll implement option (3) to convert Row to the desired schema with minimal changes to the query.

Explaining async transformer for pagination

In the FastAPI route for listing animal locations, I define a transformer function within the handler that processes each result row into the AncestorForAddEventListSchema. Since we’re using an asynchronous session, the pagination function allows for async transformers. However, we don't have to make the transformer async—it can remain sync.

Explaining the use of synchronous transformer in pagination

Since paginate supports both synchronous and asynchronous transformers, I'll use a simple synchronous lambda for readability in the route. The lambda will transform the rows into AncestorForAddEventListSchema, ensuring row.disposal_date works as expected even if row is a tuple or mapping.

Ensuring transformer works correctly with pagination

To improve clarity, I'll use destructuring within the transformer to better handle the Row objects. This way, accessing columns directly (like sheep.uuid, disposal_date, and so on) is simpler. Additionally, I should ensure that PaginatePage[AncestorForAddEventListSchema] is correctly used and that year_max_weight and month_max_weight are cast to integers.

Clarifying function and field handling in code

Pyd

Thought for 2m 0s

Antwort

Ошибка возникает из-за того, что fastapi-pagination пытается засунуть в AncestorForAddEventListSchema строки из БД типа Row/tuple, а не объект/словарь с полями uuid, birth_country и т.д.

Сейчас select(...) возвращает такую строку:

python
(<SheepDB ...>, disposal_date, weight_max_date, weight_max, year_max_weight, month_max_weight)

А Pydantic ждёт что-то вида:

python
{
    "uuid": ...,
    "birth_country": ...,
    ...
    "disposal_date": ...,
    "weight_max": ...,
}

Из-за этого и сыпятся Field required для всех полей.

Лучшее простое решение — использовать transformer у paginate, чтобы превратить строки из БД в объекты AncestorForAddEventListSchema (или словари в таком же формате).

Что нужно поменять

get_animal_all_locations можешь оставить как есть. Меняем роут:

python
@animal_router.get(
    animal_url.animal_list_all_locations,
    response_model=PaginatePage[AncestorForAddEventListSchema],
    responses=responses(
        PaginatePage[AncestorForAddEventListSchema],
        statuses=[status.HTTP_400_BAD_REQUEST, status.HTTP_404_NOT_FOUND],
    ),
    description='Animals list all locations',
)
async def animal_all_locations_list(
        search: Annotated[str | None, Query(description='Search field by number_unsm, number_harriot')] = None,
        _: UserDB = Depends(PermissionManager(ANIMAL_DB, ActionEnum.get_list)),
        session: AsyncSession = Depends(get_async_session),
) -> PaginatePage[AncestorForAddEventListSchema]:
    stmt = await AnimalSession(session).get_animal_all_locations(search)

    # transformer конвертит строки БД в нужную схему
    def transformer(rows):
        items: list[AncestorForAddEventListSchema] = []

        for sheep, disposal_date, weight_max_date, weight_max, year_max_weight, month_max_weight in rows:
            items.append(
                AncestorForAddEventListSchema(
                    uuid=sheep.uuid,
                    birth_country=sheep.birth_country,
                    birth_farm=sheep.birth_farm,
                    sex=sheep.sex,
                    number_unsm=sheep.number_unsm,
                    number_inv=sheep.number_inv,
                    number_electronic_tag=sheep.number_electronic_tag,
                    number_harriot=sheep.number_harriot,
                    number_foreign=sheep.number_foreign,
                    birth_day=sheep.birth_day,
                    birth_type=sheep.birth_type,
                    insemination_method=sheep.insemination_method,
                    disposal_date=disposal_date,
                    weight_max=weight_max,
                    weight_max_date=weight_max_date,
                    year_max_weight=int(year_max_weight) if year_max_weight is not None else None,
                    month_max_weight=int(month_max_weight) if month_max_weight is not None else None,
                )
            )

        return items

    # важно: передаём transformer в paginate
    return await paginate(
        session,
        stmt,
        transformer=transformer,
    )

⚠️ Обрати внимание:

  • В цикле я делаю распаковку for sheep, disposal_date, ... in rows: — это соответствует твоему select(SheepDB, disposal_event.c.event_date, ...).
  • func.extract обычно отдаёт Decimal/float, поэтому я привожу к int для year_max_weight и month_max_weight.
  • birth_country, birth_farm, sex, birth_type, insemination_method — это ORM-объекты; чтобы Pydantic нормально их сериализовал в UUIDNameSchema, в BaseSchema (или в самой AncestorForAddEventListSchema) должен быть включён from_attributes=True (в Pydantic v2 — через model_config = ConfigDict(from_attributes=True)).

После такой обёртки:

  • paginate сначала достанет из БД Row/tuple,
  • transformer превратит их в список AncestorForAddEventListSchema,
  • Pydantic больше не будет ругаться на Field required,
  • и все данные (disposal_date, максимум веса и т.д.) будут приходить в ответе.

Diesen Q&A teilen