uriyyo
Support for paginating Row tuples of object and aggregate fields
So, I have the following query:
query = (
select(
Object,
(func.count(ObjectSaved.id) > 0).label('is_saved'),
)
.options(joinedload(Object.owners))
.outerjoin(
ObjectSaved,
and_(
ObjectSaved.object_id == Object.id,
ObjectSaved.user_username == auth.username
),
)
.group_by(Object) # type: ignore[arg-type]
)
This produces a tuple of (Object, int). To paginate this query, I need to have the schema:
class ObjectMinimalSchema(BaseModel):
id: int
title: str
cover_image: str | None
description: str | None
owners: list['UserMinimalSchema']
class PaginationObjectSchema(BaseModel):
Object: ObjectMinimalSchema
is_saved: bool = False
which works. However, I want is_saved to be part of ObjectMinimalSchema:
class ObjectMinimalSchema(BaseModel):
id: int
title: str
cover_image: str | None
description: str | None
owners: list['UserMinimalSchema']
is_saved: bool = False
and use this schema for pagination.
Hmmm,
You can try to use transformer:
paginate(
query,
transformer=lambda items: [{**obj, "is_saved": is_saved} for obj, is_saved in items],
)
The obj is Object object, meaning, it is not dict. I was able to do a workaround like this which could be maybe adopted?
def merge_agg_to_obj(item: Row):
extra = item._asdict() # main_obj + extra
obj = extra.pop(item._parent._keys[0])
# TODO: Fix this workaround
obj.__dict__.update(extra)
return obj
def unwrap_with_merging_agg_to_obj(items: Sequence[Row]):
return [merge_agg_to_obj(item[0] if len_or_none(item) == 1 else item) for item in items]
For now, to use it, I need to patch the internal function like:
def paginate_with_merge(db: Session, query: Select):
import fastapi_pagination.ext.sqlalchemy as fp_sa
fp_sa.unwrap_scalars = unwrap_with_merging_agg_to_obj
res = fp_sa.paginate(db, query)
fp_sa.unwrap_scalars = unwrap_scalars
return res
Hi @mmzeynalli,
Sorry for long response, I guess you can achieve it without patching unwrap_scalars.
Can you try to use transformer like this?
def transformer(items: list[tuple[Object, int]]) -> list[Object]:
def _transformer(obj: Object, is_saved: bool) -> Object:
obj.is_saved = is_saved
return obj
return [_transformer(obj, is_saved) for obj, is_saved in items]
Hi! Sorry, must have missed the last one. I am still using the method I described. The logic mentioned seems doable, however, that would mean iterating over list of objects twice (first for unwrapping, second for transformation). Wouldn't it be better to apply transformation during unwrapping?
Wouldn't it be better to apply transformation during unwrapping?
It makes sense to me, I will take a look
Hi @mmzeynalli @vahidzhe,
New version 0.12.28 has been released and it introduces unwrap_mode that allows you to control how rows will be unwrapped.
In your case you are interested in no-unwrap mode, here is example:
from typing import List, Tuple
from faker import Faker
from pydantic import BaseModel
from sqlalchemy import Row, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
from fastapi_pagination import Page, Params, set_page, set_params
from fastapi_pagination.ext.sqlalchemy import paginate
faker = Faker()
engine = create_engine("sqlite:///:memory:")
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
name: Mapped[str]
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True, default=None)
Base.metadata.create_all(engine)
session = Session(engine)
class ObjectMinimalSchema(BaseModel):
id: int
name: str
is_big: bool
with session.begin():
session.add_all([User(name=faker.name()) for _ in range(100)])
set_page(Page[ObjectMinimalSchema])
set_params(Params(page=2, size=5))
with session.begin():
def transformer(items: List[Row[Tuple[User, bool]]]) -> List[User]:
for item in items:
extra = item._asdict() # main_obj + extra
obj = extra.pop(item._parent._keys[0])
obj.__dict__.update(extra)
return [item for item, _ in items]
page = paginate(
session,
select(User, (User.id > 20).label("is_big")),
transformer=transformer,
unwrap_mode="no-unwrap",
)
print(page.model_dump_json(indent=2))
{
"items": [
{
"id": 6,
"name": "Joshua Jarvis",
"is_big": false
},
{
"id": 7,
"name": "James Evans",
"is_big": false
},
{
"id": 8,
"name": "David Taylor",
"is_big": false
},
{
"id": 9,
"name": "James Baker",
"is_big": false
},
{
"id": 10,
"name": "Michael Aguirre MD",
"is_big": false
}
],
"total": 100,
"page": 2,
"size": 5,
"pages": 20
}