edri.integrations.sqlalchemy package

Submodules

edri.integrations.sqlalchemy.event module

class edri.integrations.sqlalchemy.event.ORMEvent

Bases: Event

merge(session: sqlalchemy.orm.Session, load: bool = False) → None

Merge all mapped ORM instances in this event into the given Session. Uses cached field info in _orm_fields to be fast. (Assumes @orm_event has already populated _orm_fields.)

edri.integrations.sqlalchemy.event.orm_event(cls: Type[T]) → Type[T]

Class decorator for SQLEvent subclasses.

It precomputes the _orm_fields mapping at import time so merge() doesn’t need to do any reflection.

edri.integrations.sqlalchemy.filter module

class edri.integrations.sqlalchemy.filter.FilterBuilder(unknown_field_policy: Literal['ignore', 'log', 'raise', 'collect'] = 'log')

Bases: object

Build SQLAlchemy filter expressions from client-style payloads.

Expected payload shape:

names:   ["createdAt", "subject.idSubject", ...]
options: ["before", "is", ...]
values:  ["2026-01-17T00:00:00.000Z", "177", ...]

Resolution is deliberately dumb and strict:

• name resolves only to a mapped scalar column/property on table.

• subject.name resolves subject as a mapped relationship and name as a mapped scalar column/property on the related model.

• deeper paths work the same way, e.g. subject.organization.idOrganization.

• every intermediate dotted segment must be a mapped relationship.

• the final segment must be a mapped scalar column/property.

• a relationship without a final dotted column is filterable only with isEmpty/isNotEmpty (including their null aliases).

• no FK/PK guessing is performed.

• no id_<field> fallback is performed.

• no association-object inference is performed.

• no relationship-PK inference is performed.

The frontend/API serializer is responsible for sending the exact backend field path. For example, a DataGrid column may display subject.name while its API filter field is explicitly mapped to subject.id_subject.

ARRAY_OPS = frozenset({'!=', '=', 'contains', 'equalsArray', 'has', 'notEqualsArray'})

ENUM_OPS = frozenset({'!=', '=', 'equalsIgnoreCase'})

EQUALITY_OPS = frozenset({'!=', '='})

FALSE_VALUES = frozenset({'0', 'f', 'false', 'n', 'no', 'off'})

MEMBERSHIP_OPS = frozenset({'isAnyOf', 'isNoneOf'})

NEGATED_OPS = {'!=': '=', 'doesNotContain': 'contains', 'isEmpty': 'isNotEmpty', 'isNoneOf': 'isAnyOf', 'notBetween': 'between', 'notEqualsArray': 'equalsArray'}

NULL_OPS = frozenset({'isEmpty', 'isNotEmpty'})

OPTION_ALIASES = {'after': '>', 'afterOrEqual': '>=', 'afterOrEquals': '>=', 'anyOf': 'isAnyOf', 'before': '<', 'beforeOrEqual': '<=', 'beforeOrEquals': '<=', 'beginsWith': 'startsWith', 'doesNotEqual': '!=', 'endsIn': 'endsWith', 'equals': '=', 'equalsCI': 'equalsIgnoreCase', 'ieq': 'equalsIgnoreCase', 'in': 'isAnyOf', 'inRange': 'between', 'is': '=', 'isIn': 'isAnyOf', 'isNotIn': 'isNoneOf', 'isNotNull': 'isNotEmpty', 'isNull': 'isEmpty', 'like': 'contains', 'noneOf': 'isNoneOf', 'not': '!=', 'notContains': 'doesNotContain', 'notIn': 'isNoneOf', 'notInRange': 'notBetween', 'onOrAfter': '>=', 'onOrBefore': '<='}

ORDER_OPS = frozenset({'<', '<=', '>', '>='})

RANGE_OPS = frozenset({'between', 'notBetween'})

STRING_MATCH_OPS = frozenset({'contains', 'doesNotContain', 'endsWith', 'equalsIgnoreCase', 'startsWith'})

TRUE_VALUES = frozenset({'1', 'on', 't', 'true', 'y', 'yes'})

build(table: Any, names: list[str], options: list[str], values: list[str], operator: Literal['and', 'or'] | None = 'and') → ColumnElement[bool] | None

Build one SQLAlchemy boolean expression from parallel filter arrays.

names are interpreted exactly as model paths after the existing camelCase -> snake_case normalization of each segment.

class edri.integrations.sqlalchemy.filter.FilterBuilderConfig(unknown_field_policy: Literal['ignore', 'log', 'raise', 'collect'] = 'log')

Bases: object

Configuration for strict filter resolution.

unknown_field_policy: Literal['ignore', 'log', 'raise', 'collect'] = 'log'

edri.integrations.sqlalchemy.middleware module

class edri.integrations.sqlalchemy.middleware.SQLAlchemyResponseMiddleware

Bases: Middleware

Convert already-loaded ORM response values into transport values.

process_response(event: Event, info: EventInfo | None) → MiddlewareControl

Method to process the response before it is sent to the client. This method should be implemented in a subclass to perform any necessary postprocessing or formatting on the response.

