feat: add with_pagination dependency (#13)

Author: hadrien

README.md

@@ -22,6 +22,8 @@ app = FastAPI()
 fastapi_sqla.setup(app)
 ```
 
+## SQLAlchemy
+
 ### Adding a new entity class:
 
 ```python
@@ -46,7 +48,88 @@ def example(session: Session = Depends(with_session)):
     return session.execute("SELECT now()").scalar()
 ```
 
+### Pagination
+
+```python
+from fastapi import APIRouter, Depends
+from fastapi_sqla import Base, Paginated, Session, with_pagination, with_session
+from pydantic import BaseModel
+
+router = APIRouter()
+
+
+class UserEntity(Base):
+    __tablename__ = "user"
+
+
+class User(BaseModel):
+    id: int
+    name: str
+
+
+@router.get("/users", response_model=Paginated[User])
+def all_users(
+    session: Session = Depends(with_session),
+    paginated_result=Depends(with_pagination),
+):
+    query = session.query(UserEntity)
+    return paginated_result(query)
+```
+
+By default:
+* It returns pages of 10 items, up to 100 items;
+* Total number of items in the collection is queried using
+  [`Query.count`](https://docs.sqlalchemy.org/en/13/orm/query.html#sqlalchemy.orm.query.Query.count)
+
+### Custom pagination
+
+You can customize:
+- Minimum and maximunm number of items per pages;
+- How the total number of items in the collection is queried;
+
+To customize pagination, create a dependency using `fastapi_sqla.Pagination`
+
+```python
+from fastapi import APIRouter, Depends
+from fastapi_sqla import Base, Paginated, Pagination, Session, with_session
+from pydantic import BaseModel
+from sqlalchemy import func
+from sqlalchemy.orm import Query
+
+router = APIRouter()
+
+
+class UserEntity(Base):
+    __tablename__ = "user"
+
+
+class User(BaseModel):
+    id: int
+    name: str
+
+
+def query_count(session: Session, query: Query):
+    return query.statement.with_only_columns([func.count()]).scalar()
+
+
+with_custom_pagination = Pagination(
+    min_page_size=5,
+    max_page_size=500,
+    query_count=query_count,
+)
+
+
+@router.get("/users", response_model=Paginated[User])
+def all_users(
+    session: Session = Depends(with_session),
+    paginated_result=Depends(with_custom_pagination),
+):
+    query = session.query(UserEntity)
+    return paginated_result(query)
+```
+
 ## Pytest fixtures
+
 This library provides a set of utility fixtures, through its PyTest plugin, which is
 automatically installed with the library.
 
@@ -75,7 +158,8 @@ def sqla_modules():
 
 The DB url to use.
 
-When `CI` key is set in environment variables, it defaults to using `postgres` as the host name:
+When `CI` key is set in environment variables, it defaults to using `postgres` as the
+host name:
 
 ```
 postgresql://postgres@posgres/postgres

fastapi_sqla/__init__.py

@@ -1,12 +1,17 @@
 import asyncio
+import math
 import os
 from contextlib import contextmanager
+from typing import Callable, Generic, List, TypeVar
 
 import structlog
-from fastapi import FastAPI, Request
+from fastapi import Depends, FastAPI, Query, Request
 from fastapi.concurrency import contextmanager_in_threadpool
+from pydantic import BaseModel, Field
+from pydantic.generics import GenericModel
 from sqlalchemy import engine_from_config
 from sqlalchemy.ext.declarative import DeferredReflection, declarative_base
+from sqlalchemy.orm import Query as DbQuery
 from sqlalchemy.orm.session import Session, sessionmaker
 
 __all__ = ["Base", "setup", "with_session"]
@@ -108,3 +113,76 @@ def get_users(session: sqla.Session = Depends(sqla.new_session)):
             await loop.run_in_executor(None, session.rollback)
 
     return response
+
+
+T = TypeVar("T")
+
+
+class Item(GenericModel, Generic[T]):
+    """Item container."""
+
+    data: T
+
+
+class Collection(GenericModel, Generic[T]):
+    """Collection container."""
+
+    data: List[T]
+
+
+class Meta(BaseModel):
+    """Meta information on current page and collection"""
+
+    offset: int = Field(..., description="Current page offset")
+    total_items: int = Field(..., description="Total number of items in the collection")
+    total_pages: int = Field(..., description="Total number of pages in the collection")
+    page_number: int = Field(..., description="Current page number. Starts at 1.")
+
+
+class Paginated(Collection, Generic[T]):
+    """Paginated collection with information on current page and total items in meta."""
+
+    meta: Meta
+
+
+def _query_count(session: Session, query: DbQuery) -> int:
+    """Default function used to count items returned by a query.
+
+    Default Query.count is slower than a manually written query could be: It runs the
+    query in a subquery, and count the number of elements returned:
+
+    See https://gist.github.com/hest/8798884
+    """
+    return query.count()
+
+
+def Pagination(
+    min_page_size: int = 10,
+    max_page_size: int = 100,
+    query_count: Callable[[Session, DbQuery], int] = _query_count,
+) -> Callable[[Session, int, int], Callable[[DbQuery], Paginated[T]]]:
+    def dependency(
+        session: Session = Depends(with_session),
+        offset: int = Query(0, ge=0),
+        limit: int = Query(min_page_size, ge=1, le=max_page_size),
+    ) -> Callable[[DbQuery], Paginated[T]]:
+        def paginated_result(query: DbQuery) -> Paginated[T]:
+            total_items = query_count(session, query)
+            total_pages = math.ceil(total_items / limit)
+            page_number = offset / limit + 1
+            return Paginated[T](
+                data=query.offset(offset).limit(limit).all(),
+                meta={
+                    "offset": offset,
+                    "total_items": total_items,
+                    "total_pages": total_pages,
+                    "page_number": page_number,
+                },
+            )
+
+        return paginated_result
+
+    return dependency
+
+
+with_pagination = Pagination()