prompts/docs/skills/async-fastapi-sqlmodel/references/fastapi.md

# FastAPI Database Integration

!!! info "Primary sources"
    - [FastAPI lifespan events](https://fastapi.tiangolo.com/advanced/events/)
    - [FastAPI dependencies with `yield`](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/)
    - [FastAPI dependency overrides](https://fastapi.tiangolo.com/advanced/testing-dependencies/)
    - [SQLAlchemy asyncio extension](https://docs.sqlalchemy.org/en/21/orm/extensions/asyncio.html)
    - [`nicegui-db` application lifespan](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/ui/app.py)
    - [`nicegui-db` database dependencies](https://forgejo.john-stream.com/john/nicegui-db/src/commit/126bc26ad8635a86bacf684d7bda409230347597/src/nicegui_db/ui/dependency.py)

---

## Purpose

Connect the framework-independent database tools to FastAPI:

- lifespan enters one application-owned `database_scope()`,
- application state holds settings and the resulting session factory,
- dependencies create one session per request,
- `Annotated` aliases make route ownership concise and explicit.

The underlying resource and transaction rules remain in [engine lifecycle](engine.md), [session management](session.md), and [transaction boundaries](transactions.md).

---

## Lifespan Ownership

Enter `database_scope()` once for the complete application lifecycle. Store the session factory, not the engine, because request code needs sessions rather than direct pool access:

```python
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager

from fastapi import FastAPI

from .config import Settings
from .config import get_database_url
from .db import database_scope


@asynccontextmanager
async def lifespan(settings: Settings, app: FastAPI) -> AsyncGenerator[None]:
    app.state.settings = settings
    db_url = get_database_url(settings)

    try:
        async with database_scope(db_url) as session_factory:
            app.state.session_factory = session_factory
            yield
    finally:
        del app.state.settings
        del app.state.session_factory
```

The application factory binds `settings` to lifespan, for example with `partial(lifespan, settings)`. Lifespan does not construct resources per request. It enters the same framework-independent scope used by scripts, workers, and tests, keeps that scope open while requests are served, and lets it dispose the engine and clear cached engine resolution during shutdown.

The template's unconditional `del app.state.session_factory` mirrors an expected successful startup. If `database_scope()` raises before assignment, cleanup can raise `AttributeError` and obscure the startup error. A production hardening option is to assign a sentinel before the `try` or delete conditionally; that changes failure behavior and is not part of the exact template mechanics.

Only store the engine too when application-level code genuinely needs direct Core operations, pool instrumentation, or engine-specific diagnostics. Routes and repositories should normally receive an `AsyncSession`.

---

## Session Factory Dependency

A synchronous dependency retrieves the already-created factory from application state:

```python
from typing import Annotated

from fastapi import Depends
from fastapi import Request

from .session import SessionFactory


def _get_session_factory(request: Request) -> SessionFactory:
    return request.app.state.session_factory


type SessionFactoryDep = Annotated[SessionFactory, Depends(_get_session_factory)]
```

`Depends()` does not create or cache a factory here. It only exposes the lifespan-owned object. This function is also the narrow seam that tests can override when they need a different factory.

---

## Request Session Dependencies

Use a session-only dependency for reads and other request conversations that must not commit implicitly:

```python
from collections.abc import AsyncGenerator

from sqlmodel.ext.asyncio.session import AsyncSession


async def _get_session(session_factory: SessionFactoryDep) -> AsyncGenerator[AsyncSession]:
    async with session_factory() as owned_session:
        yield owned_session


type SessionDep = Annotated[AsyncSession, Depends(_get_session)]
```

The dependency creates and closes one session per request. Closing rolls back any unfinished autobegun transaction; it does not commit.

---

## Route Usage

Read route:

```python
@router.get("/items/{item_id}")
async def get_item(item_id: int, session: SessionDep) -> Item | None:
    return await find_item(session, item_id)
```

Write route:

```python
@router.post("/items")
async def create_item(payload: ItemCreate, session: SessionDep) -> Item:
    async with session.begin():
        return await insert_item(session, payload)
```

The template exposes only `SessionDep`; it does not hide commit behavior in dependency teardown. Choose one visible write convention per application:

- place `async with session.begin():` around a complete write unit, which commits on success and rolls back on exception; or
- call `await session.commit()` explicitly after all writes when the route is the complete unit, as the template's simple UI action does.

The context-manager form scales better to several statements and makes exception rollback visible. Direct `commit()` is concise but requires the route to preserve the single-commit invariant and handle any recovery needs. Do not combine both conventions in one route. Lower-level data-access functions receive the existing session and remain unaware of FastAPI.

---

## Background Work

A request session belongs to that request and must not be retained by a background task. Inject or otherwise provide the application session factory, then create a new session inside the task:

```python
async def run_background_job(session_factory: SessionFactory) -> None:
    async with session_factory.begin() as session:
        await process_pending_items(session)
```

If work must survive application shutdown, it needs an independently owned worker lifecycle rather than the FastAPI lifespan-owned factory.

---

## Testing and Overrides

Override the narrow dependency that matches the test objective:

- Override `_get_session_factory` to preserve production request-session behavior with a test factory.
- Override `_get_session` when a test must inject one transaction-scoped session directly.
- Verify each lifespan receives a fresh engine and session factory and removes application state during teardown.
- Remove overrides during teardown so mutable application state does not leak between tests.

```python
app.dependency_overrides[_get_session] = get_test_session
try:
    yield app
finally:
    app.dependency_overrides.pop(_get_session, None)
```

See [database testing](testing.md) for outer transactions, SAVEPOINT-backed fixtures, and database target selection.

---

## Anti-Patterns

- Creating an engine or session factory in a request dependency.
- Reading settings and constructing database resources from repositories.
- Storing one mutable `AsyncSession` on `app.state`.
- Sharing a request session with concurrent or background tasks.
- Assuming `SessionDep` commits when dependency cleanup runs.
- Keeping `app.state.session_factory` after its `database_scope()` exits.
- Using deprecated startup and shutdown event handlers alongside lifespan.

---

## Integration Checklist

- Lifespan enters exactly one `database_scope()` for each application lifecycle.
- Application state stores the yielded session factory.
- Session dependencies create and close one session per request.
- The session dependency owns request session closure but not commit behavior.
- Routes use `Annotated` aliases and receive sessions, not engines.
- Background tasks create their own sessions from a still-live factory.
- Tests override and restore dependencies deterministically.