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

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.

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:

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:

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:

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:

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.

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:

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(...):

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:

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=:

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 Validation & Injection 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=...):

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():

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

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

• Validation & Injection — ForeignKeyValidator, StringValidator, and custom validators

• JSON encoding — CustomJSONEncoder base class, encoder context, and the general encoding pattern

• Writing your own integration — how to add a new optional integration to EDRI