Skip to content

Pagination

The Pagination class is a tool for managing paginated rows from the database.

It can be imported from fluid.db:

from fluid.db import Pagination, Search

fluid.db.Pagination

Bases: NamedTuple

order_by_fields instance-attribute 

order_by_fields

limit instance-attribute 

limit

filters instance-attribute 

filters

search instance-attribute 

search

cursor instance-attribute 

cursor

desc class-attribute instance-attribute 

desc = False

order_by_fields_sign property 

order_by_fields_sign

create classmethod 

create(
    *order_by_fields,
    cursor="",
    limit=0,
    filters=None,
    search=None,
    desc=False
)
Source code in fluid/db/pagination.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@classmethod
def create(
    cls,
    *order_by_fields: str,
    cursor: str = "",
    limit: int = 0,
    filters: dict[str, Any] | None = None,
    search: Search | None = None,
    desc: bool = False,
) -> Pagination:
    if limit < 0:
        raise ValidationError("limit must be greater than or equal to 0")
    if cursor:
        if limit > 0:
            raise ValidationError("limit cannot be provided with cursor")
        if filters:
            raise ValidationError("filters cannot be provided with cursor")
        if search and search.search_text:
            raise ValidationError("search text cannot be provided with cursor")
        decoded_cursor = Cursor.decode(cursor, order_by_fields)
        limit = decoded_cursor.limit
        filters = decoded_cursor.filters
        if search:
            search = search._replace(search_text=decoded_cursor.search_text)
    else:
        decoded_cursor = None
        limit = min(
            limit or settings.DEFAULT_PAGINATION_LIMIT,
            settings.DEFAULT_PAGINATION_MAX_LIMIT,
        )
    return cls(
        order_by_fields=order_by_fields,
        cursor=decoded_cursor,
        limit=limit,
        filters=filters or {},
        search=search,
        desc=desc,
    )

execute async 

execute(db, table, *, conn=None)
Source code in fluid/db/pagination.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
async def execute(
    self,
    db: CrudDB,
    table: FromClause,
    *,
    conn: AsyncConnection | None = None,
) -> tuple[Sequence[Row], str]:
    sql_query = self.query(db, table)
    async with db.ensure_connection(conn) as conn:
        result = await conn.execute(sql_query)
    data = result.all()
    cursor = ""
    if len(data) > self.limit:
        cursor = self._encode_cursor(data[-1])
        data = data[:-1]
    return data, cursor

query 

query(db, table)
Source code in fluid/db/pagination.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def query(self, db: CrudDB, table: FromClause) -> Select:
    sql_query = cast(
        Select,
        db.get_query(table, table.select(), params=self.filters),
    )
    if self.search:
        sql_query = db.search_query(
            table,
            sql_query,
            self.search.search_fields,
            self.search.search_text,
        )
    start_clause = self._start_clause(table)
    if start_clause is not None:
        sql_query = sql_query.where(start_clause)
    columns = db.order_by_columns(table, self.order_by_fields_sign)
    return sql_query.order_by(*columns).limit(self.limit + 1)

fluid.db.Search

Bases: NamedTuple

search_fields instance-attribute 

search_fields

search_text instance-attribute 

search_text