# Recipe — Async SQLAlchemy with `mgf.common.db`

> **Audience:** consumers building async services that use SQLAlchemy 2
> with Postgres / MySQL / SQLite. Pairs naturally with FastAPI but
> works in any asyncio context.
> **Optional extra:** `pip install 'mgf-common[db]'` (pulls
> `sqlalchemy[asyncio]>=2.0`).

## At a glance

```python
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from mgf.common._lifecycle import bootstrap
from mgf.common.db import get_session, setup_db

@asynccontextmanager
async def lifespan(app: FastAPI):
    with bootstrap(app_name="myapp", app_version="1.0.0"):
        async with setup_db(
            app,
            database_url="postgresql+asyncpg://user:pass@host/db",
        ):
            yield

app = FastAPI(lifespan=lifespan)

@app.get("/users")
async def list_users(
    session: Annotated[AsyncSession, Depends(get_session)],
) -> list[dict]:
    rows = await session.execute(text("SELECT id, name FROM users"))
    return [{"id": r.id, "name": r.name} for r in rows]
```

What you get for free:
- Engine constructed at startup with `pool_pre_ping=True` + 1-hour
  pool_recycle (catches stale connections after DB restart).
- Pool disposed cleanly on app shutdown.
- One async session per request via the `Depends`-compatible helper.
- Multi-tenant `SET LOCAL app.current_tenant` helper for Postgres RLS
  (with UUID validation to prevent SQL injection).

## The five pieces

### 1. `create_engine` — async engine factory

```python
from mgf.common.db import create_engine

engine = create_engine(
    "postgresql+asyncpg://user:pass@host/db",
    echo=False,                # True logs every statement (debug only)
    pool_pre_ping=True,         # default; pings on every checkout
    pool_recycle=3600,          # default; recycle conns >1h
    pool_size=5,                # default
    max_overflow=10,            # default
)
```

Production-leaning defaults applied automatically. `echo=True` only
in debug. SQLite drivers don't have a real pool — `pool_size` /
`max_overflow` are ignored for SQLite URLs.

### 2. `create_sessionmaker`

```python
from mgf.common.db import create_sessionmaker

sessionmaker = create_sessionmaker(engine)
# expire_on_commit=False by default (SQLAlchemy 2 async-recommended).

# Or override:
sessionmaker = create_sessionmaker(engine, expire_on_commit=True)
```

The default `expire_on_commit=False` keeps objects usable after
`commit()`. Pass `expire_on_commit=True` if you need legacy semantics.

### 3. `setup_db` — engine lifecycle in FastAPI lifespan

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from mgf.common._lifecycle import bootstrap
from mgf.common.db import setup_db

@asynccontextmanager
async def lifespan(app: FastAPI):
    with bootstrap(app_name="myapp", app_version="1.0.0"):
        async with setup_db(
            app,
            database_url=settings.db_url,
            echo=settings.db_echo,
            pool_size=20,
        ):
            yield  # app is now serving requests

app = FastAPI(lifespan=lifespan)
```

Inside the `setup_db` context:
- `app.state.mgf_db_engine` → `AsyncEngine`
- `app.state.mgf_db_sessionmaker` → factory

On exit (signal, ASGI shutdown), the engine pool is disposed cleanly.

### 4. `get_session` — FastAPI Depends

```python
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from mgf.common.db import get_session

@app.get("/users/{user_id}")
async def read_user(
    user_id: str,
    session: Annotated[AsyncSession, Depends(get_session)],
) -> dict:
    row = await session.get(User, user_id)
    return {"id": row.id, "name": row.name}
```

One session per request; closed via the context manager when the
handler returns.

**Commit / rollback is the consumer's call.** No global policy is
imposed. For mutation endpoints, wrap your work in
`async with session.begin():` or call `session.commit()` explicitly:

```python
@app.post("/users")
async def create_user(
    body: UserIn,
    session: Annotated[AsyncSession, Depends(get_session)],
) -> dict:
    async with session.begin():
        user = User(**body.model_dump())
        session.add(user)
    return {"id": user.id}
```

If `app.state.mgf_db_sessionmaker` isn't populated (forgot
`setup_db` in the lifespan), `get_session` raises
`AppConfigError` — which `ExceptionTranslationMiddleware` maps to
500 with an actionable message.

### 5. `tenant_session` — Postgres RLS multi-tenancy

```python
from uuid import UUID
from mgf.common.db import tenant_session

