# SQLAlchemy (`edri.integrations.sqlalchemy`)

Optional, first-party support for SQLAlchemy ORM models: explicit response
projections, safe session ownership across EDRI's multiprocessing boundary,
model serialization, strict filter/sort builders, foreign-key injection, and a
compatibility JSON encoder for simpler detached-model applications.

## Prerequisites

```bash
pip install edri[sqlalchemy]
```

SQLAlchemy is an **optional** dependency of EDRI — it is declared as an extra in
``pyproject.toml``. The integration module guards its import, so a bare
``ModuleNotFoundError`` is replaced with a clear message naming the extra.

```python
from edri.integrations.sqlalchemy import (
    ForeignKeyValidator,
    SQLAlchemyManager,
    SQLAlchemyModelMixin,
    SQLAlchemyResponseMiddleware,
    orm_projection,
    orm_validator,
    projection,
)
```

## Projected ORM responses

Use projections when an API lets clients select fields or when ORM responses
cross a Manager/router process boundary. A projection is an explicit contract:
only declared columns and relationships may be requested, SQLAlchemy loads the
whole graph before the session closes, and the broker converts it to transport
values without issuing SQL.

The flow is:

```text
request fields -> projection() -> ORM query -> Manager finalization/detach
               -> router queue -> SQLAlchemyResponseMiddleware -> JSON
```

### Model base

Put `SQLAlchemyModelMixin` before `DeclarativeBase`. `__json__` defines the
shape used only on the non-projected path — bare ORM responses and
`SessionJSONEncoder`; a model used exclusively through projections does not
need it. Override `_json_extra()` only for derived values or request-specific
access rules:

```python
from edri.integrations.sqlalchemy import SQLAlchemyModelMixin
from sqlalchemy.orm import DeclarativeBase

class Base(SQLAlchemyModelMixin, DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user"
    __json__ = ("id_user", "name", "organization")  # optional; non-projected path only
    # mapped_column(...) and relationship(...) declarations
```

### Response declaration

Declare the projection on a normal `dataclasses.field`. EDRI preserves its
default/factory and metadata natively:

```python
from dataclasses import field
from edri.dataclass.response import Response, response
from edri.integrations.sqlalchemy import orm_count, orm_projection

@response
class UserListResponse(Response):
    users: list[User] = field(
        default_factory=list,
        metadata=orm_projection(
            source="columns",
            default=("idUser", "name"),
            allowed=(
                "idUser",
                "name",
                "organization.idOrganization",
                "organization.name",
            ),
            dependencies=("idOrganization",),
            functions=(orm_count("sessions"),),
        ),
    )
```

`source` names the Event field carrying requested public paths. `default` is
used when that field is empty; with no `default`, an empty request emits the
whole `allowed` set. `allowed` is the complete public allowlist, and
`dependencies` are loaded but not emitted. The projection is a self-contained
contract — it does not consult the model's `__json__`. Public camel-case paths
are resolved strictly against mapped attributes. An explicit request for an
unknown path, scalar traversal, a relationship not in the allowlist, or a
function not declared in `functions` raises `ProjectionError`.

`orm_count("sessions")` whitelists the exact request `count(sessions)` and
adds a correlated scalar subquery. It does not load the collection, add a root
`GROUP BY`, or introduce one query per row.

Self-referential relationships can be repeated with `recursive=("parent",)`.
Recursion activates only when the relationship path itself is selected and is
capped at 64 loader levels. Repeated identities serialize as their primary-key
shape, preventing cyclic output.

### Query execution

Apply the projection before execution. Use the supplied materializers so any
computed projection values are retained:

```python
from sqlalchemy import select
from edri.integrations.sqlalchemy import projection, projection_all

statement = projection(
    select(User).order_by(User.name),
    event=event,
    field="users",
)
event.response.users = projection_all(
    session,
    statement,
    response=event.response,
    field="users",
)
```

Use `projection_one()` for exactly one model and `projection_rows()` when the
statement intentionally selects additional columns. Projection loader plans use
`load_only(..., raiseload=True)`, wildcard relationship `raiseload`, and nested
`selectinload`; omitted state therefore fails explicitly instead of silently
causing lazy SQL.

### Manager and broker registration

Managers returning ORM values inherit from `SQLAlchemyManager`. It keeps
registered sessions alive for one resolution, validates/completes projected
graphs, and detaches them before multiprocessing. For another Manager variant,
put `SQLAlchemyManagerMixin` first in the inheritance list.

```python
from edri.integrations.sqlalchemy import (
    SQLAlchemyManager,
    SQLAlchemyResponseMiddleware,
)

class UserManager(SQLAlchemyManager):
    ...

edri.start_api(middlewares=[SQLAlchemyResponseMiddleware(), ...])
```