Parameters:

• event -- The response event to be processed.

• info -- Information about the event.

Returns:

A MiddlewareControl object indicating whether to continue processing, skip further middleware, or terminate the request/response cycle.

edri.integrations.sqlalchemy.middleware.normalize_event_response(event: Event) → None

Convert already-loaded ORM response values into transport values.

The manager has already completed and validated projected graphs before queue transport. This broker-side step only creates transport values.

edri.integrations.sqlalchemy.model module

Reusable serialization mixin for projected SQLAlchemy models.

class edri.integrations.sqlalchemy.model.SQLAlchemyModelMixin

Bases: object

Serialize mapped columns and relationships using EDRI projections.

Declarative bases can inherit from this mixin before DeclarativeBase. __json__ supplies the serialized shape only on the non-projected path (bare ORM responses and SessionJSONEncoder); a model used solely through projections may leave it empty. Applications may override _json_extra() to add derived values or apply request-specific access rules.

to_json(columns: Sequence[str] | None = None, session: Any = None, skip_relationships: Collection[str] | None = None, *, projection: ProjectionNode | None = None, _seen_orm: set[tuple[type[Any], tuple[Any, ...]]] | None = None) → dict[str, Any]

Serialize only mapped state permitted by the active projection.

edri.integrations.sqlalchemy.path module

exception edri.integrations.sqlalchemy.path.SQLAlchemyPathError

Bases: ValueError

Raised when a public dotted path does not match mapped ORM attributes.

edri.integrations.sqlalchemy.path.canonical_orm_path(model: type[Any], public_path: str, *, require_scalar: bool = False) → tuple[str, ...]

Resolve a camelCase dotted path to strict mapped attribute names.

Every non-final segment must be a relationship. A column may only occur as the final segment. When require_scalar is true, a relationship may not be the final segment either.

edri.integrations.sqlalchemy.projection module

exception edri.integrations.sqlalchemy.projection.IncompleteProjectionError

Bases: RuntimeError

Raised when a response ORM graph does not match its resolved plan.

exception edri.integrations.sqlalchemy.projection.ProjectionError

Bases: ValueError

Raised when response projection metadata or a requested path is invalid.

class edri.integrations.sqlalchemy.projection.ProjectionFunctionSpec(function: 'str', path: 'str', output: 'str')

Bases: object

property request: str

class edri.integrations.sqlalchemy.projection.ProjectionNode(model: 'type[Any]', visible_columns: 'tuple[str, ...]', loaded_columns: 'tuple[str, ...]', relationships: "dict[str, 'ProjectionNode']", visible_relationships: 'tuple[str, ...]', recursive_relationships: 'tuple[str, ...]' = (), functions: 'tuple[ResolvedProjectionFunction, ...]' = ())

Bases: object

class edri.integrations.sqlalchemy.projection.ProjectionSpec(source: 'str', default: 'tuple[str, ...]', allowed: 'frozenset[str]', dependencies: 'tuple[str, ...]' = (), model: 'type[Any] | None' = None, recursive: 'tuple[str, ...]' = (), functions: 'tuple[ProjectionFunctionSpec, ...]' = ())

Bases: object

edri.integrations.sqlalchemy.projection.get_projection_spec(response: Response, field_name: str) → ProjectionSpec | None

Read projection metadata while isolating the current EDRI workaround.

edri.integrations.sqlalchemy.projection.orm_count(path: str, *, output: str | None = None) → ProjectionFunctionSpec

Declare an allowed COUNT projection over a direct relationship.

edri.integrations.sqlalchemy.projection.orm_projection(*, source: str, default: tuple[str, ...] = (), allowed: tuple[str, ...] = (), dependencies: tuple[str, ...] = (), model: type[Any] | None = None, recursive: tuple[str, ...] = (), functions: tuple[ProjectionFunctionSpec, ...] = ()) → dict[str, ProjectionSpec]

Return metadata for an ORM-backed EDRI response field.

edri.integrations.sqlalchemy.projection.projection(statement: sqlalchemy.sql.selectable.Select.typing.Any, *, event: Any, field: str) → sqlalchemy.sql.selectable.Select.typing.Any

Validate a response selection and apply its SQLAlchemy loader plan.

edri.integrations.sqlalchemy.projection.projection_all(session: sqlalchemy.orm.Session, statement: sqlalchemy.sql.selectable.Select.typing.Any, *, response: Response, field: str) → list[Any]

Execute a projected collection query and preserve computed values.

edri.integrations.sqlalchemy.projection.projection_one(session: sqlalchemy.orm.Session, statement: sqlalchemy.sql.selectable.Select.typing.Any, *, response: Response, field: str) → Any

Execute a projected single-row query and preserve computed values.

edri.integrations.sqlalchemy.projection.projection_rows(session: sqlalchemy.orm.Session, statement: sqlalchemy.sql.selectable.Select.typing.Any, *, response: Response, field: str) → list[sqlalchemy.engine.row.Row.typing.Any]

Execute a projected query and preserve its computed values.

edri.integrations.sqlalchemy.session module

