Support for paginating Row tuples of object and aggregate fields

[mmzeynalli]

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.

[uriyyo]

Hmmm,

You can try to use transformer:

paginate(
    query,
    transformer=lambda items: [{**obj, "is_saved": is_saved} for obj, is_saved in items],
)

[mmzeynalli]

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

[uriyyo]

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]