`SQLAlchemyResponseMiddleware` runs on the broker side. It validates the whole
response before mutation, calls the model mixin with the resolved plan, and
never opens a session or reloads state. Missing plans, unloaded fields, and
cycles raise `IncompleteProjectionError` with the failing response path.

Create application sessions with `ResponseAwareSession` (usually through
`sessionmaker(class_=ResponseAwareSession, expire_on_commit=False)`). Calls to
`close()` during an active Manager resolution are deferred until projected
responses have been finalized.

## ORM Events

`ORMEvent` and `@orm_event` identify fields injected through
`ForeignKeyValidator` and merge their detached values into a handler's session.
The merge session is automatically registered with the current resolution:

```python
from dataclasses import dataclass
from edri.integrations.sqlalchemy import ORMEvent, ForeignKeyValidator, orm_event
from edri.utility.function import inject

@orm_event
@dataclass
class UpdateUser(ORMEvent):
    user: inject(ForeignKeyValidator[User], session_factory=SessionLocal)

# inside the handler
event.merge(session, load=False)
```

## Validators derived from ORM columns

`orm_validator()` is a validator factory for request fields. It reads one
mapped column when the event class is declared and returns a one-argument
validator callable for `inject(...)`:

```python
from edri.integrations.sqlalchemy import orm_validator
from edri.utility.function import inject

@api(url="/users", method=HTTPMethod.POST)
class CreateUser(Event):
    name: inject(orm_validator(User.name))
    nickname: inject(orm_validator(User.nickname)) | None = None
    response: CreateUserResponse
```

For `User.name = mapped_column(String(100))`, the generated callable uses
`StringValidator(maximum_length=100)`. Changing the ORM length therefore also
changes request validation without repeating `100` in the event. The factory
also selects the matching integer, float, date, time, or datetime validator,
and falls back to the column type's Python type through normal EDRI coercion
for types without a dedicated validator (`UUID`, `Decimal`, `bool`).

A `SQLAlchemyEnum("public", "private")` column restricts the field to those
strings — the allowed values come from the column, not a repeated
`Literal[...]` in the event. A column mapped to a Python `enum.Enum` needs
nothing from the factory: it is coerced against that class (`SomeEnum(value)`,
matched by member value), the same as a plain `field: SomeEnum` annotation.

Keyword arguments are forwarded to the derived validator and take precedence
over anything read from the column:

```python
name: inject(orm_validator(User.name, maximum_length=64, minimum_length=2))
```

A keyword the underlying validator does not accept, or any keyword on a
column with no dedicated validator, is rejected when the event class is
declared. Overrides go to `orm_validator()`, not to `inject()` — a keyword
passed to `inject()` around the generated callable is filtered out and has
no effect.

Nullability remains explicit in the event because database nullability and
request-field optionality are separate contracts. Uniqueness and foreign keys
also remain database/session concerns; use `ForeignKeyValidator` when an input
id must load a related model.

## Strict query builders

`FilterBuilder` turns parallel client filter arrays into typed SQLAlchemy
expressions. Dotted paths use correlated `has()`/`any()` expressions; they do
not guess foreign keys or association objects. Negative operators (`not`,
`doesNotContain`, `isNoneOf`, `notBetween`, `isEmpty`, …) match `NULL` values:
on a scalar column they add `OR … IS NULL`, and on a dotted path they negate the
positive `EXISTS`. So `owner.name not "Acme"` also matches rows without an
owner, and `items.name not "Acme"` means that no item is named `Acme`.
Configure unknown paths with
`unknown_field_policy="ignore"`, `"log"`, `"raise"`, or `"collect"`.

`SortBuilder.build()` resolves scalar sort paths. `apply_sorting()` then adds
the required aliased outer joins and `ORDER BY` clauses. Keeping resolution and
statement mutation separate makes accepted client paths auditable.

## `ForeignKeyValidator[T]`

An ``inject(...)`` validator that converts a raw value (typically a URL path
segment) into a loaded ORM model instance.  The target model type is taken from
the subscript (``ForeignKeyValidator[User]``), and the database session is
supplied per field via ``session_factory=``:

```python
from edri.integrations.sqlalchemy import ForeignKeyValidator
from app.db import SessionLocal  # a sessionmaker
from app.models import User

@api(url="/users/{user}", method=HTTPMethod.GET)
class GetUser(Event):
    user: inject(ForeignKeyValidator[User], session_factory=SessionLocal)
    response: GetUserResponse
```

The validator:

