How to paginate raw SQLAlchemy query

[paw-bor]

How to paginate raw SQLAlchemy query, e.g.:

from fastapi_pagination.ext.sqlalchemy import paginate
(...)
query = 'SELECT * from User'
(...)
return paginate(session, execute(text(query)))

[ppanchal-mobiz]

I have the same query, how do I paginate the result of a stored procedure?

[uriyyo]

Unfortunately, there is no way to do it now, I will think about how we can support such cases

[uriyyo]

@paw-bor @ppanchal-mobiz I'm planing to add experimental support for raw-queries, here is implementation PR- #1083

[ppanchal-mobiz]

@uriyyo, unfortunately, this doesn't seem to work in my scenario with the stored procedure. I get the following error:

sqlalchemy.exc.ProgrammingError: (pyodbc.ProgrammingError) ('42000', "[42000] [Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Incorrect syntax near the keyword 'EXEC'. (156) (SQLExecDirectW); [42000] [Microsoft][ODBC Driver 18 for SQL Server][SQL Server]Incorrect syntax near ')'. (102)")
[SQL: SELECT count(*) FROM (EXEC usp_fo_dashboard_table 2024, 2) AS __count_query__]

[uriyyo]

@ppanchal-mobiz I will take a look

[pshrest2]

Hey @uriyyo, is there a way to paginate parameterized raw SQLAlchemy query? For example:

statement = text("select * from projects where user_id = :user_id")
page = paginate(conn, statement, transformer=lambda rows: [r._asdict() for r in rows]) # how do I pass :user_id here?

The only way I can make it work right now is to build the query first and then paginate it. However, doing so brings the risk of SQL injection. The recommended way for SQLAlchemy to execute these parameterized query is by passing the parameters while executing the query as follows:

result = session.execute(statement, { 'user_id': 1 }).all()

[uriyyo]

@pshrest2 I will take a look

[pshrest2]

@uriyyo, I was able to run this by creating a custom paginate method as shown below

from typing import Any, Optional

from fastapi_pagination.api import apply_items_transformer, create_page
from fastapi_pagination.bases import AbstractPage, AbstractParams
from fastapi_pagination.ext.sqlalchemy import (
    Selectable,
    _maybe_unique,
    create_count_query,
    create_paginate_query,
)
from fastapi_pagination.ext.utils import unwrap_scalars
from fastapi_pagination.types import AdditionalData, ItemsTransformer
from fastapi_pagination.utils import verify_params
from sqlalchemy.engine.interfaces import _CoreAnyExecuteParams
from sqlalchemy.orm import Session


def paginate(
    conn: Session,
    query: Selectable,
    query_params: Optional[_CoreAnyExecuteParams] = None,
    params: Optional[AbstractParams] = None,
    *,
    subquery_count: bool = True,
    count_query: Optional[Selectable] = None,
    transformer: Optional[ItemsTransformer] = None,
    additional_data: Optional[AdditionalData] = None,
    unique: bool = True,
) -> AbstractPage[Any]:
    params, raw_params = verify_params(params, "limit-offset")

    total = None
    if raw_params.include_total:
        if count_query is None:
            count_query = create_count_query(query, use_subquery=subquery_count)
        total = conn.scalar(count_query, query_params)

    query = create_paginate_query(query, params)
    items = conn.execute(query, query_params).all()
    items = _maybe_unique(conn.execute(query, query_params), unique)
    items = unwrap_scalars(items)
    items = apply_items_transformer(items, transformer)

    return create_page(
        items,
        total=total,
        params=params,
        **(additional_data or {}),
    )

Perhaps something similar can be added in a future release? Thanks.