msflib.actions¶
Part of the msflib core package.
msflib.actions
¶
ModelAction
¶
Bases: Generic[ModelType, CreateSchemaType, UpdateSchemaType]
Generic reusable CRUD abstraction.
Supports both:
ModelAction[User, UserCreate, UserUpdate]()
and
class UserAction(ModelAction[User, UserCreate, UserUpdate]): ...
UserAction()
get_multi(session: Session, *, offset: int = 0, limit: int | None = 100) -> list[ModelType]
¶
Return multiple rows, optionally without a LIMIT clause.
Pass limit=None to fetch all matching rows from offset onward.
get_multi_by_all(session: Session, *, offset: int = 0, limit: int | None = 100, **filters: Any) -> list[ModelType]
¶
Return rows matching all filters, optionally without a LIMIT clause.
Pass limit=None to fetch all matching rows from offset onward.
get_multi_by_any(session: Session, *, offset: int = 0, limit: int | None = 100, **filters: Any) -> list[ModelType]
¶
Return rows matching any filter, optionally without a LIMIT clause.
Pass limit=None to fetch all matching rows from offset onward.
get_multi_by_expressions(session: Session, *exprs: BinaryExpression, offset: int = 0, limit: int | None = 100, order_by: Any | None = None) -> list[ModelType]
¶
Return rows matching expressions, optionally without a LIMIT clause.
Pass limit=None to fetch all matching rows from offset onward.
Pass order_by (e.g. a model column) for a deterministic row order
-- required for correct keyset pagination across repeated calls.
row_snapshot(row: Any) -> SimpleNamespace
¶
row's column values, readable after commit expires or detaches the ORM instance.
run_after_commit(session: SaSession, callback: Callable[[SaSession], None]) -> None
¶
Run callback(session) once session's outermost transaction commits.
On an idle session the callback waits for the next transaction it begins,
or is dropped if the session is closed first. It is also dropped if that
transaction, or the savepoint it was queued in, rolls back, and when it is
queued after the outermost commit (e.g. from another callback). Callbacks
run from after_commit: they must not emit SQL on session (use
session_after_commit), and their errors are logged rather than raised.
session_after_commit(committed: SaSession) -> Session
¶
A new session on committed's bind, for SQL from an after-commit callback.
action(model_type: type[ModelType], create_type: type[CreateSchemaType], update_type: type[UpdateSchemaType], random_override: Callable[..., CreateSchemaType] | None = None) -> ModelAction[ModelType, CreateSchemaType, UpdateSchemaType]
¶
Helper factory wrapper, that overrides the random method if provided, and caches instances by type tuple.
Returns:
| Type | Description |
|---|---|
ModelAction[ModelType, CreateSchemaType, UpdateSchemaType]
|
emit_after_commit(session: Session, event_name: str, instance: Any = None, payload: dict[str, Any] | None = None) -> None
¶
Queue emitter.emit_optional(event_name, instance, payload) for the outermost commit.
Use this whenever an event must not be observable before the data it
describes is actually durable — in particular when the caller may have
passed commit=False and owns an outer transaction that could still
roll back. Emitting synchronously in that case would let a listener
(e.g. a polling background worker) act on a row that isn't committed
yet, or that never lands at all.
Queued with run_after_commit, like ModelAction lifecycle events: a rollback
discards the queued event (a savepoint rollback only discards events
queued inside that savepoint), releasing a savepoint does not fire it,
and the outermost commit drains it exactly once via emit_optional
(best-effort: listener errors are logged, not raised).
on_model_write(model: type | str, callback: ModelWriteCallback, *, emitter: AppEmitter, key: str, operations: Iterable[ModelWriteOperation] = _OPERATIONS, on_error: Literal['log', 'raise'] = 'log', ignore: tuple[type[BaseException], ...] = (), force: bool = False) -> None
¶
Call callback(session, row, operation) on each pre-commit write event of model.
The callback runs in a savepoint, so a failure undoes its own writes.
on_error="log" then logs it and the app's write goes ahead;
"raise" aborts the write instead. Exceptions in ignore are
swallowed silently either way. Binding is
idempotent per (emitter, key) unless force=True.
reset_model_write_hooks(*, emitter: AppEmitter, key: str | None = None, key_prefix: str | None = None) -> None
¶
Remove the listeners bound under key, keys starting with key_prefix, or all.
after_commit
¶
Run work only when a session's outermost transaction commits: run_after_commit.
SQLAlchemy fires the session after_commit event when a SAVEPOINT
(session.begin_nested()) is released, not only on the real commit, and
after_rollback when a savepoint rolls back. Hooks that treat either
event as "the transaction is over" therefore run too early (work is exposed
before the outer commit, which may still roll back) or discard too much (a
savepoint rollback drops work queued outside that savepoint).
run_after_commit remembers the transaction each callback was queued in:
- a released savepoint keeps its callbacks pending;
- a rolled-back savepoint drops only the callbacks queued inside it;
- the outermost commit runs every remaining callback;
- the outermost rollback (or close without commit) drops them all.
Callbacks queued on an idle session wait for the next transaction it begins, unless the session is closed first.
OuterCommitQueue(key: str, on_commit: Callable[[SaSession, list[Any]], None])
¶
on_commit(session, items) runs from the outermost after_commit.
It inherits that event's restrictions: no SQL may be emitted on the
session, and nobody is left to catch its errors, so it should log
rather than raise. Items queued from inside on_commit belong to
the transaction that just finished and are discarded with it, rather
than leaking into whatever unrelated transaction commits next.
add(session: SaSession, item: Any) -> None
¶
Queue item for the outermost commit of session's current transaction.
On an idle session the item waits for the next transaction it begins, unless the session is closed first.
run_after_commit(session: SaSession, callback: Callable[[SaSession], None]) -> None
¶
Run callback(session) once session's outermost transaction commits.
On an idle session the callback waits for the next transaction it begins,
or is dropped if the session is closed first. It is also dropped if that
transaction, or the savepoint it was queued in, rolls back, and when it is
queued after the outermost commit (e.g. from another callback). Callbacks
run from after_commit: they must not emit SQL on session (use
session_after_commit), and their errors are logged rather than raised.
session_after_commit(committed: SaSession) -> Session
¶
A new session on committed's bind, for SQL from an after-commit callback.
row_snapshot(row: Any) -> SimpleNamespace
¶
row's column values, readable after commit expires or detaches the ORM instance.
base
¶
ModelAction
¶
Bases: Generic[ModelType, CreateSchemaType, UpdateSchemaType]
Generic reusable CRUD abstraction.
Supports both:
ModelAction[User, UserCreate, UserUpdate]()
and
class UserAction(ModelAction[User, UserCreate, UserUpdate]): ...
UserAction()
get_multi(session: Session, *, offset: int = 0, limit: int | None = 100) -> list[ModelType]
¶
Return multiple rows, optionally without a LIMIT clause.
Pass limit=None to fetch all matching rows from offset onward.
get_multi_by_all(session: Session, *, offset: int = 0, limit: int | None = 100, **filters: Any) -> list[ModelType]
¶
Return rows matching all filters, optionally without a LIMIT clause.
Pass limit=None to fetch all matching rows from offset onward.
get_multi_by_any(session: Session, *, offset: int = 0, limit: int | None = 100, **filters: Any) -> list[ModelType]
¶
Return rows matching any filter, optionally without a LIMIT clause.
Pass limit=None to fetch all matching rows from offset onward.
get_multi_by_expressions(session: Session, *exprs: BinaryExpression, offset: int = 0, limit: int | None = 100, order_by: Any | None = None) -> list[ModelType]
¶
Return rows matching expressions, optionally without a LIMIT clause.
Pass limit=None to fetch all matching rows from offset onward.
Pass order_by (e.g. a model column) for a deterministic row order
-- required for correct keyset pagination across repeated calls.
emit_after_commit(session: Session, event_name: str, instance: Any = None, payload: dict[str, Any] | None = None) -> None
¶
Queue emitter.emit_optional(event_name, instance, payload) for the outermost commit.
Use this whenever an event must not be observable before the data it
describes is actually durable — in particular when the caller may have
passed commit=False and owns an outer transaction that could still
roll back. Emitting synchronously in that case would let a listener
(e.g. a polling background worker) act on a row that isn't committed
yet, or that never lands at all.
Queued with run_after_commit, like ModelAction lifecycle events: a rollback
discards the queued event (a savepoint rollback only discards events
queued inside that savepoint), releasing a savepoint does not fire it,
and the outermost commit drains it exactly once via emit_optional
(best-effort: listener errors are logged, not raised).
action(model_type: type[ModelType], create_type: type[CreateSchemaType], update_type: type[UpdateSchemaType], random_override: Callable[..., CreateSchemaType] | None = None) -> ModelAction[ModelType, CreateSchemaType, UpdateSchemaType]
¶
Helper factory wrapper, that overrides the random method if provided, and caches instances by type tuple.
Returns:
| Type | Description |
|---|---|
ModelAction[ModelType, CreateSchemaType, UpdateSchemaType]
|
hooks
¶
Run a callback inside the transaction of every ModelAction write to a model.
Listens to <model>-{create,update,delete}-pre-commit, which fire after
flush and before commit, so work the callback writes commits atomically
with the row. Writes that bypass ModelAction fire no events.
on_model_write(model: type | str, callback: ModelWriteCallback, *, emitter: AppEmitter, key: str, operations: Iterable[ModelWriteOperation] = _OPERATIONS, on_error: Literal['log', 'raise'] = 'log', ignore: tuple[type[BaseException], ...] = (), force: bool = False) -> None
¶
Call callback(session, row, operation) on each pre-commit write event of model.
The callback runs in a savepoint, so a failure undoes its own writes.
on_error="log" then logs it and the app's write goes ahead;
"raise" aborts the write instead. Exceptions in ignore are
swallowed silently either way. Binding is
idempotent per (emitter, key) unless force=True.
reset_model_write_hooks(*, emitter: AppEmitter, key: str | None = None, key_prefix: str | None = None) -> None
¶
Remove the listeners bound under key, keys starting with key_prefix, or all.