Merge pull request #203 from uriyyo/feature/sqlmodel

🔧 Add sqlmodel integration

Author: uriyyo

.github/workflows/test.yml

@@ -4,8 +4,6 @@ on:
   push:
   pull_request:
   workflow_dispatch:
-  schedule:
-    - cron: "0 0 * * *"
 
 jobs:
   build:
@@ -36,11 +34,12 @@ jobs:
           poetry install -E all
 
       - name: Unit tests
-        run: poetry run pytest tests --cov-report=xml --ignore=tests/ext/test_async_sqlalchemy.py --ignore=tests/ext/test_sqlalchemy_future.py
+        run: poetry run pytest tests --cov-report=xml --ignore=tests/ext/test_async_sqlalchemy.py --ignore=tests/ext/test_sqlalchemy_future.py --ignore=tests/ext/test_sqlmodel.py
 
       - name: Unit tests sqlalchemy 1.4
         run: |
-          poetry run pip install -U "sqlalchemy>=1.4.11"  # FIXME: dependency conflict must be solved
+          poetry run pip install -U "sqlalchemy>=1.4.11" sqlmodel  # FIXME: dependency conflict must be solved
+          poetry run pytest tests/ext/test_sqlmodel.py --cov-append --cov-report=xml
           poetry run pytest tests/ext/test_async_sqlalchemy.py tests/ext/test_sqlalchemy_future.py --cov-append --cov-report=xml
 
       - name: Upload coverage to Codecov

README.md

@@ -27,6 +27,7 @@ Available integrations:
 * [tortoise](https://github.com/tortoise/tortoise-orm)
 * [django](https://github.com/django/django)
 * [piccolo](https://github.com/piccolo-orm/piccolo)
+* [sqlmodel](https://github.com/tiangolo/sqlmodel)
 
 ## Example
 

fastapi_pagination/ext/sqlmodel.py

@@ -0,0 +1,29 @@
+from typing import Optional, TypeVar, Union
+
+from sqlmodel import Session, SQLModel, func, select
+from sqlmodel.sql.expression import Select, SelectOfScalar
+
+from ..api import create_page, resolve_params
+from ..bases import AbstractPage, AbstractParams
+
+T = TypeVar("T", bound=SQLModel)
+
+
+def paginate(
+    session: Session,
+    query: Union[T, Select[T], SelectOfScalar[T]],
+    params: Optional[AbstractParams] = None,
+) -> AbstractPage[T]:
+    params = resolve_params(params)
+    raw_params = params.to_raw_params()
+
+    if not isinstance(query, (Select, SelectOfScalar)):
+        query = select(query)
+
+    total = session.scalar(select(func.count("*")).select_from(query.subquery()))
+    items = session.exec(query.limit(raw_params.limit).offset(raw_params.offset)).unique().all()
+
+    return create_page(items, total, params)
+
+
+__all__ = ["paginate"]

pyproject.toml

@@ -35,6 +35,8 @@ ormar = { version = ">=0.10.5", optional = true}
 Django = { version = "<3.3.0", optional = true}
 piccolo = { version = ">=0.29,<0.35", optional = true}
 
+# sqlmodel = { version = "^0.0.4", optional = true} Should be uncommented when gino will support sqlalchemy 1.4
+
 [tool.poetry.dev-dependencies]
 pytest = "^6.2"
 pytest-cov = "^2.12.1"

tests/ext/test_sqlmodel.py

@@ -0,0 +1,85 @@
+from functools import partial
+from typing import Iterator
+
+from fastapi import Depends, FastAPI
+from pytest import fixture, mark
+from sqlmodel import Field, Session, SQLModel, create_engine, select
+
+from fastapi_pagination import LimitOffsetPage, Page, add_pagination
+from fastapi_pagination.ext.sqlmodel import paginate
+
+from ..base import BasePaginationTestCase, SafeTestClient, UserOut
+from ..utils import faker
+
+
+@fixture(scope="session")
+def engine(database_url):
+    if database_url.startswith("sqlite"):
+        connect_args = {"check_same_thread": False}
+    else:
+        connect_args = {}
+
+    return create_engine(database_url, connect_args=connect_args)
+
+
+@fixture(scope="session")
+def SessionLocal(engine):
+    return partial(Session, engine)
+
+
+@fixture(scope="session")
+def User():
+    class User(SQLModel, table=True):
+        id: int = Field(primary_key=True)
+        name: str
+
+    return User
+
+
+@fixture(
+    scope="session",
+    params=[True, False],
+    ids=["model", "query"],
+)
+def query(request, User):
+    if request.param:
+        return User
+    else:
+        return select(User)
+
+
+@fixture(scope="session")
+def app(query, engine, User, SessionLocal):
+    app = FastAPI()
+
+    @app.on_event("startup")
+    def on_startup():
+        SQLModel.metadata.create_all(engine)
+
+        with SessionLocal() as session:
+            session.add_all([User(name=faker.name()) for _ in range(100)])
+
+    def get_db() -> Iterator[Session]:
+        with SessionLocal() as db:
+            yield db
+
+    @app.get("/default", response_model=Page[UserOut])
+    @app.get("/limit-offset", response_model=LimitOffsetPage[UserOut])
+    def route(db: Session = Depends(get_db)):
+        return paginate(db, query)
+
+    add_pagination(app)
+    return app
+
+
+@mark.future_sqlalchemy
+class TestSQLModel(BasePaginationTestCase):
+    @fixture(scope="session")
+    def client(self, app):
+        with SafeTestClient(app) as c:
+            yield c
+
+    @fixture(scope="session")
+    def entities(self, SessionLocal, User):
+        with SessionLocal() as session:
+            return session.exec(select(User)).unique().all()