1. Reads the model's primary-key type from the SQLAlchemy mapper and coerces
   the raw value to it — ``int``, ``uuid.UUID``, ``str`` and similar scalar keys
   all work. A value that cannot be coerced raises ``ValueError`` (which EDRI
   maps to a validation failure); a model with a **composite** primary key is
   unsupported and also raises ``ValueError``.
2. Opens a short-lived session via ``session_factory()``.
3. Calls ``session.get(model, pk)``.
4. Raises ``ValueError`` if no row is found.

The result is the loaded ORM instance, available as ``self.user`` inside the
handler — no manual lookup code needed.

Because the session is opened and closed inside the validator call, the
returned model instance is **detached** by the time the handler runs. For read
operations this is rarely a problem (all column values were loaded by
``session.get``). If you need lazy relationships or further database access,
see `SessionJSONEncoder` below for serialization, or open a longer-lived
session in your handler.

See {doc}`/validation` for the general validator pattern.

## `SessionJSONEncoder` (compatibility mode)

```{note}
Prefer the projection flow above for new code. `SessionJSONEncoder` opens a
database session and issues SQL from the broker at serialization time, once per
mapped instance; it exists for existing applications and for simple single-node
cases where that cost is acceptable. It is not the path for selective fields or
for responses that cross a process boundary.
```

A JSON encoder that re-attaches detached ORM instances to a fresh session
before serializing them.  This is necessary because a model instance loaded in
one session and then returned in a response is **detached** by the time EDRI
renders the payload — accessing a lazy relationship or an expired column would
fail.

`SessionJSONEncoder` extends ``CustomJSONEncoder``.  Only mapped ORM instances
that expose a ``to_json()`` method are intercepted; everything else falls
through to the base class encoding (dataclasses, enums, UUIDs, …).

### Usage

Subclass `SessionJSONEncoder` and implement ``get_session()`` to return a
context-managed ``Session``, then pass the subclass to
``edri.start_api(json_encoder=...)``:

```python
from edri.integrations.sqlalchemy import SessionJSONEncoder
from app.db import SessionLocal  # a sessionmaker

class AppJSONEncoder(SessionJSONEncoder):
    def get_session(self):
        return SessionLocal()

edri.start_api(json_encoder=AppJSONEncoder)
```

### How it works

When the encoder encounters an object that has a ``to_json()`` method **and**
is a mapped SQLAlchemy ORM instance (detected via ``object_mapper`` — base
class agnostic), it:

1. Opens a session via ``self.get_session()``.
2. Merges the detached instance into that session (``session.merge(data, load=True)``).
3. Calls ``merged.to_json()``, which can safely access lazy relationships and
   expired columns.
4. Closes the session.

Every mapped instance in the response therefore costs one session and at least
one round trip, from the broker. The projection flow above does this work once
in the Manager and leaves broker-side serialization free of database access;
use it whenever responses select fields or cross a Manager/router boundary.

The encoder is invoked per field, but the session is opened and closed for each
mapped instance individually — there is no long-lived session shared across the
entire response.  For more control, use ``self.context`` (populated from
``Event.get_context()``) to influence session creation per request.

### Per-request session via context

The encoder's ``self.context`` dictionary is set from the event's keyword-only
fields (see ``Event.get_context()``).  If your ``SessionJSONEncoder`` subclass
needs a request-scoped session (for example one that carries the authenticated
user's tenant filter), read ``self.context`` inside ``get_session()``:

```python
class TenantAwareEncoder(SessionJSONEncoder):
    def get_session(self):
        tenant_id = (self.context or {}).get("tenant_id")
        return SessionLocal(tenant_id=tenant_id)
```

## Example: query and return related data

```python
from edri import EDRI, event, response
from edri.api import HTTPMethod, api
from edri.integrations.sqlalchemy import ForeignKeyValidator, SessionJSONEncoder
from app.db import SessionLocal
from app.models import Order

class AppEncoder(SessionJSONEncoder):
    def get_session(self):
        return SessionLocal()

@event
class GetOrder(Event):
    order_id: inject(ForeignKeyValidator[Order], session_factory=SessionLocal)
    response: OrderResponse

@response
def handle_get_order(self, event: GetOrder) -> None:
    # event.order_id is a loaded Order instance (detached)
    event.response.order = event.order_id

edri = EDRI()
edri.start_api(json_encoder=AppEncoder, ...)
edri.run()
```

The ``Order`` model's ``to_json()`` can access relationships because
``SessionJSONEncoder`` re-attaches it before calling ``to_json()``.

## See also

* {doc}`/validation` — ``ForeignKeyValidator``, ``StringValidator``,
  and custom validators
* {doc}`/json_encoder` — ``CustomJSONEncoder`` base class,
  encoder context, and the general encoding pattern
* {doc}`/integrations/custom` — how to add a new optional
  integration to EDRI