LITE-29680 Supports of fastapi pagination

connect_extension_utils/api/pagination.py

@@ -0,0 +1,32 @@
+from fastapi import Query
+from fastapi_pagination import LimitOffsetPage, LimitOffsetParams, set_page
+from fastapi_pagination.api import resolve_params
+from fastapi_pagination.ext.sqlalchemy import paginate
+
+
+set_page(LimitOffsetPage)
+
+
+class PaginationParams(LimitOffsetParams):
+    """Here we can redefine default size value"""
+
+    limit: int = Query(1000, ge=1, le=1000, description="Page size")
+
+
+def apply_pagination(query, db, params, response):
+    """Apply pagination for the query
+    * Paginate query according to applied parameters (size and page)
+    * Add pagination headers to the response
+    Don't filter or remove elements from query after the pagination
+    """
+    paginated = paginate(db, query, params)
+    resolve_params(params)
+    init = paginated.offset
+    # If we are selecting a page that it's out of range, we return the same start and end for on
+    # header range, copying the same behavior as connect.
+    end = init
+    if paginated.items:
+        end += len(paginated.items) - 1
+
+    response.headers['Content-Range'] = f'items {init}-{end}/{paginated.total}'
+    return paginated.items

pyproject.toml

@@ -15,6 +15,7 @@ python = ">=3.8,<4"
 sqlalchemy = "^1.3.12"
 pyjwt = "^2.8.0"
 connect-eaas-core = "^30.3"
+fastapi-pagination = "<=0.12.17"
 
 [tool.poetry.dev-dependencies]
 pytest = ">=6.1.2,<8"

tests/api/test_pagination.py

@@ -0,0 +1,32 @@
+import pytest
+from fastapi import Response
+
+from connect_extension_utils.api.pagination import PaginationParams, apply_pagination
+
+
+@pytest.mark.parametrize(
+    ('limit', 'offset', 'expected_length', 'expected_header'),
+    (
+        (10, 0, 10, 'items 0-9/20'),
+        (10, 20, 0, 'items 20-20/20'),
+        (9, 18, 2, 'items 18-19/20'),
+    ),
+)
+def test_apply_pagination(
+    limit,
+    offset,
+    expected_length,
+    expected_header,
+    dbsession,
+    my_model_factory,
+):
+    params = PaginationParams(limit=limit, offset=offset)
+    for _ in range(20):
+        my_model_factory()
+
+    items = dbsession.query(my_model_factory._meta.model)
+    response = Response()
+    paginated_items = apply_pagination(items, dbsession, params, response)
+
+    assert len(paginated_items) == expected_length
+    assert response.headers['Content-Range'] == expected_header