Manager-local SQLAlchemy response finalization.

ORM values remain ORM values until the broker middleware.

This module keeps ORM sessions alive for one ManagerBase.resolve() call, completes projected response graphs while their original sessions are still available, validates the graphs, and closes the registered sessions before the event enters multiprocessing.

class edri.integrations.sqlalchemy.session.ORMResolutionScope

Bases: object

Keep ORM sessions alive for one manager resolution.

close_sessions() → None

Close all registered sessions in reverse registration order.

Session.close(session) deliberately bypasses ResponseAwareSession.close() so the close is performed immediately instead of being deferred again.

classmethod current() → ORMResolutionScope | None

Return the scope active in the current execution context.

register(session: sqlalchemy.orm.Session) → None

Register a session for deferred closing.

class edri.integrations.sqlalchemy.session.ResponseAwareSession(*args: Any, **kwargs: Any)

Bases: Session

Session behavior required by ORM events and projected responses.

close() → None

merge(instance: Any, *, load: bool = True, options: Any = None) → Any

class edri.integrations.sqlalchemy.session.SQLAlchemyManager(router_queue: Any | None = None, logger: Logger | None = None, from_time: datetime | None = None, max_workers: int = -1)

Bases: SQLAlchemyManagerMixin, ManagerBase

Ready-to-use synchronous manager with SQLAlchemy response support.

class edri.integrations.sqlalchemy.session.SQLAlchemyManagerMixin

Bases: object

Add SQLAlchemy response finalization to an EDRI manager.

Place the mixin before the concrete manager base so cooperative super() calls reach ManagerBase or one of its variants.

prepare_outgoing_event(event: Event) → None

resolve(event: Event) → None

resolve_non_router_event(event: Event, pipe: Connection) → None

edri.integrations.sqlalchemy.session.finalize_orm_response(event: Event) → None

Complete and validate projected ORM response graphs.

ORM instances remain ORM instances. Serialization to dictionaries happens later in the broker-side SQLAlchemy response middleware.

If a response field already has a resolved ProjectionNode, this function only validates that all requested state is available.

If the ProjectionNode is missing, each ORM root must still be attached to its original Session. Roots are grouped by Session and the projection query is executed once per Session, using populate_existing=True so the original response instances are populated rather than unrelated instances from another Session.

Any attached Session encountered here is registered with the active ORMResolutionScope and all registered Sessions are closed after successful finalization, detaching the response graph before multiprocessing.

edri.integrations.sqlalchemy.session.register_orm_session(session: sqlalchemy.orm.Session) → None

Register an ORM session for the current manager resolution.

edri.integrations.sqlalchemy.sort module

class edri.integrations.sqlalchemy.sort.ResolvedSort(root_model: type[Any], path: tuple[str, ...], order: Literal['asc', 'desc'])

Bases: object

A validated sort path; applying SQL joins is deliberately separate.

property column_name: str

property relationships: tuple[str, ...]

class edri.integrations.sqlalchemy.sort.SortBuilder

Bases: object

Resolve explicit client sort fields to strict mapped ORM paths.

The builder performs no relationship guessing, raw SQL construction, or implicit joins. A root field must be a scalar column. Every intermediate segment in a dotted field must be a relationship and its final segment must be a scalar column.

static build(table: Any | Sequence[Any], names: list[str], orders: list[str], logger: Any | None = None) → list[ResolvedSort]

edri.integrations.sqlalchemy.sort.apply_sorting(statement: sqlalchemy.sql.selectable.Select.typing.Any, sorts: Sequence[ResolvedSort]) → sqlalchemy.sql.selectable.Select.typing.Any

Explicitly add the JOINs and ORDER BY clauses required by sort plans.

edri.integrations.sqlalchemy.validator module

Build EDRI validators from SQLAlchemy mapped columns.

edri.integrations.sqlalchemy.validator.orm_validator(attribute: Any, /, **overrides: Any) → Callable[[Any], Any]

Return a validator callable derived from one ORM mapped column.

Pass the result to edri.utility.function.inject() in an event field:

name: inject(orm_validator(User.name))

The column's python_type selects the matching built-in EDRI validator. A bounded SQLAlchemy string also supplies maximum_length; a string-backed SQLAlchemyEnum("a", "b") supplies its allowed values. Types without a dedicated EDRI validator — including a column mapped to a Python enum.Enum, which is coerced against that class — use EDRI's normal scalar coercion.

overrides are forwarded to that built-in validator, taking precedence over anything derived from the column — orm_validator(User.name, maximum_length=3, minimum_length=2) tightens the column's own length. An override the underlying validator does not accept, or any override on a column with no dedicated validator (UUID, Decimal, bool, a Python enum), is rejected here rather than silently ignored.

Request optionality stays explicit at the event boundary; use inject(orm_validator(Model.field)) | None = None when None is valid. Database-backed rules such as uniqueness and foreign keys are intentionally outside this scalar validator.

Module contents

First-party SQLAlchemy integration for EDRI.

Install the optional dependency with pip install edri[sqlalchemy].