@app.get("/devices")
async def list_devices(
    request: Request,
    session: Annotated[AsyncSession, Depends(get_session)],
) -> list[dict]:
    tenant_id = extract_tenant_from_jwt(request)  # e.g., UUID
    async with tenant_session(session, tenant_id):
        # All queries on `session` (same object — tenant context is
        # set on the session's current transaction) automatically
        # see only rows where the RLS policy admits the tenant.
        rows = await session.execute(text("SELECT id FROM devices"))
        return [{"id": r.id} for r in rows]
```

Runs `SET LOCAL app.current_tenant = '<uuid>'` for the session.
`LOCAL` scopes the setting to the current transaction — auto-resets
on commit / rollback. Cross-tenant leakage is impossible at the SQL
layer (assuming your RLS policy reads
`current_setting('app.current_tenant')`).

**Tenant id is UUID-validated.** Postgres `SET LOCAL` doesn't accept
bind parameters, so we have to interpolate the value — which is safe
iff the input is shape-validated. `tenant_session(session, "not-a-uuid")`
raises `ValueError` before any SQL runs.

#### Setting up the Postgres side

The custom GUC `app.current_tenant` must exist. Add to
`postgresql.conf`:

```
custom_variable_classes = 'app' # Postgres < 9.2 only
```

For modern Postgres (9.2+), no config needed — custom GUCs work out
of the box if the name has a dot. Then your RLS policy:

```sql
ALTER TABLE devices ENABLE ROW LEVEL SECURITY;

CREATE POLICY tenant_isolation ON devices
    USING (tenant_id = current_setting('app.current_tenant')::uuid);
```

## Common patterns

### Manual session (outside FastAPI)

```python
from mgf.common.db import create_engine, create_sessionmaker

engine = create_engine("postgresql+asyncpg://...")
sessionmaker = create_sessionmaker(engine)

async with sessionmaker() as session:
    async with session.begin():
        row = await session.execute(text("SELECT 1"))

await engine.dispose()
```

### Combining `bootstrap` + `setup_db` + `mgf.common.fastapi`

The full service shape:

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from mgf.common._lifecycle import bootstrap
from mgf.common.db import setup_db
from mgf.common.fastapi import (
    ExceptionTranslationMiddleware,
    RequestIdMiddleware,
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    with bootstrap(app_name="myapp", app_version="1.0.0"):
        async with setup_db(app, database_url=settings.db_url):
            yield

app = FastAPI(lifespan=lifespan)
app.add_middleware(ExceptionTranslationMiddleware)
app.add_middleware(RequestIdMiddleware)
```

That's:
- structured JSON logging at WARNING level (or your settings level)
- crash reporter installed
- OTel wired (when configured)
- per-request UUID4 with X-Request-Id propagation
- typed error → JSON response (404 / 409 / 503 / 500)
- async DB session per request
- engine pool disposed on shutdown

### Testing

```python
import pytest
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.testclient import TestClient
from mgf.common.db import setup_db

def make_test_app() -> FastAPI:
    @asynccontextmanager
    async def _lifespan(app: FastAPI):
        async with setup_db(app, database_url="sqlite+aiosqlite:///:memory:"):
            yield
    return FastAPI(lifespan=_lifespan)

def test_my_endpoint() -> None:
    with TestClient(make_test_app()) as client:
        response = client.get("/health")
    assert response.status_code == 200
```

In-memory SQLite gives a fast, hermetic per-test database. For
integration tests against Postgres / MySQL specifics (RLS, JSONB,
geospatial), use testcontainers.

## Anti-patterns

- **Don't reuse sessions across requests.** The `Depends`-driven
  pattern gives one session per request; that's the right shape.
  Persistent sessions create connection leaks under load.
- **Don't call `tenant_session` outside a transaction.** `SET LOCAL`
  scopes to the current transaction. Without one, the setting may
  leak to subsequent uses of the same connection.
- **Don't pass user input to `tenant_session` without validation.**
  The wrapper validates, but if you're constructing your own SQL
  with tenant ids, bind-or-validate; never interpolate raw input.
- **Don't forget `await engine.dispose()`.** `setup_db` does this
  for you in the lifespan; if you create engines manually, dispose
  them or you'll leak connections at process exit.

## See also

- `docs/public/recipes/fastapi.md` — FastAPI middleware + Depends.
- `docs/public/recipes/http.md` — `mgf.common.http` for upstream APIs.
- `docs/public/recipes/testing.md` — pytest fixtures.
- SQLAlchemy 2.x async docs:
  https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html
