sandloom.sdk.effects¶

Effects of a unit: audit, commit hooks, the durable inbox, operation keys, sensitive effects.

Public API 1.0 (spec §2.2, Appendix C).

from sandloom.sdk.effects import <name>. Signatures in a terminal: sandloom docs <name>. Beginner names are the ones generated code and the examples use; any other name is advanced (mark its import line # sandloom: advanced <reason> in generated code and examples).

Start with (beginner)¶

Advanced¶

Beginner names¶

on_commit¶

async def on_commit(callback: OnCommitCallback, *, database: str | None = None) -> None

Run callback only after the current Unit of Work COMMITS.

For side effects that live OUTSIDE the transaction (a cache write such as revoke_user_tokens, a notification) and must not happen if the business change rolls back. callback takes no arguments and may be sync or async (use functools.partial / a closure to bind arguments).

• Inside a UoW (request DbSession, db_session(), a contract call): queued on that UoW's session and run right after its commit — never on rollback, and never when it was registered inside a begin_nested() savepoint that rolled back. database= disambiguates when several databases are active (same rule as publish_event).
• Outside any UoW (a script, a test that owns its own transaction): there is no commit to wait for, so it runs immediately.

Callbacks run after the outbox's post-commit dispatch was handed off (on the request path the events may still be on their way to a networked broker — never order a callback against event delivery), in registration order. A failing callback is logged and does NOT fail the already-committed request (the change is durable; make the effect retryable if it matters). A callback runs under the tenant that was bound when it was queued (the UoW's / request's tenant scope has already closed by then).

record_audit¶

def record_audit(
    action: str,
    *,
    actor_id: _uuid.UUID | str | None = None,
    actor_type: str = 'user',
    resource_type: str | None = None,
    resource_id: str | None = None,
    data: dict[str, Any] | None = None,
    required: bool | None = None,
    database: str | None = None,
) -> None

Append an entry to the audit log within the current Unit of Work.

Writes into the active request/task transaction, so the audit row commits iff the audited change commits (correct for auditing SUCCESSFUL actions). To audit a FAILED attempt that must survive a rollback, write it in a separate transaction opened after the request.

Modules call this by importing from sandloom.core.di — no ORM import needed.

Like publish_event, when an endpoint holds more than one database session at once pass database= to choose which database's audit_log the record lands in (so a billing action is audited in the billing database, not whichever session happened to be innermost). With a single active database it may be omitted; with several active and no choice the call raises rather than auditing in the wrong database.

required defaults to True (since 0.7): no UoW scope, a backend without an audit store, or a non-UUID actor_id string raise AuditUnavailableError instead of dropping the record. Best-effort is explicit — required=False skips (logged) in those cases; business modules need an # sandloom: best-effort <reason> marker on such a call (sandloom check rule best-effort-effects).

Advanced names¶

before_commit¶

async def before_commit(
    callback: OnCommitCallback,
    *,
    database: str | None = None,
) -> None

Run callback right BEFORE the current Unit of Work commits.

A last-moment check whose failure must abort the transaction: the callback (no arguments, sync or async) runs after the handler finished and before COMMIT; if it raises, the UoW ROLLS BACK and the exception propagates (a request answers with it — e.g. an HTTPException(401)). Used to re-validate an authorization input that lives in ANOTHER service, where no row lock can make the check atomic with the commit (ADR-0003 G2): the commit-time re-check (spec §4.5) re-reads the caller's subject-state record here, after the queued callbacks. It runs BEFORE the commit, not atomically with it: a change landing between the check and COMMIT is not seen (the check narrows a window to that gap; it does not close it).

• Inside a UoW (request DbSession, db_session(), a contract command): queued on its session, run in registration order before the commit; skipped when registered inside a begin_nested() savepoint that rolled back (like on_commit). database= disambiguates several active databases.
• Outside any UoW there is no commit to precede: it runs immediately.

Keep callbacks short and side-effect free (a read): they run while the transaction still holds its locks.

durable_inbox¶

def durable_inbox(consumer: str) -> Any

Return a DurableConsumerInbox for exactly-once consumer effects.

For money / orders / provisioning handlers: claim the event in the handler's own UoW session before applying the side effect:

inbox = durable_inbox("billing.credits@orders.paid.v1")
if not await inbox.claim(session, envelope["id"]):
    return  # duplicate delivery — effect already applied

Non-critical / idempotent handlers can keep using the cache-backed at-least-once dedupe instead.

mark_reauthenticated¶

def mark_reauthenticated() -> None

Record a successful re-authentication in this request (the caller just verified the current password, or a just-completed sign-in): it satisfies every step-up of the request (ADR-0009 item 8).

operation_key¶

def operation_key(*parts: object) -> str

A derived operation key: ":".join(parts) with checked parts.

The one rule for keys derived from business ids (ADR-0008 §3): every part is non-empty and contains neither : nor whitespace, so two different tuples can never produce the same key. Use it for any command a flow issues more than once for the same business fact:

operation_key("enroll", enrollment_id, "charge")  # "enroll:<id>:charge"

run_async¶

def run_async(coro: Coroutine[Any, Any, T]) -> T

Run coro to completion on this process's persistent event loop.

Use from synchronous code that has no running loop — a Celery task body, a script. Inside async code await the coroutine instead.

sensitive_effect¶

def sensitive_effect(risk: Risk, *, step_up: int | None = None) -> Callable[[F], F]

Declare the risk of an effect service, once, at its source (ADR-0009 items 7-8). Stack it UNDER @classmethod / @staticmethod.

step_up (seconds) defaults to 300 for security and to none for admin / money. See the module docstring for what the risk triggers.

Examples and related guides¶

• Deduplicate effects
• Python API overview