We read every piece of feedback, and take your input very seriously.
To see all available qualifiers, see our documentation.
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
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:
is_saved
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.
The text was updated successfully, but these errors were encountered:
Hmmm,
You can try to use transformer:
transformer
paginate( query, transformer=lambda items: [{**obj, "is_saved": is_saved} for obj, is_saved in items], )
Sorry, something went wrong.
The obj is Object object, meaning, it is not dict. I was able to do a workaround like this which could be maybe adopted?
obj
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
No branches or pull requests
So, I have the following query:
This produces a tuple of (Object, int). To paginate this query, I need to have the schema:
which works. However, I want
is_saved
to be part ofObjectMinimalSchema
:and use this schema for pagination.
The text was updated successfully, but these errors were encountered: