Gino: Query Filters, Pagination and Sorting

[Psykepro]

• GINO version: 0.8.6
• Python version: 3.7.0
• asyncpg version: 0.20.1
• aiocontextvars version: 0.2.2
• PostgreSQL version: 11

Description

I'm trying to find solution in Gino for filtering, pagination, and sorting. Like this one which is for SQL Alchemy:
sqlalchemy-filters

I know that GINO isn't working with the sqlalchemy.orm part. But I wanna ask if there is some way to adapt it to work with GINO or is there any similar solution for it?

[fantix]

With a quick search, I didn't find anything similar to sqlalchemy-filters. For filtering and sorting, I think you can easily build something to encapsulate the logic you need.

However for pagination, you may want to use server-side cursor:

• https://python-gino.org/docs/en/master/reference/api/gino.engine.html#gino.engine.GinoConnection.iterate
• https://magicstack.github.io/asyncpg/current/api/index.html#cursors

[Psykepro]

Thank you anyway! So I'll start writing my own sorting and filtering. :)

[Pentusha]

Maybe it will be helpful.

class QueryExecutor(GinoExecutor):
    def paginate(self, condition: Any, limit: int, order_by: ColumnElement) -> GinoExecutor:
        return (
            self.query
            .order_by(order_by)
            .limit(limit + 1)
            .gino
            .where_if(condition)
        )

    def paginate_dumb(self, order_by: ColumnElement, limit: int, offset: int) -> GinoExecutor:
        return (
            self.query
            .order_by(order_by)
            .limit(limit + 1)
            .offset(offset)
            .gino
        )

    def where_if(self, condition: Any) -> GinoExecutor:
        if condition is not ...:
            return self.query.where(condition).gino
        else:
            return self


class CustomGino(Gino):
    query_executor = QueryExecutor

db = CustomGino()

How to use it:

def news(limit: int = 5, created_after: Optional[datetime] = None, optional_filter: Optional[str]) -> Awaitable[List[NewsEntry]]:
    return (
        NewsEntry.query
        .where(NewsEntry.deleted_at.is_(None))
        .gino
        .where_if(NewsEntry.some_field == optional_filter if optional_filter is not None else ...)
        .paginate(
            limit=limit,
            order_by=NewsEntry.created_at.desc(),
            condition=created_after > NewsEntry.created_at if created_after else ...,
        )
        .all()
    )

where_if method is for optional filtering.
paginate method is for pagination
paginate_dumb method is cost ineffective pagination with limit, offset.

[Psykepro]

Is there any easy working solution for filtering from a query string similar to this solution for pagination? :)

Maybe it will be helpful.

class QueryExecutor(GinoExecutor):
    def paginate(self, condition: Any, limit: int, order_by: ColumnElement) -> GinoExecutor:
        return (
            self.query
            .order_by(order_by)
            .limit(limit + 1)
            .gino
            .where_if(condition)
        )

    def paginate_dumb(self, order_by: ColumnElement, limit: int, offset: int) -> GinoExecutor:
        return (
            self.query
            .order_by(order_by)
            .limit(limit + 1)
            .offset(offset)
            .gino
        )

    def where_if(self, condition: Any) -> GinoExecutor:
        if condition is not ...:
            return self.query.where(condition).gino
        else:
            return self


class CustomGino(Gino):
    query_executor = QueryExecutor

db = CustomGino()

How to use it:

def news(limit: int = 5, created_after: Optional[datetime] = None, optional_filter: Optional[str]) -> Awaitable[List[NewsEntry]]:
    return (
        NewsEntry.query
        .where(NewsEntry.deleted_at.is_(None))
        .gino
        .where_if(NewsEntry.some_field == optional_filter if optional_filter is not None else ...)
        .paginate(
            limit=limit,
            order_by=NewsEntry.created_at.desc(),
            condition=created_after > NewsEntry.created_at if created_after else ...,
        )
        .all()
    )

where_if method is for optional filtering.
paginate method is for pagination
paginate_dumb method is cost ineffective pagination with limit, offset.

[Pentusha]

Not sure if there can be a universal and good solution. The query must first be validated and deserialized before being sent to the database.
Query string is obviously the string, but your database query works with integers, datetimes and so on.

[Psykepro]

Yes, I know that but I wanted to check if there is already some solution to be used as good practice using Gino before starting to write my own solution. Okay, I'm closing the question. Thank you both and have a nice day! :)

[asfaltboy]

Could the above simple pagination proposition fit a page in the docs perhaps? If the idea resonates positively, I'll be more than happy to write up a small PR 🙏

[fantix]

Yeah, I think it's worth a page if solving at least a few use cases. PR is definitely welcome and appreciated!