chore: add tests covering session dependency usage

Author: hadrien

pyproject.toml

@@ -4,8 +4,12 @@ version = "0.2.0"
 description = "SQLAlchemy extension for FastAPI with support for asynchronous SQLAlchemy sessions and pagination."
 readme = "README.md"
 requires-python = ">=3.12"
-authors = [{ name = "Hadrien David", email = "[email protected]" }]
-dependencies = ["sqlalchemy[asyncio]>=2.0.34,<3", "structlog>=24.4.0"]
+authors = [{ name = "Hadrien David", email = "[email protected]" }]
+dependencies = [
+    "fastapi>=0.115.6",
+    "sqlalchemy[asyncio]>=2.0.34,<3",
+    "structlog>=24.4.0",
+]
 license = { text = "MIT License" }
 
 [tool.uv]
@@ -22,7 +26,6 @@ dev-dependencies = [
     "ruff>=0.6.4",
     "toml>=0.10.2",
     "aiosqlite>=0.20.0",
-    "fastapi>=0.114.0",
     "python-semantic-release>=9.8.8",
     "twine>=5.1.1",
 ]

src/fastapi_async_sqla.py

@@ -19,7 +19,7 @@
 
 __all__ = ["Session", "lifespan", "open_session"]
 
-SessionFactory = async_sessionmaker(class_=AsyncSession, expire_on_commit=False)
+SessionFactory = async_sessionmaker(expire_on_commit=False)
 
 logger = get_logger(__name__)
 

tests/conftest.py

@@ -1,7 +1,7 @@
 from unittest.mock import patch
 
 from pytest import fixture
-from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 
 
 @fixture
@@ -22,6 +22,12 @@ async def engine(environ):
     await engine.dispose()
 
 
+@fixture
+async def session(engine):
+    async with engine.connect() as conn:
+        yield AsyncSession(bind=conn)
+
+
 @fixture(autouse=True)
 def tear_down():
     from sqlalchemy.orm import clear_mappers

tests/integration/test_session_dependency.py

@@ -1,20 +1,89 @@
+from http import HTTPStatus
+
+from fastapi import HTTPException
+from pydantic import BaseModel, ConfigDict
 from pytest import fixture
 from sqlalchemy import select, text
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Mapped, mapped_column
+
+
+@fixture(autouse=True)
+async def setup_tear_down(engine):
+    async with engine.connect() as conn:
+        await conn.execute(
+            text("""
+            CREATE TABLE user (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                email TEXT UNIQUE NOT NULL,
+                name TEXT NOT NULL
+            )
+            """)
+        )
 
 
 @fixture
-def app(app):
-    from fastapi_async_sqla import Session
+def app(setup_tear_down, app):
+    from fastapi_async_sqla import Base, Item, Session
+
+    class User(Base):
+        __tablename__ = "user"
+        id: Mapped[int] = mapped_column(primary_key=True)
+        email: Mapped[str] = mapped_column(unique=True)
+        name: Mapped[str]
+
+    class UserIn(BaseModel):
+        email: str
+        name: str
+
+    class UserModel(UserIn):
+        model_config = ConfigDict(from_attributes=True)
+        id: int
 
     @app.get("/session-dependency")
     async def get_session(session: Session):
         res = await session.execute(select(text("'OK'")))
         return {"data": res.scalar()}
 
+    @app.post("/users", response_model=Item[UserModel], status_code=HTTPStatus.CREATED)
+    async def create_user(user_in: UserIn, session: Session):
+        user = User(**user_in.model_dump())
+        user_in.model_dump
+        session.add(user)
+        try:
+            await session.flush()
+        except IntegrityError:
+            raise HTTPException(status_code=400)
+        return {"data": user}
+
     return app
 
 
 async def test_it(client):
-    response = await client.get("/session-dependency")
-    assert response.status_code == 200
-    assert response.json() == {"data": "OK"}
+    res = await client.get("/session-dependency")
+    assert res.status_code == HTTPStatus.OK, (res.status_code, res.content)
+    assert res.json() == {"data": "OK"}
+
+
+async def test_session_is_commited(client, session):
+    payload = {"email": "[email protected]", "name": "Bobby"}
+    res = await client.post("/users", json=payload)
+
+    assert res.status_code == HTTPStatus.CREATED, (res.status_code, res.content)
+
+    all_users = (await session.execute(text("SELECT * FROM user"))).mappings().all()
+    assert all_users == [{"id": 1, **payload}]
+
+
+@fixture
+async def bob_exists(session):
+    await session.execute(
+        text("INSERT INTO user (email, name) VALUES ('[email protected]', 'Bobby')")
+    )
+    await session.commit()
+    yield
+
+
+async def test_with_an_integrity_error(client, bob_exists):
+    res = await client.post("/users", json={"email": "[email protected]", "name": "Bobby"})
+    assert res.status_code == HTTPStatus.BAD_REQUEST, (res.status_code, res.content)