API reference

Generated from the source docstrings.

Note

Only the public surface is documented here. jasil._core is internal — see API stability.

Providers

jasil.providers

Platform providers — the tiny interfaces domain code depends on.

Pure module: only stdlib typing and the event envelope. No infrastructure (redis / boto3 / sqlalchemy) and no domain imports, so any module can depend on these providers without pulling in a backend. Concrete backends live in jasil.backends and are selected by the composition root (jasil.container.build_platform).

ClockProvider

Bases: Protocol

Injectable time source for testability.

Source code in jasil/providers.py

@runtime_checkable
class ClockProvider(Protocol):
    """Injectable time source for testability."""

    def now(self) -> datetime: ...
    def monotonic(self) -> float: ...

EventBusProvider

Bases: Protocol

Publish/subscribe — synchronous in-process, or durable via Redis Streams.

Source code in jasil/providers.py

@runtime_checkable
class EventBusProvider(Protocol):
    """Publish/subscribe — synchronous in-process, or durable via Redis Streams."""

    def publish(self, event: Event) -> None: ...
    def subscribe(self, event_type: str, handler: Callable[[Event], None]) -> None: ...
    def start(self) -> None: ...
    def stop(self) -> None: ...

EventRecorder

Bases: Protocol

Records each event's lifecycle to durable storage for observability.

Injected into the event bus by the composition root when event logging is enabled. Recording must never disrupt event processing, so implementations swallow and log their own storage errors. :meth:record_published and :meth:record_queued insert the initial row for a bus-delivered or durable-delivered event respectively. :meth:track wraps handler execution — it marks the event processing on entry (unless record_processing is False, for a synchronous single-process dispatch where the intermediate state is never observed) and completed / failed on exit, re-raising any handler exception so the bus keeps its own error semantics (propagate in-process, leave pending on Redis Streams).

Source code in jasil/providers.py

@runtime_checkable
class EventRecorder(Protocol):
    """Records each event's lifecycle to durable storage for observability.

    Injected into the event bus by the composition root when event logging is
    enabled. Recording must never disrupt event processing, so implementations
    swallow and log their own storage errors. :meth:`record_published` and
    :meth:`record_queued` insert the initial row for a bus-delivered or
    durable-delivered event respectively. :meth:`track` wraps handler
    execution — it marks the event *processing* on entry (unless
    ``record_processing`` is False, for a synchronous single-process dispatch
    where the intermediate state is never observed) and *completed* / *failed* on
    exit, re-raising any handler exception so the bus keeps its own error
    semantics (propagate in-process, leave pending on Redis Streams).
    """

    def record_published(self, event: Event) -> None: ...
    def record_queued(self, event: Event) -> None: ...
    def track(
        self,
        event: Event,
        *,
        worker_id: str,
        handler_name: str | None,
        record_processing: bool = True,
    ) -> AbstractContextManager[None]: ...

GeocodedPlace dataclass

A place resolved from coordinates. Any field may be None.

Attributes:

Name	Type	Description
city	str | None	Populated place name, when the provider resolved one.
town	str | None	Sub-locality (district/town), when the provider resolved one.
country	str | None	Country name, when the provider resolved one.

Source code in jasil/providers.py

@dataclass(frozen=True)
class GeocodedPlace:
    """A place resolved from coordinates. Any field may be ``None``.

    Attributes:
        city: Populated place name, when the provider resolved one.
        town: Sub-locality (district/town), when the provider resolved one.
        country: Country name, when the provider resolved one.
    """

    city: str | None = None
    town: str | None = None
    country: str | None = None

GeocodingProvider

Bases: Protocol

Reverse-geocoding — turn a coordinate into a place name.

The one seam through which domain code performs reverse geocoding, so a caller never knows which upstream service is configured, nor that a network call happens at all. That matters more here than for the other providers: this is the platform's only outbound call to a third party, so the egress hardening it needs (host validation, address denylist, no redirects, rate limiting) lives behind this interface instead of in a domain module.

Implementations must not raise for an upstream failure — geocoding is best-effort enrichment, and a provider outage must never fail the import or backfill that triggered it. Return None when nothing resolves.

Source code in jasil/providers.py

@runtime_checkable
class GeocodingProvider(Protocol):
    """Reverse-geocoding — turn a coordinate into a place name.

    The one seam through which domain code performs reverse geocoding, so a
    caller never knows which upstream service is configured, nor that a network
    call happens at all. That matters more here than for the other providers:
    this is the platform's only *outbound* call to a third party, so the
    egress hardening it needs (host validation, address denylist, no redirects,
    rate limiting) lives behind this interface instead of in a domain module.

    Implementations must not raise for an upstream failure — geocoding is
    best-effort enrichment, and a provider outage must never fail the import or
    backfill that triggered it. Return ``None`` when nothing resolves.
    """

    def reverse(self, latitude: float, longitude: float) -> GeocodedPlace | None: ...

LockProvider

Bases: Protocol

Best-effort mutual exclusion for scheduled/backfill work.

Source code in jasil/providers.py

@runtime_checkable
class LockProvider(Protocol):
    """Best-effort mutual exclusion for scheduled/backfill work."""

    def try_acquire(self, name: str, ttl_seconds: int | None = None) -> AbstractContextManager[bool]: ...

StateBackendUnavailableError

Bases: RuntimeError

Raised by a :class:StateProvider when its backing store is unreachable.

Lets domain stores react to an infrastructure outage (e.g. surface a 503 or swallow a best-effort cleanup) without knowing or importing anything about the concrete backend (Redis). The in-memory backend never raises it.

Source code in jasil/providers.py

class StateBackendUnavailableError(RuntimeError):
    """Raised by a :class:`StateProvider` when its backing store is unreachable.

    Lets domain stores react to an infrastructure outage (e.g. surface a 503 or
    swallow a best-effort cleanup) without knowing or importing anything about
    the concrete backend (Redis). The in-memory backend never raises it.
    """

StateProvider

Bases: Protocol

Ephemeral keyed state (counters, TTL flags, small blobs).

The single seam through which domain code reads and writes short-lived shared state, so a store never needs to know whether it is backed by a process-local dict (local) or Redis (distributed). Beyond plain key/value access it exposes the few atomic primitives that login-throttling and single-use-token stores need (set_if_absent, get_and_delete, record_tiered_failure) so their correctness does not depend on the backend.

Source code in jasil/providers.py

@runtime_checkable
class StateProvider(Protocol):
    """Ephemeral keyed state (counters, TTL flags, small blobs).

    The single seam through which domain code reads and writes short-lived
    shared state, so a store never needs to know whether it is backed by a
    process-local dict (``local``) or Redis (``distributed``). Beyond plain
    key/value access it exposes the few *atomic* primitives that login-throttling
    and single-use-token stores need (``set_if_absent``, ``get_and_delete``,
    ``record_tiered_failure``) so their correctness does not depend on the
    backend.
    """

    def get(self, key: str) -> bytes | None: ...
    def set(self, key: str, value: bytes, ttl_seconds: int | None = None) -> None: ...
    def set_if_absent(self, key: str, value: bytes, ttl_seconds: int | None = None) -> bool: ...
    def get_and_delete(self, key: str) -> bytes | None: ...
    def delete(self, key: str) -> None: ...
    def delete_prefix(self, prefix: str) -> int: ...
    def iter_keys(self, prefix: str) -> Iterator[str]: ...
    def incr(self, key: str, amount: int = 1, ttl_seconds: int | None = None) -> int: ...
    def record_tiered_failure(
        self,
        counter_key: str,
        gate_key: str,
        tiers: tuple[tuple[int, int], ...],
        counter_ttl_seconds: int,
    ) -> TieredFailureOutcome: ...

StorageProvider

Bases: Protocol

Opaque byte-blob storage addressed by a key within a named area.

An area is a domain-owned namespace (e.g. "avatars", "exports") so one backend serves every subsystem: locally it maps to a subdirectory, on S3 to a key prefix. The database stores only the key (the area is a fixed constant of the calling domain); url is computed at serialization time so migrating local -> S3 needs no data migration.

Most subsystems only ever write a blob and later serve it via url, but some need the bytes back in-process (e.g. bundling stored files into an export); get is that read path, returning None when the blob is absent.

list_keys exists for the subsystems whose keys are not derivable from a domain id — for instance when a key carries a random component. Without a prefix listing, such a subsystem would have to reach past the provider to the filesystem to clean up after a deleted record, which is exactly what this provider exists to prevent.

Source code in jasil/providers.py

@runtime_checkable
class StorageProvider(Protocol):
    """Opaque byte-blob storage addressed by a key within a named *area*.

    An area is a domain-owned namespace (e.g. ``"avatars"``, ``"exports"``) so
    one backend serves every subsystem: locally it maps to a subdirectory, on S3
    to a key prefix. The database stores only the key (the area is a fixed
    constant of the calling domain); ``url`` is computed at serialization time so
    migrating local -> S3 needs no data migration.

    Most subsystems only ever write a blob and later serve it via ``url``, but
    some need the bytes back in-process (e.g. bundling stored files into an
    export); ``get`` is that read path, returning ``None`` when the blob is
    absent.

    ``list_keys`` exists for the subsystems whose keys are *not* derivable from a
    domain id — for instance when a key carries a random component. Without a
    prefix listing, such a subsystem would have to reach past the provider to the
    filesystem to clean up after a deleted record, which is exactly what this
    provider exists to prevent.
    """

    def save(self, area: str, key: str, data: bytes, content_type: str | None = None) -> str: ...
    def get(self, area: str, key: str) -> bytes | None: ...
    def exists(self, area: str, key: str) -> bool: ...
    def delete(self, area: str, key: str) -> None: ...
    def list_keys(self, area: str, prefix: str = "") -> list[str]: ...
    def url(self, area: str, key: str, expires_in: int = 3600) -> str: ...

TieredFailureOutcome dataclass

Result of an atomic tiered-lockout increment.

Attributes:

Name	Type	Description
count	int	The failure counter value after this attempt.
locked_until_epoch	int | None	Wall-clock epoch (seconds) the lock is active until, or None when not locked.
newly_locked	bool	True only when this call created (or renewed) the lock.

Source code in jasil/providers.py

@dataclass(frozen=True)
class TieredFailureOutcome:
    """Result of an atomic tiered-lockout increment.

    Attributes:
        count: The failure counter value after this attempt.
        locked_until_epoch: Wall-clock epoch (seconds) the lock is active until,
            or ``None`` when not locked.
        newly_locked: True only when *this* call created (or renewed) the lock.
    """

    count: int
    locked_until_epoch: int | None
    newly_locked: bool

Events

jasil.events

The event envelope, new_event helper, and standard metadata keys.

Pure module — the single structured shape every event travels in, so the pipeline can route, trace, correlate, dedup, and retry without knowing the domain.

Channel names (event_type values) are owned by the domain that publishes them, not defined here — e.g. the orders module owns order.created. Keeping them out of the substrate stops this generic layer from accumulating domain knowledge; a producer and its subscribers import the same domain-side constant so they cannot drift on the string. Convention: <domain>.<fact> in past tense. event_type stays a plain str on the envelope so the bus is open to new events with no edits here.

The envelope is defined ahead of the bus so the wire format never has to change once the first producer ships.

Payload versioning

schema_version describes the shape of payload for its event_type. It lives on the envelope rather than inside each payload dict so the substrate can carry it without parsing domain data, and so no producer can forget it.

It exists because the code that writes a payload and the code that reads it are not guaranteed to be the same version: a durable event is staged in the outbox, relayed on a schedule, retried with backoff, and may sit dead-lettered indefinitely — and during a rolling deploy old and new replicas run at once. A consumer that silently ignores unknown keys (every payload model sets extra="ignore") would read a renamed or repurposed field as its default and do the wrong thing quietly. The version turns that into something a consumer can detect: see :mod:jasil.event_versioning.

The number is owned by the publishing domain, like the channel name. Bump it in the domain's publisher when the payload's shape or a field's meaning changes; purely additive optional fields do not need a bump.

Event dataclass

Envelope wrapping every event in the system.

Attributes:

Name	Type	Description
event_id	str	UUIDv4 identifying this event instance; stable across retries.
event_type	str	Dot-notation channel, e.g. order.created.
source	str	Where the event originated, e.g. api:create_order.
timestamp	str	ISO-8601 UTC timestamp of the first publish (not the retry).
payload	dict	Domain data, homogeneous per event_type.
metadata	dict	Correlation context (request_id, plus any host-defined keys).
retry_count	int	Processing attempts so far; 0 on first publish.
schema_version	int	Which version of payload's shape this event carries, owned by the publishing domain. Defaults to :data:INITIAL_SCHEMA_VERSION so existing producers are unchanged.

Source code in jasil/events.py

@dataclass(frozen=True)
class Event:
    """Envelope wrapping every event in the system.

    Attributes:
        event_id: UUIDv4 identifying this event instance; stable across retries.
        event_type: Dot-notation channel, e.g. ``order.created``.
        source: Where the event originated, e.g. ``api:create_order``.
        timestamp: ISO-8601 UTC timestamp of the first publish (not the retry).
        payload: Domain data, homogeneous per ``event_type``.
        metadata: Correlation context (request_id, plus any host-defined keys).
        retry_count: Processing attempts so far; 0 on first publish.
        schema_version: Which version of ``payload``'s shape this event carries,
            owned by the publishing domain. Defaults to
            :data:`INITIAL_SCHEMA_VERSION` so existing producers are unchanged.
    """

    event_id: str
    event_type: str
    source: str
    timestamp: str
    payload: dict
    metadata: dict = field(default_factory=dict)
    retry_count: int = 0
    schema_version: int = INITIAL_SCHEMA_VERSION

new_event

new_event(
    event_type,
    payload,
    *,
    source,
    metadata=None,
    event_id=None,
    retry_count=0,
    schema_version=INITIAL_SCHEMA_VERSION,
)

Mint an :class:Event, generating event_id and timestamp.

Parameters:

Name	Type	Description	Default
event_type	str	The channel/type, e.g. order.created.	required
payload	dict	Domain data for the event.	required
source	str	Origin label, e.g. api:create_order.	required
metadata	dict | None	Optional correlation context.	None
event_id	str | None	Optional explicit id (defaults to a fresh UUIDv4); reuse the original id when re-publishing a retry so tracing stays stable.	None
retry_count	int	Attempt counter (incremented on re-publish).	0
schema_version	int	The payload-shape version this producer writes.	INITIAL_SCHEMA_VERSION

Returns:

Type	Description
Event	A frozen :class:Event with a fresh id and UTC timestamp.

Raises:

Type	Description
ValueError	When event_id, event_type or source is longer than the column it is persisted in.

Source code in jasil/events.py

def new_event(
    event_type: str,
    payload: dict,
    *,
    source: str,
    metadata: dict | None = None,
    event_id: str | None = None,
    retry_count: int = 0,
    schema_version: int = INITIAL_SCHEMA_VERSION,
) -> Event:
    """Mint an :class:`Event`, generating ``event_id`` and ``timestamp``.

    Args:
        event_type: The channel/type, e.g. ``order.created``.
        payload: Domain data for the event.
        source: Origin label, e.g. ``api:create_order``.
        metadata: Optional correlation context.
        event_id: Optional explicit id (defaults to a fresh UUIDv4); reuse the
            original id when re-publishing a retry so tracing stays stable.
        retry_count: Attempt counter (incremented on re-publish).
        schema_version: The payload-shape version this producer writes.

    Returns:
        A frozen :class:`Event` with a fresh id and UTC timestamp.

    Raises:
        ValueError: When ``event_id``, ``event_type`` or ``source`` is longer
            than the column it is persisted in.
    """
    resolved_event_id = event_id or str(uuid.uuid4())
    check_length(resolved_event_id, field="event_id", limit=MAX_EVENT_ID_LENGTH)
    check_length(event_type, field="event_type", limit=MAX_EVENT_TYPE_LENGTH)
    check_length(source, field="source", limit=MAX_SOURCE_LENGTH)
    return Event(
        event_id=resolved_event_id,
        event_type=event_type,
        source=source,
        timestamp=datetime.now(UTC).isoformat(),
        payload=payload,
        metadata=metadata if metadata is not None else {},
        retry_count=retry_count,
        schema_version=schema_version,
    )

Payload versioning

jasil.event_versioning

Version-aware payload parsing for durable and bus subscribers.

A subscriber never calls Model.model_validate(event.payload) directly; it goes through :func:parse_payload, which compares the envelope's schema_version against the version the local code understands and does the right thing for each direction of skew.

Why the direction matters

The writer and the reader of an event are frequently not the same build: an event is staged in the outbox, relayed on a schedule, retried with exponential backoff, and can sit dead-lettered indefinitely — and during a rolling deploy old and new replicas serve traffic simultaneously.

• Older event, newer consumer (the common case — the outbox backlog after a deploy). This must actually work, so the payload model registers an explicit upgrader per version step. Silently defaulting a missing field is exactly the failure mode this module exists to prevent: every payload model sets extra="ignore", so a renamed or repurposed field would otherwise be dropped and read as its default, producing a wrong result with no error anywhere.
• Newer event, older consumer (a new replica publishes before the old workers drain). Nothing sensible can be done — the local code has never seen this shape. :class:UnsupportedEventVersionError is raised so the existing job machinery retries with backoff; the condition is self-healing, because by the time the attempts are spent the stale worker is normally gone. Failing loudly and retrying is strictly better than guessing.

No new machinery is needed for either: durable handlers already raise on failure and the runner already retries and then dead-letters.

UnsupportedEventVersionError

Bases: Exception

Raised when an event's payload version is newer than this build understands.

Deliberately an ordinary exception rather than a domain error: it is never seen by an HTTP caller, only by the durable-job runner (which retries, then dead-letters) or the best-effort bus wrapper (which logs and swallows).

Source code in jasil/event_versioning.py

class UnsupportedEventVersionError(Exception):
    """Raised when an event's payload version is newer than this build understands.

    Deliberately an ordinary exception rather than a domain error: it is never
    seen by an HTTP caller, only by the durable-job runner (which retries, then
    dead-letters) or the best-effort bus wrapper (which logs and swallows).
    """

VersionedPayload

Bases: BaseModel

Base for every event payload model, carrying its own schema version.

Subclasses set :attr:SCHEMA_VERSION and, once they have evolved past version 1, register an upgrader per version step in :attr:UPGRADERS.

Both are ClassVar — without that annotation Pydantic would treat them as model fields, so they would be expected in every payload dict and would not be readable off the class at all.

Attributes:

Name	Type	Description
SCHEMA_VERSION	int	The payload shape this build reads and writes.
UPGRADERS	dict[int, Callable[[dict], dict]]	{from_version: fn(payload_dict) -> payload_dict}, each entry converting a payload one version forward. Applied in sequence, so evolving 1 -> 3 only needs a 1->2 and a 2->3 entry.

Source code in jasil/event_versioning.py

class VersionedPayload(BaseModel):
    """Base for every event payload model, carrying its own schema version.

    Subclasses set :attr:`SCHEMA_VERSION` and, once they have evolved past
    version 1, register an upgrader per version step in :attr:`UPGRADERS`.

    Both are ``ClassVar`` — without that annotation Pydantic would treat them as
    *model fields*, so they would be expected in every payload dict and would not
    be readable off the class at all.

    Attributes:
        SCHEMA_VERSION: The payload shape this build reads and writes.
        UPGRADERS: ``{from_version: fn(payload_dict) -> payload_dict}``, each
            entry converting a payload one version forward. Applied in sequence,
            so evolving 1 -> 3 only needs a 1->2 and a 2->3 entry.
    """

    SCHEMA_VERSION: ClassVar[int] = 1
    UPGRADERS: ClassVar[dict[int, Callable[[dict], dict]]] = {}

parse_payload

parse_payload(model, event)

Validate an event's payload, upgrading it from an older version if needed.

Parameters:

Name	Type	Description	Default
model	type[T]	The payload model this subscriber consumes.	required
event	Event	The event envelope, carrying schema_version and payload.	required

Returns:

Type	Description
T	The validated payload at the local build's schema version.

Raises:

Type	Description
UnsupportedEventVersionError	When the event was written by a newer build than this one.
ValidationError	When the payload does not match its declared version.

Source code in jasil/event_versioning.py

def parse_payload[T: VersionedPayload](model: type[T], event: Event) -> T:
    """Validate an event's payload, upgrading it from an older version if needed.

    Args:
        model: The payload model this subscriber consumes.
        event: The event envelope, carrying ``schema_version`` and ``payload``.

    Returns:
        The validated payload at the local build's schema version.

    Raises:
        UnsupportedEventVersionError: When the event was written by a newer build
            than this one.
        ValidationError: When the payload does not match its declared version.
    """
    target = model.SCHEMA_VERSION
    version = event.schema_version

    if version > target:
        logger.error(
            "Refusing an event written by a newer build",
            extra={
                "event_type": event.event_type,
                "event_id": event.event_id,
                "event_version": version,
                "supported_version": target,
            },
        )
        raise UnsupportedEventVersionError(
            f"{event.event_type} payload is version {version}; this build understands {target}"
        )

    payload = event.payload
    if version < target:
        logger.info(
            "Upgrading an event payload written by an older build",
            extra={
                "event_type": event.event_type,
                "event_id": event.event_id,
                "event_version": version,
                "supported_version": target,
            },
        )
        payload = _upgrade(model, payload, version, target, event)

    return model.model_validate(payload)

Publishing

jasil.publisher

The single publish seam every producer goes through.

One tiny facade so no producer ever assembles an :class:~jasil.events.Event or touches the active platform directly. Centralising publishing here means the transactional outbox is a change to this function alone, not to every call site.

It resolves the active platform, stamps the ambient request id for correlation, and mints the envelope. Delivery then takes one of two routes:

• Durable (outbox): when durable jobs are enabled, the caller supplies its DB session, and the event type has registered durable subscribers, the event is written to the event_outbox — the relay later fans it out into retryable per-subscriber jobs. The event is also recorded queued in event_log so the observability dashboard reflects durable events too (execution detail then lives in the Jobs dashboard).
• Best-effort (bus): otherwise the event is dispatched through the event bus (inline in local, via Redis Streams in distributed), which records the full lifecycle itself.

Delivery guarantee. The producer's domain row is the source of truth; this publish is best-effort from the producer's perspective — failures are logged and swallowed so publishing never breaks the producer's own work. The outbox is not committed in the same transaction as the domain change (the ingestion path commits per-CRUD), so a crash between the domain commit and the outbox write can drop an event. Every subscriber must therefore have a reconciliation net — a backfill or sweeper that re-derives missed work). A future unit-of-work refactor can upgrade this to a genuinely atomic outbox; until then, "durable" means retryable once written, not never lost. Channel names and payload shape stay owned by the publishing domain; this layer only knows the generic envelope.

publish

publish(
    event_type,
    payload,
    *,
    source,
    metadata=None,
    db=None,
    schema_version=INITIAL_SCHEMA_VERSION,
)

Publish a domain event through the active platform, best-effort.

Parameters:

Name	Type	Description	Default
event_type	str	The domain-owned channel, e.g. order.created.	required
payload	dict	Domain data for the event (homogeneous per event_type).	required
source	str	Origin label, e.g. api:create_order.	required
metadata	dict | None	Optional correlation context; merged with the ambient request id when one is set on the current request.	None
db	Any	The producer's SQLAlchemy session. When provided and durable jobs are enabled for this event type, the event is written to the outbox using this session (durable delivery); otherwise it is ignored.	None

Returns:

Type	Description
None	None. Delivery failures are logged and swallowed so a publish never
None	breaks the producer. The domain row remains the source of truth and each
None	subscriber's reconciliation net recovers anything missed.

Source code in jasil/publisher.py

def publish(
    event_type: str,
    payload: dict,
    *,
    source: str,
    metadata: dict | None = None,
    db: Any = None,
    schema_version: int = INITIAL_SCHEMA_VERSION,
) -> None:
    """Publish a domain event through the active platform, best-effort.

    Args:
        event_type: The domain-owned channel, e.g. ``order.created``.
        payload: Domain data for the event (homogeneous per ``event_type``).
        source: Origin label, e.g. ``api:create_order``.
        metadata: Optional correlation context; merged with the ambient request
            id when one is set on the current request.
        db: The producer's SQLAlchemy session. When provided and durable jobs are
            enabled for this event type, the event is written to the outbox using
            this session (durable delivery); otherwise it is ignored.

    Returns:
        None. Delivery failures are logged and swallowed so a publish never
        breaks the producer. The domain row remains the source of truth and each
        subscriber's reconciliation net recovers anything missed.
    """
    try:
        platform = platform_runtime.get_active_platform()
        event = _mint(event_type, payload, source, metadata, schema_version)
        if db is not None and _durable_delivery_enabled(event_type):
            _stage_in_outbox(platform.recorder, event, db=db, now=platform.clock.now(), commit=True)
        else:
            platform.events.publish(event)
    except Exception as err:
        logger.error(f"Failed to publish event {event_type}: {err}", exc_info=err)

publish_committing

publish_committing(
    event_type,
    payload,
    *,
    source,
    metadata=None,
    db,
    commit,
    schema_version=INITIAL_SCHEMA_VERSION,
)

Publish a domain event atomically around the caller's domain commit.

Unlike :func:publish (which the caller invokes after it has already committed its own change), this variant owns the commit ordering so durable delivery can be made atomic with the domain write. commit is a zero-arg callable that commits the caller's unit of work.

• Durable delivery enabled (durable jobs on + a durable subscriber for event_type): the outbox row is staged on db without committing, then commit() flushes the domain rows and the outbox row in one transaction — so the event can never be lost relative to the change that produced it. A staging failure propagates (the caller's transaction is left uncommitted for rollback), so the whole unit of work is all-or-nothing.
• Otherwise (best-effort bus path): commit() runs first so the domain row — the source of truth — is durable regardless, then the event is dispatched on the bus and any dispatch failure is logged and swallowed.

Parameters:

Name	Type	Description	Default
event_type	str	The domain-owned channel, e.g. order.created.	required
payload	dict	Domain data for the event.	required
source	str	Origin label, e.g. api:create_order.	required
metadata	dict | None	Optional correlation context; merged with the ambient request id.	None
db	Any	The producer's SQLAlchemy session (holds the uncommitted domain change).	required
commit	Callable[[], None]	Zero-arg callable that commits the caller's unit of work.	required

Returns:

Type	Description
None	None.

Source code in jasil/publisher.py

def publish_committing(
    event_type: str,
    payload: dict,
    *,
    source: str,
    metadata: dict | None = None,
    db: Any,
    commit: Callable[[], None],
    schema_version: int = INITIAL_SCHEMA_VERSION,
) -> None:
    """Publish a domain event atomically around the caller's domain commit.

    Unlike :func:`publish` (which the caller invokes *after* it has already
    committed its own change), this variant owns the commit ordering so durable
    delivery can be made atomic with the domain write. ``commit`` is a zero-arg
    callable that commits the caller's unit of work.

    * **Durable delivery enabled** (durable jobs on + a durable subscriber for
      ``event_type``): the outbox row is staged on ``db`` **without** committing,
      then ``commit()`` flushes the domain rows and the outbox row in one
      transaction — so the event can never be lost relative to the change that
      produced it. A staging failure propagates (the caller's transaction is left
      uncommitted for rollback), so the whole unit of work is all-or-nothing.
    * **Otherwise** (best-effort bus path): ``commit()`` runs first so the domain
      row — the source of truth — is durable regardless, then the event is
      dispatched on the bus and any dispatch failure is logged and swallowed.

    Args:
        event_type: The domain-owned channel, e.g. ``order.created``.
        payload: Domain data for the event.
        source: Origin label, e.g. ``api:create_order``.
        metadata: Optional correlation context; merged with the ambient request id.
        db: The producer's SQLAlchemy session (holds the uncommitted domain change).
        commit: Zero-arg callable that commits the caller's unit of work.

    Returns:
        None.
    """
    publish_many_committing(
        event_type,
        [payload],
        source=source,
        metadata_for=lambda _payload: metadata,
        db=db,
        commit=commit,
        schema_version=schema_version,
    )

publish_many_committing

publish_many_committing(
    event_type,
    payloads,
    *,
    source,
    metadata_for=None,
    db,
    commit,
    schema_version=INITIAL_SCHEMA_VERSION,
)

Publish many same-type events atomically around one domain commit.

The batch counterpart of :func:publish_committing, for producers that remove or create many rows in a single unit of work (bulk deletes) and must emit one event per affected row without committing once per event. All outbox rows are staged on db uncommitted and land in the caller's transaction, so the domain change and every event commit together or not at all.

Parameters:

Name	Type	Description	Default
event_type	str	The domain-owned channel shared by every event in the batch.	required
payloads	Sequence[dict]	One payload per event. An empty sequence still runs commit so the caller's unit of work is committed exactly once either way.	required
source	str	Origin label, e.g. api:bulk_delete.	required
metadata_for	Callable[[dict], dict | None] | None	Optional per-payload correlation metadata builder.	None
db	Any	The producer's SQLAlchemy session (holds the uncommitted change).	required
commit	Callable[[], None]	Zero-arg callable that commits the caller's unit of work.	required

Returns:

Type	Description
None	None.

Source code in jasil/publisher.py

def publish_many_committing(
    event_type: str,
    payloads: Sequence[dict],
    *,
    source: str,
    metadata_for: Callable[[dict], dict | None] | None = None,
    db: Any,
    commit: Callable[[], None],
    schema_version: int = INITIAL_SCHEMA_VERSION,
) -> None:
    """Publish many same-type events atomically around one domain commit.

    The batch counterpart of :func:`publish_committing`, for producers that
    remove or create *many* rows in a single unit of work (bulk deletes) and must
    emit one event per affected row without committing once per event. All outbox
    rows are staged on ``db`` uncommitted and land in the caller's transaction, so
    the domain change and every event commit together or not at all.

    Args:
        event_type: The domain-owned channel shared by every event in the batch.
        payloads: One payload per event. An empty sequence still runs ``commit``
            so the caller's unit of work is committed exactly once either way.
        source: Origin label, e.g. ``api:bulk_delete``.
        metadata_for: Optional per-payload correlation metadata builder.
        db: The producer's SQLAlchemy session (holds the uncommitted change).
        commit: Zero-arg callable that commits the caller's unit of work.

    Returns:
        None.
    """
    metadata_of = metadata_for if metadata_for is not None else (lambda _payload: None)
    if db is not None and _durable_delivery_enabled(event_type):
        # Atomic path: stage every outbox row inside the caller's transaction, so
        # a failure here leaves it uncommitted and the caller rolls back as a
        # whole — no partial domain change, no orphaned event.
        try:
            platform = platform_runtime.get_active_platform()
            now = platform.clock.now()
            for payload in payloads:
                event = _mint(event_type, payload, source, metadata_of(payload), schema_version)
                _stage_in_outbox(platform.recorder, event, db=db, now=now, commit=False)
        except Exception as err:
            logger.error(
                f"Failed to stage {len(payloads)} {event_type} event(s) in the domain transaction: {err}", exc_info=err
            )
            raise
        commit()
    else:
        # Best-effort path: the domain change is the source of truth, so commit it
        # first, then dispatch each event on the bus (swallowing failures — the
        # subscriber's reconciliation net recovers anything dropped).
        commit()
        for payload in payloads:
            _publish_on_bus(event_type, payload, source, metadata_of(payload), schema_version)

Subscribers

jasil.subscribers

Helpers for writing event-bus subscribers.

Every domain subscriber in the codebase is written twice: a raising core (the durable-job handler, so the runner can retry and eventually dead-letter) and a swallowing wrapper (the bus subscriber, so a derived-work failure never breaks the producing request). :func:best_effort is that wrapper, defined once here instead of being copy-pasted per subsystem — which previously meant each copy could drift in what it logged and which exceptions it caught.

best_effort

best_effort(handler)

Wrap a raising event handler into a swallowing bus subscriber.

The returned subscriber logs and absorbs any exception, so derived work can never fail the request or consumer that produced the event. The wrapped handler stays available for durable-job registration, where failures must propagate to drive retry and dead-lettering.

Usage::

def do_work_for_event(event: Event) -> None: ...

on_event_do_work = best_effort(do_work_for_event)

Parameters:

Name	Type	Description	Default
handler	Callable[[Event], None]	The raising handler to wrap.	required

Returns:

Type	Description
Callable[[Event], None]	A subscriber with the same signature that never raises.

Source code in jasil/subscribers.py

def best_effort(handler: Callable[[Event], None]) -> Callable[[Event], None]:
    """Wrap a raising event handler into a swallowing bus subscriber.

    The returned subscriber logs and absorbs **any** exception, so derived work
    can never fail the request or consumer that produced the event. The wrapped
    handler stays available for durable-job registration, where failures must
    propagate to drive retry and dead-lettering.

    Usage::

        def do_work_for_event(event: Event) -> None: ...

        on_event_do_work = best_effort(do_work_for_event)

    Args:
        handler: The raising handler to wrap.

    Returns:
        A subscriber with the same signature that never raises.
    """

    @functools.wraps(handler)
    def _subscriber(event: Event) -> None:
        try:
            handler(event)
        except Exception as err:
            logger.error(
                "Event subscriber failed",
                exc_info=err,
                extra={
                    "event_type": event.event_type,
                    "event_id": event.event_id,
                    "subscriber": handler.__name__,
                    # The whole correlation dict: which keys matter is the host's
                    # to decide, so none are singled out here.
                    "event_metadata": event.metadata,
                },
            )

    return _subscriber

Deployment profile

jasil.profile

Deployment profile and capability-topology resolution.

Pure module — no I/O and no infrastructure imports. It answers two questions the rest of the platform substrate builds on:

• How is this deployment shaped? DeploymentProfile (local / distributed / custom) plus the configured number of web workers.
• Does that shape require state shared across processes? DeploymentTopology.requires_shared_state.

It also classifies a storage URI as memory- or Redis-backed so callers can check the effective wiring against the required shape. Keeping this logic free of infrastructure imports is what lets :mod:jasil.settings and :mod:jasil.capabilities build on it without an import cycle.

DeploymentProfile

Bases: StrEnum

How a deployment is shaped.

Attributes:

Name	Type	Description
LOCAL		Single process/node — in-memory state, local disk, in-process events. The default; requires no extra infrastructure.
DISTRIBUTED		Multi-node — requires shared state (Redis), object storage, and cross-process coordination.
CUSTOM		No profile defaults; every capability must be set explicitly.

Source code in jasil/profile.py

class DeploymentProfile(StrEnum):
    """How a deployment is shaped.

    Attributes:
        LOCAL: Single process/node — in-memory state, local disk, in-process
            events. The default; requires no extra infrastructure.
        DISTRIBUTED: Multi-node — requires shared state (Redis), object storage,
            and cross-process coordination.
        CUSTOM: No profile defaults; every capability must be set explicitly.
    """

    LOCAL = "local"
    DISTRIBUTED = "distributed"
    CUSTOM = "custom"

DeploymentTopology dataclass

Resolved deployment shape.

Attributes:

Name	Type	Description
profile	DeploymentProfile	The deployment profile.
web_workers	int	Number of web-server worker processes (always >= 1).

Source code in jasil/profile.py

@dataclass(frozen=True)
class DeploymentTopology:
    """Resolved deployment shape.

    Attributes:
        profile: The deployment profile.
        web_workers: Number of web-server worker processes (always >= 1).
    """

    profile: DeploymentProfile
    web_workers: int

    @property
    def requires_shared_state(self) -> bool:
        """Whether ephemeral state must be shared across processes.

        True for the ``distributed`` profile or any multi-worker deployment,
        because process-local memory cannot be shared between workers/replicas.
        """
        return self.profile is DeploymentProfile.DISTRIBUTED or self.web_workers > 1

requires_shared_state property

requires_shared_state

Whether ephemeral state must be shared across processes.

True for the distributed profile or any multi-worker deployment, because process-local memory cannot be shared between workers/replicas.

StateBackendKind

Bases: StrEnum

Classification of a state storage URI.

Attributes:

Name	Type	Description
MEMORY		Process-local memory (memory://).
REDIS		Redis-backed (redis:// / rediss:// / unix://).
UNKNOWN		Unrecognized or unset scheme.

Source code in jasil/profile.py

class StateBackendKind(StrEnum):
    """Classification of a state storage URI.

    Attributes:
        MEMORY: Process-local memory (``memory://``).
        REDIS: Redis-backed (``redis://`` / ``rediss://`` / ``unix://``).
        UNKNOWN: Unrecognized or unset scheme.
    """

    MEMORY = "memory"
    REDIS = "redis"
    UNKNOWN = "unknown"

classify_state_uri

classify_state_uri(storage_uri)

Classify a state storage URI as memory / redis / unknown.

Parameters:

Name	Type	Description	Default
storage_uri	str | None	A storage URI (e.g. memory:// or redis://host:6379/0).	required

Returns:

Type	Description
StateBackendKind	The matching StateBackendKind; UNKNOWN when unset or unrecognized.

Source code in jasil/profile.py

def classify_state_uri(storage_uri: str | None) -> StateBackendKind:
    """Classify a state storage URI as memory / redis / unknown.

    Args:
        storage_uri: A storage URI (e.g. ``memory://`` or ``redis://host:6379/0``).

    Returns:
        The matching ``StateBackendKind``; ``UNKNOWN`` when unset or unrecognized.
    """
    if storage_uri is None:
        return StateBackendKind.UNKNOWN
    normalized = storage_uri.strip().lower()
    if normalized.startswith(_MEMORY_SCHEMES):
        return StateBackendKind.MEMORY
    if normalized.startswith(_REDIS_SCHEMES):
        return StateBackendKind.REDIS
    return StateBackendKind.UNKNOWN

parse_profile

parse_profile(value)

Parse a raw profile value into a DeploymentProfile.

Parameters:

Name	Type	Description	Default
value	str | DeploymentProfile | None	Raw environment value, an existing profile, or None.	required

Returns:

Type	Description
DeploymentProfile	The parsed profile; DeploymentProfile.LOCAL when unset or empty.

Raises:

Type	Description
ValueError	When the value is a non-empty, unrecognized profile name. Raising (rather than defaulting) prevents a typo like distributd from silently running the local profile and disabling the shared-state fail-fast.

Source code in jasil/profile.py

def parse_profile(value: str | DeploymentProfile | None) -> DeploymentProfile:
    """Parse a raw profile value into a ``DeploymentProfile``.

    Args:
        value: Raw environment value, an existing profile, or ``None``.

    Returns:
        The parsed profile; ``DeploymentProfile.LOCAL`` when unset or empty.

    Raises:
        ValueError: When the value is a non-empty, unrecognized profile name.
            Raising (rather than defaulting) prevents a typo like
            ``distributd`` from silently running the ``local`` profile and
            disabling the shared-state fail-fast.
    """
    if isinstance(value, DeploymentProfile):
        return value
    if value is None:
        return DeploymentProfile.LOCAL
    normalized = value.strip().lower()
    if not normalized:
        return DeploymentProfile.LOCAL
    try:
        return DeploymentProfile(normalized)
    except ValueError as err:
        valid = ", ".join(profile.value for profile in DeploymentProfile)
        raise ValueError(f"Invalid DEPLOYMENT_PROFILE '{value}'. Valid values: {valid}.") from err

resolve_topology

resolve_topology(profile, web_workers)

Resolve the deployment topology, clamping web_workers to >= 1.

Parameters:

Name	Type	Description	Default
profile	DeploymentProfile	The deployment profile.	required
web_workers	int	Configured worker count.	required

Returns:

Type	Description
DeploymentTopology	The resolved DeploymentTopology.

Source code in jasil/profile.py

def resolve_topology(profile: DeploymentProfile, web_workers: int) -> DeploymentTopology:
    """Resolve the deployment topology, clamping ``web_workers`` to >= 1.

    Args:
        profile: The deployment profile.
        web_workers: Configured worker count.

    Returns:
        The resolved ``DeploymentTopology``.
    """
    return DeploymentTopology(profile=profile, web_workers=max(1, web_workers))

Capability report

jasil.capabilities

Startup capability report and deployment-consistency checks.

Renders how each infrastructure capability (state, storage, events, lock, clock) is wired for a human-readable startup log, and detects fatal inconsistencies — a deployment that requires a shared backend but resolves one to a process- or node-local implementation.

Three consistency rules are enforced:

• Cross-process backends — ephemeral state (rate limiting, throttling, single-use tokens) and the event bus — must not resolve to process-local memory when the topology requires shared state (the distributed profile or more than one web worker); the stores and the bus would diverge silently across processes. Both share the memory:// / redis:// vocabulary and are validated by :func:check_state_consistency.
• Cross-node storage must not resolve to the local filesystem under the distributed profile, where replicas run on separate nodes with no shared disk. A multi-worker local deployment shares one host disk, so local storage stays valid there. Validated by :func:check_storage_consistency.
• The coordination lock must not resolve to the in-process noop lock whenever the deployment runs more than one process (the distributed profile or any multi-worker deployment), or every process would run every scheduled job. Validated by :func:check_lock_consistency.

The clock is always the system clock, so its default is never a fatal choice here.

:func:check_deployment_consistency applies all three to a :class:~jasil.settings.JasilSettings, and jasil.container.build_platform calls it at startup: an inconsistency raises unless the host sets enforce_deployment_consistency=False, which downgrades it to a warning.

Pure module — no infrastructure imports, so every check runs before a single backend has been constructed.

CapabilityReport dataclass

A rendered snapshot of how each capability is wired at startup.

Source code in jasil/capabilities.py

@dataclass(frozen=True)
class CapabilityReport:
    """A rendered snapshot of how each capability is wired at startup."""

    topology: DeploymentTopology
    rows: tuple[CapabilityRow, ...]

    def render(self) -> str:
        """Render the report as a multi-line, aligned string."""
        header = (
            f"Deployment profile: {self.topology.profile.value} "
            f"(web_workers={self.topology.web_workers}, "
            f"requires_shared_state={self.topology.requires_shared_state})"
        )
        width = max((len(row.name) for row in self.rows), default=0)
        lines = [f"  {row.name.ljust(width)} -> {row.backend}  (source: {row.source})" for row in self.rows]
        return "\n".join([header, *lines])

render

render()

Render the report as a multi-line, aligned string.

Source code in jasil/capabilities.py

def render(self) -> str:
    """Render the report as a multi-line, aligned string."""
    header = (
        f"Deployment profile: {self.topology.profile.value} "
        f"(web_workers={self.topology.web_workers}, "
        f"requires_shared_state={self.topology.requires_shared_state})"
    )
    width = max((len(row.name) for row in self.rows), default=0)
    lines = [f"  {row.name.ljust(width)} -> {row.backend}  (source: {row.source})" for row in self.rows]
    return "\n".join([header, *lines])

CapabilityRow dataclass

One line of the capability report.

Attributes:

Name	Type	Description
name	str	Capability name (state / storage / events / lock / clock).
backend	str	The resolved backend (e.g. memory, local, system).
source	str	Where the value came from (setting name or profile default).

Source code in jasil/capabilities.py

@dataclass(frozen=True)
class CapabilityRow:
    """One line of the capability report.

    Attributes:
        name: Capability name (state / storage / events / lock / clock).
        backend: The resolved backend (e.g. ``memory``, ``local``, ``system``).
        source: Where the value came from (setting name or ``profile default``).
    """

    name: str
    backend: str
    source: str

StateSource dataclass

A configured source of ephemeral state.

Attributes:

Name	Type	Description
label	str	The setting backing this state (e.g. state_uri).
uri	str	The effective storage URI.
applies	bool	Whether this source is active — a capability the host disabled cannot be misconfigured.

Source code in jasil/capabilities.py

@dataclass(frozen=True)
class StateSource:
    """A configured source of ephemeral state.

    Attributes:
        label: The setting backing this state (e.g. ``state_uri``).
        uri: The effective storage URI.
        applies: Whether this source is active — a capability the host disabled
            cannot be misconfigured.
    """

    label: str
    uri: str
    applies: bool = True

    @property
    def backend(self) -> StateBackendKind:
        """The classified backend for this source's URI."""
        return classify_state_uri(self.uri)

backend property

backend

The classified backend for this source's URI.

build_capability_report

build_capability_report(settings)

Build the observational capability report for startup logging.

Parameters:

Name	Type	Description	Default
settings	JasilSettings	The configuration the platform is being built from.	required

Returns:

Type	Description
CapabilityReport	A CapabilityReport reflecting the effective wiring: the backend each
CapabilityReport	capability resolved to, and whether that came from an explicit setting or
CapabilityReport	from the deployment profile's default.

Raises:

Type	Description
ValueError	When a capability URI is unset under a profile that has no default for it.

Source code in jasil/capabilities.py

def build_capability_report(settings: JasilSettings) -> CapabilityReport:
    """Build the observational capability report for startup logging.

    Args:
        settings: The configuration the platform is being built from.

    Returns:
        A ``CapabilityReport`` reflecting the effective wiring: the backend each
        capability resolved to, and whether that came from an explicit setting or
        from the deployment profile's default.

    Raises:
        ValueError: When a capability URI is unset under a profile that has no
            default for it.
    """
    rows = (
        CapabilityRow("state", _scheme_of(settings.resolved_state_uri), _source_of(settings.state_uri, "state_uri")),
        CapabilityRow(
            "storage", _scheme_of(settings.resolved_storage_uri), _source_of(settings.storage_uri, "storage_uri")
        ),
        CapabilityRow(
            "events", _scheme_of(settings.resolved_events_uri), _source_of(settings.events_uri, "events_uri")
        ),
        CapabilityRow("lock", _scheme_of(settings.resolved_lock_uri), _source_of(settings.lock_uri, "lock_uri")),
        CapabilityRow("clock", "system", "always the system clock"),
    )
    return CapabilityReport(topology=resolve_topology(settings.profile, settings.web_workers), rows=rows)

check_deployment_consistency

check_deployment_consistency(settings)

Return every fatal wiring inconsistency in settings (empty when sound).

The entry point build_platform uses: it resolves the capability URIs the profile would actually build from, then applies all three rules to them.

Parameters:

Name	Type	Description	Default
settings	JasilSettings	The configuration the platform is being built from.	required

Returns:

Type	Description
list[str]	Human-readable issue messages; empty when the wiring is consistent.

Raises:

Type	Description
ValueError	When a capability URI is unset under a profile that has no default for it.

Source code in jasil/capabilities.py

def check_deployment_consistency(settings: JasilSettings) -> list[str]:
    """Return every fatal wiring inconsistency in ``settings`` (empty when sound).

    The entry point ``build_platform`` uses: it resolves the capability URIs the
    profile would actually build from, then applies all three rules to them.

    Args:
        settings: The configuration the platform is being built from.

    Returns:
        Human-readable issue messages; empty when the wiring is consistent.

    Raises:
        ValueError: When a capability URI is unset under a profile that has no
            default for it.
    """
    profile = settings.profile
    web_workers = settings.web_workers
    return [
        *check_state_consistency(
            profile=profile,
            web_workers=web_workers,
            state_sources=(
                StateSource(label="state_uri", uri=settings.resolved_state_uri),
                StateSource(label="events_uri", uri=settings.resolved_events_uri),
            ),
        ),
        *check_storage_consistency(
            profile=profile,
            storage_uri=settings.resolved_storage_uri,
            storage_label="storage_uri",
        ),
        *check_lock_consistency(
            profile=profile,
            web_workers=web_workers,
            lock_uri=settings.resolved_lock_uri,
            lock_label="lock_uri",
        ),
    ]

check_lock_consistency

check_lock_consistency(
    *, profile, web_workers, lock_uri, lock_label
)

Return a fatal issue when a multi-process deployment uses a no-op lock.

The coordination lock makes scheduled and backfill work single-runner across processes. Whenever a deployment runs more than one process — the distributed profile or any multi-worker deployment (:attr:~jasil.profile.DeploymentTopology.requires_shared_state) — an in-process noop:// lock coordinates nothing, so every process would run every interval job (the retention prune, a backfill, an upstream sync). The profile-aware default already resolves to postgres-advisory:// in that case, so this only trips on an explicit lock_uri="noop://" override. A single-process local deployment has nothing to coordinate, so noop stays valid there.

Parameters:

Name	Type	Description	Default
profile	DeploymentProfile	The deployment profile.	required
web_workers	int	Configured worker count.	required
lock_uri	str	The resolved coordination-lock URI.	required
lock_label	str	The setting backing the lock (for the message).	required

Returns:

Type	Description
list[str]	A single-item issue list when misconfigured; empty otherwise. The
list[str]	custom profile is exempt — it promises no defaults.

Source code in jasil/capabilities.py

def check_lock_consistency(
    *,
    profile: DeploymentProfile,
    web_workers: int,
    lock_uri: str,
    lock_label: str,
) -> list[str]:
    """Return a fatal issue when a multi-process deployment uses a no-op lock.

    The coordination lock makes scheduled and backfill work single-runner across
    processes. Whenever a deployment runs more than one process — the
    ``distributed`` profile or any multi-worker deployment
    (:attr:`~jasil.profile.DeploymentTopology.requires_shared_state`) — an
    in-process ``noop://`` lock coordinates nothing, so every process would run
    every interval job (the retention prune, a backfill, an upstream sync). The
    profile-aware default already resolves to ``postgres-advisory://`` in that
    case, so this only trips on an explicit ``lock_uri="noop://"`` override. A
    single-process ``local`` deployment has nothing to coordinate, so ``noop``
    stays valid there.

    Args:
        profile: The deployment profile.
        web_workers: Configured worker count.
        lock_uri: The resolved coordination-lock URI.
        lock_label: The setting backing the lock (for the message).

    Returns:
        A single-item issue list when misconfigured; empty otherwise. The
        ``custom`` profile is exempt — it promises no defaults.
    """
    if profile is DeploymentProfile.CUSTOM:
        return []
    topology = resolve_topology(profile, web_workers)
    if not topology.requires_shared_state:
        return []
    if not lock_uri.strip().lower().startswith("noop://"):
        return []
    return [
        f"{lock_label} resolves to an in-process no-op lock, but "
        f"profile={profile.value} with web_workers={topology.web_workers} "
        f"runs multiple processes that would each run scheduled jobs. Point it at "
        f"the shared database lock (postgres-advisory://)."
    ]

check_state_consistency

check_state_consistency(
    *, profile, web_workers, state_sources
)

Return fatal issues for cross-process backends (empty when consistent).

A deployment that requires shared state (the distributed profile or more than one web worker) but resolves a cross-process backend to process-local memory is fatally misconfigured: ephemeral state (rate-limit counters, lockout gates, single-use tokens) and the in-process event bus would silently diverge across processes. State stores and the event bus share the memory:// / redis:// vocabulary, so both are validated here.

Parameters:

Name	Type	Description	Default
profile	DeploymentProfile	The deployment profile.	required
web_workers	int	Configured worker count.	required
state_sources	Sequence[StateSource]	The active cross-process backends to validate (the resolved state and event-bus URIs).	required

Returns:

Type	Description
list[str]	Human-readable issue messages; empty when consistent. The custom
list[str]	profile is exempt — it promises no defaults, so nothing can contradict one.

Source code in jasil/capabilities.py

def check_state_consistency(
    *,
    profile: DeploymentProfile,
    web_workers: int,
    state_sources: Sequence[StateSource],
) -> list[str]:
    """Return fatal issues for cross-process backends (empty when consistent).

    A deployment that requires shared state (the ``distributed`` profile or more
    than one web worker) but resolves a cross-process backend to process-local
    memory is fatally misconfigured: ephemeral state (rate-limit counters, lockout
    gates, single-use tokens) and the in-process event bus would silently diverge
    across processes. State stores and the event bus share the ``memory://`` /
    ``redis://`` vocabulary, so both are validated here.

    Args:
        profile: The deployment profile.
        web_workers: Configured worker count.
        state_sources: The active cross-process backends to validate (the
            resolved state and event-bus URIs).

    Returns:
        Human-readable issue messages; empty when consistent. The ``custom``
        profile is exempt — it promises no defaults, so nothing can contradict one.
    """
    if profile is DeploymentProfile.CUSTOM:
        return []
    topology = resolve_topology(profile, web_workers)
    if not topology.requires_shared_state:
        return []
    issues: list[str] = []
    for source in state_sources:
        if not source.applies or source.backend is StateBackendKind.REDIS:
            continue
        reason = (
            "process-local memory" if source.backend is StateBackendKind.MEMORY else "an unrecognized storage scheme"
        )
        issues.append(
            f"{source.label} resolves to {reason}, but "
            f"profile={profile.value} with web_workers={topology.web_workers} "
            f"requires a backend shared across processes. Point it at Redis "
            f"(redis://...) or run a single worker under the local profile."
        )
    return issues

check_storage_consistency

check_storage_consistency(
    *, profile, storage_uri, storage_label
)

Return a fatal issue when distributed storage resolves to local disk.

Under the distributed profile replicas run on separate nodes that do not share a filesystem, so blob storage must be object storage. A multi-worker local deployment shares one host disk, so local storage stays valid there and is not flagged.

Parameters:

Name	Type	Description	Default
profile	DeploymentProfile	The deployment profile.	required
storage_uri	str	The resolved blob-storage URI.	required
storage_label	str	The setting backing blob storage (for the message).	required

Returns:

Type	Description
list[str]	A single-item issue list when misconfigured; empty otherwise.

Source code in jasil/capabilities.py

def check_storage_consistency(
    *,
    profile: DeploymentProfile,
    storage_uri: str,
    storage_label: str,
) -> list[str]:
    """Return a fatal issue when distributed storage resolves to local disk.

    Under the ``distributed`` profile replicas run on separate nodes that do not
    share a filesystem, so blob storage must be object storage. A multi-worker
    ``local`` deployment shares one host disk, so local storage stays valid there
    and is not flagged.

    Args:
        profile: The deployment profile.
        storage_uri: The resolved blob-storage URI.
        storage_label: The setting backing blob storage (for the message).

    Returns:
        A single-item issue list when misconfigured; empty otherwise.
    """
    if profile is not DeploymentProfile.DISTRIBUTED:
        return []
    if not storage_uri.strip().lower().startswith("local://"):
        return []
    return [
        f"{storage_label} resolves to the local filesystem, but "
        f"profile={profile.value} runs replicas on separate nodes that do "
        f"not share a disk. Point it at object storage (s3://bucket/...)."
    ]

Settings

jasil.settings

Host-supplied configuration for JASIL.

JASIL never reads environment variables or secret files itself. The host builds a :class:JasilSettings from whatever configuration source it likes and installs it once at startup::

import jasil.settings as jasil_settings

jasil_settings.configure(
    jasil_settings.JasilSettings(
        profile=jasil.DeploymentProfile.DISTRIBUTED,
        state_uri="redis://cache:6379/0",
        storage_uri="s3://bucket",
        events_uri="redis://cache:6379/1",
    )
)

Every component reads the installed settings through :func:get_settings.

Shape. Configuration is grouped by concern — :class:JobSettings, :class:EventLogSettings, :class:GeocodingSettings, :class:NetworkSettings — rather than one flat list, so a host enabling durable jobs reads one small class. Only the values describing the deployment shape sit at the top level.

Profile defaults. The four capability URIs may be left unset, in which case the deployment profile supplies the default: local resolves to the single-process backends, while distributed and custom refuse to guess and raise, because a Redis host or bucket name cannot be inferred.

EventLogSettings dataclass

Event-observability trail configuration.

Attributes:

Name	Type	Description
enabled	bool	Record every event's lifecycle to the event_log table.
retention_days	int	Age at which trail rows are pruned. <= 0 disables pruning.

Source code in jasil/settings.py

@dataclass(frozen=True)
class EventLogSettings:
    """Event-observability trail configuration.

    Attributes:
        enabled: Record every event's lifecycle to the ``event_log`` table.
        retention_days: Age at which trail rows are pruned. ``<= 0`` disables
            pruning.
    """

    enabled: bool = False
    retention_days: int = 30

GeocodingSettings dataclass

Reverse-geocoding backend configuration.

Attributes:

Name	Type	Description
provider	str	"nominatim", "photon", or "geocode". Any other value (including the empty default) disables the capability.
rate_limit	float	Maximum requests per second; <= 0 disables throttling.
api_key	str | None	API key, for the services requiring one (geocode.maps.co).
nominatim_host	str	Bare host[:port] authority for Nominatim.
nominatim_use_https	bool	Address Nominatim over HTTPS.
photon_host	str	Bare host[:port] authority for Photon.
photon_use_https	bool	Address Photon over HTTPS.
user_agent	str	User-Agent sent upstream. Nominatim's usage policy requires an identifying value, so hosts should set their own.

Source code in jasil/settings.py

@dataclass(frozen=True)
class GeocodingSettings:
    """Reverse-geocoding backend configuration.

    Attributes:
        provider: ``"nominatim"``, ``"photon"``, or ``"geocode"``. Any other
            value (including the empty default) disables the capability.
        rate_limit: Maximum requests per second; ``<= 0`` disables throttling.
        api_key: API key, for the services requiring one (geocode.maps.co).
        nominatim_host: Bare ``host[:port]`` authority for Nominatim.
        nominatim_use_https: Address Nominatim over HTTPS.
        photon_host: Bare ``host[:port]`` authority for Photon.
        photon_use_https: Address Photon over HTTPS.
        user_agent: ``User-Agent`` sent upstream. Nominatim's usage policy
            requires an identifying value, so hosts should set their own.
    """

    provider: str = ""
    rate_limit: float = 1.0
    api_key: str | None = None
    nominatim_host: str = ""
    nominatim_use_https: bool = True
    photon_host: str = ""
    photon_use_https: bool = True
    user_agent: str = "jasil (ReverseGeocoding)"

JasilSettings dataclass

The full JASIL configuration.

Attributes:

Name	Type	Description
profile	DeploymentProfile	The deployment shape; supplies the capability-URI defaults.
web_workers	int	How many web-server worker processes the host runs. Only the count matters: four workers under the local profile are still four processes, and process-local state cannot be shared between them, so this drives the consistency checks as much as the profile does.
enforce_deployment_consistency	bool	Refuse to build a platform whose wiring contradicts its topology (see :mod:jasil.capabilities). Set False to log the issues as warnings instead — useful on a development machine running the distributed profile without Redis.
data_dir	str	Root directory for the local storage backend.
state_uri	str | None	memory:// or redis:// / rediss:// / unix://.
storage_uri	str | None	local:// or s3://.
events_uri	str | None	memory:// or redis:// / rediss:// / unix://.
lock_uri	str | None	noop:// or postgres-advisory://.
jobs	JobSettings	Durable-job pipeline configuration.
event_log	EventLogSettings	Event-observability trail configuration.
geocoding	GeocodingSettings	Reverse-geocoding backend configuration.
network	NetworkSettings	Outbound-egress configuration.

Source code in jasil/settings.py

@dataclass(frozen=True)
class JasilSettings:
    """The full JASIL configuration.

    Attributes:
        profile: The deployment shape; supplies the capability-URI defaults.
        web_workers: How many web-server worker processes the host runs. Only the
            *count* matters: four workers under the ``local`` profile are still
            four processes, and process-local state cannot be shared between them,
            so this drives the consistency checks as much as the profile does.
        enforce_deployment_consistency: Refuse to build a platform whose wiring
            contradicts its topology (see :mod:`jasil.capabilities`). Set False to
            log the issues as warnings instead — useful on a development machine
            running the distributed profile without Redis.
        data_dir: Root directory for the local storage backend.
        state_uri: ``memory://`` or ``redis://`` / ``rediss://`` / ``unix://``.
        storage_uri: ``local://`` or ``s3://``.
        events_uri: ``memory://`` or ``redis://`` / ``rediss://`` / ``unix://``.
        lock_uri: ``noop://`` or ``postgres-advisory://``.
        jobs: Durable-job pipeline configuration.
        event_log: Event-observability trail configuration.
        geocoding: Reverse-geocoding backend configuration.
        network: Outbound-egress configuration.
    """

    profile: DeploymentProfile = DeploymentProfile.LOCAL
    web_workers: int = 1
    enforce_deployment_consistency: bool = True
    data_dir: str = "data"
    state_uri: str | None = None
    storage_uri: str | None = None
    events_uri: str | None = None
    lock_uri: str | None = None
    jobs: JobSettings = field(default_factory=JobSettings)
    event_log: EventLogSettings = field(default_factory=EventLogSettings)
    geocoding: GeocodingSettings = field(default_factory=GeocodingSettings)
    network: NetworkSettings = field(default_factory=NetworkSettings)

    def _resolve(self, name: str) -> str:
        """Return an explicit capability URI, or the profile's default.

        Raises:
            ValueError: When the URI is unset and the profile has no default.
        """
        configured = getattr(self, name)
        if configured:
            return str(configured)
        if self.profile is DeploymentProfile.LOCAL:
            return _LOCAL_DEFAULT_URIS[name]
        raise ValueError(
            f"{name} must be set explicitly for the {self.profile.value!r} deployment profile; "
            f"only the 'local' profile has a default ({_LOCAL_DEFAULT_URIS[name]})."
        )

    @property
    def resolved_state_uri(self) -> str:
        """The effective state-backend URI."""
        return self._resolve("state_uri")

    @property
    def resolved_storage_uri(self) -> str:
        """The effective storage-backend URI."""
        return self._resolve("storage_uri")

    @property
    def resolved_events_uri(self) -> str:
        """The effective event-bus URI."""
        return self._resolve("events_uri")

    @property
    def resolved_lock_uri(self) -> str:
        """The effective lock-backend URI."""
        return self._resolve("lock_uri")

resolved_events_uri property

resolved_events_uri

The effective event-bus URI.

resolved_lock_uri property

resolved_lock_uri

The effective lock-backend URI.

resolved_state_uri property

resolved_state_uri

The effective state-backend URI.

resolved_storage_uri property

resolved_storage_uri

The effective storage-backend URI.

JobSettings dataclass

Durable-job pipeline configuration.

Attributes:

Name	Type	Description
enabled	bool	Route events through the transactional outbox instead of the event bus. When false the whole jobs layer stays dormant.
lease_seconds	int	How long a claimed job is leased to a worker before the reaper may reclaim it.
batch_size	int	Maximum rows claimed or relayed per pass.
backoff_base_seconds	int	First retry delay; doubles per attempt.
backoff_max_seconds	int	Ceiling for the exponential backoff.
poll_interval_seconds	float	Idle wait between empty polls.
max_attempts	int	Attempts before a job is dead-lettered.
retention_days	int	Age at which relayed outbox rows and completed jobs are pruned. <= 0 disables pruning.

Source code in jasil/settings.py

@dataclass(frozen=True)
class JobSettings:
    """Durable-job pipeline configuration.

    Attributes:
        enabled: Route events through the transactional outbox instead of the
            event bus. When false the whole jobs layer stays dormant.
        lease_seconds: How long a claimed job is leased to a worker before the
            reaper may reclaim it.
        batch_size: Maximum rows claimed or relayed per pass.
        backoff_base_seconds: First retry delay; doubles per attempt.
        backoff_max_seconds: Ceiling for the exponential backoff.
        poll_interval_seconds: Idle wait between empty polls.
        max_attempts: Attempts before a job is dead-lettered.
        retention_days: Age at which relayed outbox rows and completed jobs are
            pruned. ``<= 0`` disables pruning.
    """

    enabled: bool = False
    lease_seconds: int = 300
    batch_size: int = 20
    backoff_base_seconds: int = 60
    backoff_max_seconds: int = 3600
    poll_interval_seconds: float = 5.0
    max_attempts: int = 5
    retention_days: int = 30

NetworkSettings dataclass

Outbound-egress configuration.

Attributes:

Name	Type	Description
ssrf_allowed_hosts	tuple[str, ...]	Hostnames and CIDRs exempt from the SSRF address denylist, so a self-hosted service on a private network stays reachable. Every use is logged.

Source code in jasil/settings.py

@dataclass(frozen=True)
class NetworkSettings:
    """Outbound-egress configuration.

    Attributes:
        ssrf_allowed_hosts: Hostnames and CIDRs exempt from the SSRF address
            denylist, so a self-hosted service on a private network stays
            reachable. Every use is logged.
    """

    ssrf_allowed_hosts: tuple[str, ...] = ()

configure

configure(settings)

Install the host's settings for the process.

Parameters:

Name	Type	Description	Default
settings	JasilSettings	The configuration to install.	required

Source code in jasil/settings.py

def configure(settings: JasilSettings) -> None:
    """Install the host's settings for the process.

    Args:
        settings: The configuration to install.
    """
    _settings.configure(settings)

get_settings

get_settings()

Return the installed settings, or the all-defaults instance.

Source code in jasil/settings.py

def get_settings() -> JasilSettings:
    """Return the installed settings, or the all-defaults instance."""
    return _settings.get()

is_configured

is_configured()

Return whether :func:configure has been called.

Source code in jasil/settings.py

def is_configured() -> bool:
    """Return whether :func:`configure` has been called."""
    return _settings.is_configured()

reset

reset()

Restore the all-defaults settings.

For tests; production code configures once at startup and never resets.

Source code in jasil/settings.py

def reset() -> None:
    """Restore the all-defaults settings.

    For tests; production code configures once at startup and never resets.
    """
    _settings.reset()

Correlation

jasil.correlation

Correlation-id seam for stamping events with the ambient request id.

Events carry a correlation id so a downstream failure can be traced back to the request that triggered it. Where that id comes from is the host's business: a web framework's request middleware, a task-queue header, or nothing at all.

By default the id lives in a module-local :class:~contextvars.ContextVar that the host sets with :func:set_correlation_id. A host that already tracks one (request-id middleware, an OpenTelemetry span) installs a reader instead::

import jasil.correlation as correlation
correlation.configure_provider(my_middleware.get_request_id)

Both paths are optional; with neither configured, events simply carry no correlation id.

configure_provider

configure_provider(provider)

Install a host callable that returns the current correlation id.

Call once at startup. Passing None restores the built-in contextvar.

Parameters:

Name	Type	Description	Default
provider	Callable[[], str | None] | None	Returns the ambient correlation id, or None when there is none (e.g. outside a request).	required

Source code in jasil/correlation.py

def configure_provider(provider: Callable[[], str | None] | None) -> None:
    """Install a host callable that returns the current correlation id.

    Call once at startup. Passing ``None`` restores the built-in contextvar.

    Args:
        provider: Returns the ambient correlation id, or ``None`` when there is
            none (e.g. outside a request).
    """
    global _provider
    _provider = provider

get_correlation_id

get_correlation_id()

Return the ambient correlation id, or None.

Never raises: a provider that fails is treated as "no id", because a correlation id is diagnostic metadata and must not break publishing.

Source code in jasil/correlation.py

def get_correlation_id() -> str | None:
    """Return the ambient correlation id, or ``None``.

    Never raises: a provider that fails is treated as "no id", because a
    correlation id is diagnostic metadata and must not break publishing.
    """
    if _provider is not None:
        try:
            return _provider()
        except Exception:
            return None
    return _correlation_id.get()

reset

reset()

Clear the installed provider and the current context's id.

Source code in jasil/correlation.py

def reset() -> None:
    """Clear the installed provider and the current context's id."""
    global _provider
    _provider = None
    _correlation_id.set(None)

set_correlation_id

set_correlation_id(value)

Set the correlation id for the current context.

Only consulted while no provider is installed via :func:configure_provider.

Parameters:

Name	Type	Description	Default
value	str | None	The id to stamp on events minted in this context.	required

Source code in jasil/correlation.py

def set_correlation_id(value: str | None) -> None:
    """Set the correlation id for the current context.

    Only consulted while no provider is installed via :func:`configure_provider`.

    Args:
        value: The id to stamp on events minted in this context.
    """
    _correlation_id.set(value)

ORM integration

jasil.orm

JASIL's SQLAlchemy registry plumbing and session-factory helpers.

Option B (the host owns the Base). JASIL's companion tables — event_log, processing_jobs, event_outbox — must live in the same declarative registry as the host's own models so that one create_all / migration run and one metadata object cover the whole schema. The host owns that registry; JASIL maps its models into it.

Host applications:

1. Own a declarative base::

   from sqlalchemy.orm import DeclarativeBase

   class Base(DeclarativeBase): ... # the host's own base (naming conventions, schema, ...)

2. Map JASIL's tables into that base's registry, once, at startup::

   import jasil.orm as jasil_orm jasil_orm.map_models(Base)

This must happen before any JASIL database use — importing a JASIL model module (or a CRUD module that imports one) beforehand is a configuration error. A host that would rather not own a base may call map_models() with no argument and use JASIL's convenience :data:Base.

1. Register a session factory bound to their own engine::

   from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker

   engine = create_engine(...) jasil_orm.configure_sessionmaker(sessionmaker(bind=engine))

JASIL never creates the engine; the host owns the connection.

Base

Bases: DeclarativeBase

JASIL's convenience declarative base.

Under Option B the host owns the registry: define your own :class:~sqlalchemy.orm.DeclarativeBase and pass it to :func:map_models. Use this default only if you would rather not own one.

Source code in jasil/orm.py

class Base(DeclarativeBase):
    """JASIL's convenience declarative base.

    Under Option B the **host** owns the registry: define your own
    :class:`~sqlalchemy.orm.DeclarativeBase` and pass it to :func:`map_models`.
    Use this default only if you would rather not own one.
    """

configure_sessionmaker

configure_sessionmaker(factory)

Install the host's session factory.

Call once at startup with a sessionmaker bound to the application's engine. The event-log recorder, the job runner, the relay, and the retention sweeps all obtain their sessions from it.

Parameters:

Name	Type	Description	Default
factory	sessionmaker[Session]	A configured sessionmaker.	required

Source code in jasil/orm.py

def configure_sessionmaker(factory: sessionmaker[Session]) -> None:
    """Install the host's session factory.

    Call once at startup with a ``sessionmaker`` bound to the application's
    engine. The event-log recorder, the job runner, the relay, and the retention
    sweeps all obtain their sessions from it.

    Args:
        factory: A configured ``sessionmaker``.
    """
    _session_factory.configure(factory)

get_active_base

get_active_base()

Return the declarative base JASIL's models are mapped onto.

Model modules call this at import time to obtain their base, so importing a model module before :func:map_models is a configuration error.

Raises:

Type	Description
RuntimeError	If :func:map_models has not been called yet.

Source code in jasil/orm.py

def get_active_base() -> type[DeclarativeBase]:
    """Return the declarative base JASIL's models are mapped onto.

    Model modules call this at import time to obtain their base, so importing a
    model module before :func:`map_models` is a configuration error.

    Raises:
        RuntimeError: If :func:`map_models` has not been called yet.
    """
    if _active_base is None:
        raise RuntimeError(
            "JASIL's models are not mapped yet. Call jasil.orm.map_models(YourBase) "
            "once at startup, before any database use. Omit the base to use jasil.orm.Base."
        )
    return _active_base

get_engine

get_engine()

Return the engine the session factory is bound to.

Needed by the Postgres advisory-lock backend, which holds a dedicated connection for the lifetime of a lock and so cannot work through a session.

Raises:

Type	Description
RuntimeError	If :func:configure_sessionmaker has not been called, or its factory is not bound to an engine.

Source code in jasil/orm.py

def get_engine() -> Any:
    """Return the engine the session factory is bound to.

    Needed by the Postgres advisory-lock backend, which holds a dedicated
    connection for the lifetime of a lock and so cannot work through a session.

    Raises:
        RuntimeError: If :func:`configure_sessionmaker` has not been called, or
            its factory is not bound to an engine.
    """
    bind = get_sessionmaker().kw.get("bind")
    if bind is None:
        raise RuntimeError(
            "JASIL's session factory is not bound to an engine. Pass bind= to "
            "sessionmaker(...) so capabilities needing a raw connection (the "
            "postgres-advisory lock) can reach it."
        )
    return bind

get_sessionmaker

get_sessionmaker()

Return the installed session factory.

Raises:

Type	Description
RuntimeError	If :func:configure_sessionmaker has not been called.

Source code in jasil/orm.py

def get_sessionmaker() -> sessionmaker[Session]:
    """Return the installed session factory.

    Raises:
        RuntimeError: If :func:`configure_sessionmaker` has not been called.
    """
    return _session_factory.get()

is_models_mapped

is_models_mapped()

Return whether :func:map_models has been called.

Source code in jasil/orm.py

def is_models_mapped() -> bool:
    """Return whether :func:`map_models` has been called."""
    return _active_base is not None

is_sessionmaker_configured

is_sessionmaker_configured()

Return whether :func:configure_sessionmaker has been called.

Source code in jasil/orm.py

def is_sessionmaker_configured() -> bool:
    """Return whether :func:`configure_sessionmaker` has been called."""
    return _session_factory.is_configured()

jasil_table_names

jasil_table_names()

Return the names of the tables JASIL owns.

Source code in jasil/orm.py

def jasil_table_names() -> frozenset[str]:
    """Return the names of the tables JASIL owns."""
    return _TABLE_NAMES

map_models

map_models(base=None)

Define and map JASIL's companion tables into base's registry.

Call once at startup, before any database use. Calling it again with the same base is a no-op, so a host with several entry points need not coordinate.

Parameters:

Name	Type	Description	Default
base	type[DeclarativeBase] | None	The host's :class:~sqlalchemy.orm.DeclarativeBase subclass. Omit it to use JASIL's own :data:Base.	None

Raises:

Type	Description
RuntimeError	If called again with a different base.

Source code in jasil/orm.py

def map_models(base: type[DeclarativeBase] | None = None) -> None:
    """Define and map JASIL's companion tables into ``base``'s registry.

    Call once at startup, before any database use. Calling it again with the same
    base is a no-op, so a host with several entry points need not coordinate.

    Args:
        base: The host's :class:`~sqlalchemy.orm.DeclarativeBase` subclass. Omit
            it to use JASIL's own :data:`Base`.

    Raises:
        RuntimeError: If called again with a different base.
    """
    global _active_base
    target = base if base is not None else Base
    if _active_base is not None:
        if _active_base is not target:
            raise RuntimeError("jasil.orm.map_models() was already called with a different base; call it once.")
        return
    _active_base = target
    try:
        for module_name in _MODEL_MODULES:
            importlib.import_module(module_name)
        # Resolve every mapper now so a misconfiguration fails fast at startup
        # rather than on the first query.
        target.registry.configure()
    except Exception:
        _active_base = None  # let the host fix the problem and retry
        raise

reset

reset()

Clear the mapped base and the session factory.

For tests that need a clean process-wide state between cases; production code configures once at startup and never resets.

Source code in jasil/orm.py

def reset() -> None:
    """Clear the mapped base and the session factory.

    For tests that need a clean process-wide state between cases; production code
    configures once at startup and never resets.
    """
    global _active_base
    _active_base = None
    _session_factory.reset()

Composition root

jasil.container

The composition root: build the platform substrate from settings.

build_platform resolves each capability (state, storage, events, lock, clock) to a concrete backend based on the deployment profile and returns a frozen Platform holding the providers. It is called once at startup and published process-wide via jasil.runtime so both request and non-request code resolve the same instance.

Every capability resolves its backend by URI scheme, independently of the profile: memory/redis for the state URI; local/s3 for the storage URI; memory/redis for the events URI; noop/postgres-advisory for the lock URI. The deployment profile only shapes the defaults those URIs resolve to (see the resolved_* properties on :class:~jasil.settings.JasilSettings), so local, distributed, and custom all build the same way — the profile just picks memory-vs-Redis and local-fs-vs-S3 defaults.

Before anything is constructed, the resolved wiring is checked against the deployment topology (see :mod:jasil.capabilities): a combination that would silently diverge across processes or nodes stops the build rather than becoming a production mystery.

Platform dataclass

The assembled platform substrate — one instance per process.

Attributes:

Name	Type	Description
profile	DeploymentProfile	The active deployment profile.
state	StateProvider	Ephemeral keyed-state provider.
storage	StorageProvider	Blob-storage provider.
events	EventBusProvider	Publish/subscribe provider.
lock	LockProvider	Coordination-lock provider.
clock	ClockProvider	Time-source provider.
geocoding	GeocodingProvider	Reverse-geocoding provider. Always present — when geocoding is unconfigured or misconfigured this is a no-op backend, so callers never branch on whether the capability exists.
recorder	EventRecorder | None	Event-log recorder, or None when event logging is disabled. Shared by the event bus (best-effort delivery) and the publish facade (durable delivery) so both paths land in the event_log dashboard.

Source code in jasil/container.py

@dataclass(frozen=True)
class Platform:
    """The assembled platform substrate — one instance per process.

    Attributes:
        profile: The active deployment profile.
        state: Ephemeral keyed-state provider.
        storage: Blob-storage provider.
        events: Publish/subscribe provider.
        lock: Coordination-lock provider.
        clock: Time-source provider.
        geocoding: Reverse-geocoding provider. Always present — when geocoding is
            unconfigured or misconfigured this is a no-op backend, so callers
            never branch on whether the capability exists.
        recorder: Event-log recorder, or ``None`` when event logging is disabled.
            Shared by the event bus (best-effort delivery) and the publish facade
            (durable delivery) so both paths land in the event_log dashboard.
    """

    profile: DeploymentProfile
    state: StateProvider
    storage: StorageProvider
    events: EventBusProvider
    lock: LockProvider
    clock: ClockProvider
    geocoding: GeocodingProvider
    recorder: EventRecorder | None

build_platform

build_platform(settings=None)

Assemble the Platform for the configured deployment profile.

Parameters:

Name	Type	Description	Default
settings	JasilSettings | None	The configuration to build from. Defaults to the settings the host installed via jasil.settings.configure.	None

Returns:

Type	Description
Platform	A frozen Platform wiring each provider to its selected backend.

Raises:

Type	Description
ValueError	When a capability URI uses an unsupported scheme, is unset under a profile that has no default for it, or when the resolved wiring contradicts the deployment topology.
RuntimeError	When a selected Redis backend cannot be reached.

Source code in jasil/container.py

def build_platform(settings: JasilSettings | None = None) -> Platform:
    """Assemble the ``Platform`` for the configured deployment profile.

    Args:
        settings: The configuration to build from. Defaults to the settings the
            host installed via ``jasil.settings.configure``.

    Returns:
        A frozen ``Platform`` wiring each provider to its selected backend.

    Raises:
        ValueError: When a capability URI uses an unsupported scheme, is unset
            under a profile that has no default for it, or when the resolved
            wiring contradicts the deployment topology.
        RuntimeError: When a selected Redis backend cannot be reached.
    """
    settings = settings if settings is not None else get_settings()
    _check_deployment_consistency(settings)
    logger.info("JASIL platform capabilities:\n%s", capabilities.build_capability_report(settings).render())
    profile = settings.profile
    # Build the recorder once and share it: the event bus records the lifecycle
    # of best-effort (bus-delivered) events, while the publish facade uses the
    # same recorder to record 'published' for durable (outbox-delivered) events
    # so the event_log dashboard never goes dark when durable jobs are enabled.
    recorder = _build_event_recorder(settings)
    return Platform(
        profile=profile,
        state=_build_state(settings),
        storage=_build_storage(settings),
        events=_build_events(settings, recorder),
        lock=_build_lock(settings),
        clock=SystemClock(),
        geocoding=_build_geocoding(settings),
        recorder=recorder,
    )

Runtime handle

jasil.runtime

Process-wide access to the assembled Platform.

The composition root publishes the assembled Platform here at startup via :func:set_active_platform, and every caller — request handlers, the scheduler, the durable-job worker, a background thread — resolves the one instance through :func:get_active_platform (or :func:get_state). Stores that must work in any context resolve their provider lazily through :func:get_state.

get_active_platform

get_active_platform()

Return the process-wide platform, or fail if startup has not run.

Returns:

Type	Description
Platform	The active platform substrate.

Raises:

Type	Description
RuntimeError	When no platform has been published yet.

Source code in jasil/runtime.py

def get_active_platform() -> "Platform":
    """Return the process-wide platform, or fail if startup has not run.

    Returns:
        The active platform substrate.

    Raises:
        RuntimeError: When no platform has been published yet.
    """
    if _active_platform is None:
        raise RuntimeError("Platform is not initialized; build_platform must run at startup before this is used.")
    return _active_platform

get_state

get_state()

Return the process-wide ephemeral-state provider.

Returns:

Type	Description
StateProvider	The active StateProvider.

Raises:

Type	Description
RuntimeError	When no platform has been published yet.

Source code in jasil/runtime.py

def get_state() -> "StateProvider":
    """Return the process-wide ephemeral-state provider.

    Returns:
        The active ``StateProvider``.

    Raises:
        RuntimeError: When no platform has been published yet.
    """
    return get_active_platform().state

set_active_platform

set_active_platform(platform)

Publish the assembled platform for process-wide access.

Called once from lifespan startup after build_platform.

Parameters:

Name	Type	Description	Default
platform	Platform	The assembled platform substrate.	required

Source code in jasil/runtime.py

def set_active_platform(platform: "Platform") -> None:
    """Publish the assembled platform for process-wide access.

    Called once from lifespan startup after ``build_platform``.

    Args:
        platform: The assembled platform substrate.
    """
    global _active_platform
    _active_platform = platform

FastAPI dependencies

jasil.deps

FastAPI dependencies exposing the platform providers to routes and handlers.

The composition root attaches the assembled Platform to app.state.platform at startup; these thin dependencies read it back so routes depend on providers (via Depends) rather than importing backends.

get_clock

get_clock(request)

Return the clock provider.

Source code in jasil/deps.py

def get_clock(request: Request) -> ClockProvider:
    """Return the clock provider."""
    return request.app.state.platform.clock

get_events

get_events(request)

Return the event-bus provider.

Source code in jasil/deps.py

def get_events(request: Request) -> EventBusProvider:
    """Return the event-bus provider."""
    return request.app.state.platform.events

get_lock

get_lock(request)

Return the coordination-lock provider.

Source code in jasil/deps.py

def get_lock(request: Request) -> LockProvider:
    """Return the coordination-lock provider."""
    return request.app.state.platform.lock

get_platform

get_platform(request)

Return the process-wide Platform from application state.

Source code in jasil/deps.py

def get_platform(request: Request) -> Platform:
    """Return the process-wide ``Platform`` from application state."""
    return request.app.state.platform

get_state

get_state(request)

Return the ephemeral-state provider.

Source code in jasil/deps.py

def get_state(request: Request) -> StateProvider:
    """Return the ephemeral-state provider."""
    return request.app.state.platform.state

get_storage

get_storage(request)

Return the blob-storage provider.

Source code in jasil/deps.py

def get_storage(request: Request) -> StorageProvider:
    """Return the blob-storage provider."""
    return request.app.state.platform.storage

Durable jobs

jasil.jobs.registry

Registry mapping a durable subscriber_id to its handler and event type.

A durable subscriber declares a stable id (independent of its Python module path) and the event type it reacts to, and registers a handler here. The relay uses the event-type mapping to fan an event out into one job per subscriber; the worker uses the id mapping to resolve a claimed job back to its handler. The same code therefore runs subscribers in-process or out-of-process in a worker.

JobHandlerRegistry

A process-local registry of durable subscribers.

Source code in jasil/jobs/registry.py

class JobHandlerRegistry:
    """A process-local registry of durable subscribers."""

    def __init__(self) -> None:
        self._handlers: dict[str, JobHandler] = {}
        self._by_event_type: dict[str, list[str]] = defaultdict(list)

    def register(self, event_type: str, subscriber_id: str, handler: JobHandler) -> None:
        """
        Register (or replace) a durable subscriber.

        Args:
            event_type: The domain-event channel the subscriber reacts to.
            subscriber_id: The stable durable-subscriber identifier.
            handler: The callable that processes an event for this subscriber.

        Returns:
            None.

        Raises:
            ValueError: When ``event_type`` or ``subscriber_id`` is longer than
                the ``processing_jobs`` column it is written to. Checked at
                registration — startup — rather than when the relay first tries
                to enqueue a job, where the failure would be far from its cause.
        """
        check_length(event_type, field="event_type", limit=MAX_EVENT_TYPE_LENGTH)
        check_length(subscriber_id, field="subscriber_id", limit=MAX_SUBSCRIBER_ID_LENGTH)
        self._handlers[subscriber_id] = handler
        if subscriber_id not in self._by_event_type[event_type]:
            self._by_event_type[event_type].append(subscriber_id)

    def get(self, subscriber_id: str) -> JobHandler | None:
        """
        Look up the handler for a subscriber id.

        Args:
            subscriber_id: The durable-subscriber identifier to resolve.

        Returns:
            The registered handler, or ``None`` when none is registered.
        """
        return self._handlers.get(subscriber_id)

    def subscribers_for(self, event_type: str) -> tuple[str, ...]:
        """
        Return the durable subscriber ids registered for an event type.

        Args:
            event_type: The domain-event channel to fan out.

        Returns:
            The subscriber ids, in registration order.
        """
        return tuple(self._by_event_type.get(event_type, ()))

    def subscriber_ids(self) -> frozenset[str]:
        """
        Return every registered subscriber id, across all event types.

        Returns:
            The registered durable-subscriber ids.
        """
        return frozenset(self._handlers)

    def clear(self) -> None:
        """Remove every registration (used to reset state between tests)."""
        self._handlers.clear()
        self._by_event_type.clear()

clear

clear()

Remove every registration (used to reset state between tests).

Source code in jasil/jobs/registry.py

def clear(self) -> None:
    """Remove every registration (used to reset state between tests)."""
    self._handlers.clear()
    self._by_event_type.clear()

get

get(subscriber_id)

Look up the handler for a subscriber id.

Parameters:

Name	Type	Description	Default
subscriber_id	str	The durable-subscriber identifier to resolve.	required

Returns:

Type	Description
JobHandler | None	The registered handler, or None when none is registered.

Source code in jasil/jobs/registry.py

def get(self, subscriber_id: str) -> JobHandler | None:
    """
    Look up the handler for a subscriber id.

    Args:
        subscriber_id: The durable-subscriber identifier to resolve.

    Returns:
        The registered handler, or ``None`` when none is registered.
    """
    return self._handlers.get(subscriber_id)

register

register(event_type, subscriber_id, handler)

Register (or replace) a durable subscriber.

Parameters:

Name	Type	Description	Default
event_type	str	The domain-event channel the subscriber reacts to.	required
subscriber_id	str	The stable durable-subscriber identifier.	required
handler	JobHandler	The callable that processes an event for this subscriber.	required

Returns:

Type	Description
None	None.

Raises:

Type	Description
ValueError	When event_type or subscriber_id is longer than the processing_jobs column it is written to. Checked at registration — startup — rather than when the relay first tries to enqueue a job, where the failure would be far from its cause.

Source code in jasil/jobs/registry.py

def register(self, event_type: str, subscriber_id: str, handler: JobHandler) -> None:
    """
    Register (or replace) a durable subscriber.

    Args:
        event_type: The domain-event channel the subscriber reacts to.
        subscriber_id: The stable durable-subscriber identifier.
        handler: The callable that processes an event for this subscriber.

    Returns:
        None.

    Raises:
        ValueError: When ``event_type`` or ``subscriber_id`` is longer than
            the ``processing_jobs`` column it is written to. Checked at
            registration — startup — rather than when the relay first tries
            to enqueue a job, where the failure would be far from its cause.
    """
    check_length(event_type, field="event_type", limit=MAX_EVENT_TYPE_LENGTH)
    check_length(subscriber_id, field="subscriber_id", limit=MAX_SUBSCRIBER_ID_LENGTH)
    self._handlers[subscriber_id] = handler
    if subscriber_id not in self._by_event_type[event_type]:
        self._by_event_type[event_type].append(subscriber_id)

subscriber_ids

subscriber_ids()

Return every registered subscriber id, across all event types.

Returns:

Type	Description
frozenset[str]	The registered durable-subscriber ids.

Source code in jasil/jobs/registry.py

def subscriber_ids(self) -> frozenset[str]:
    """
    Return every registered subscriber id, across all event types.

    Returns:
        The registered durable-subscriber ids.
    """
    return frozenset(self._handlers)

subscribers_for

subscribers_for(event_type)

Return the durable subscriber ids registered for an event type.

Parameters:

Name	Type	Description	Default
event_type	str	The domain-event channel to fan out.	required

Returns:

Type	Description
tuple[str, ...]	The subscriber ids, in registration order.

Source code in jasil/jobs/registry.py

def subscribers_for(self, event_type: str) -> tuple[str, ...]:
    """
    Return the durable subscriber ids registered for an event type.

    Args:
        event_type: The domain-event channel to fan out.

    Returns:
        The subscriber ids, in registration order.
    """
    return tuple(self._by_event_type.get(event_type, ()))

jasil.jobs.reconciliation

The reconciliation-net contract every durable subscriber is held to.

A durable subscriber reacts to an event to derive state. Delivery is at-least-once but never guaranteed: a Redis-Streams consumer can drop a message, a provider can be briefly down, and some write paths publish no event at all (a bulk import that persists rows directly). So a subscriber that writes durable derived state must ship a scheduled backfill that re-derives whatever the create path missed.

The vocabulary lives in the substrate rather than in whichever module happened to need it first. A module owning the type would force every other module to import it just to declare a net — a dependency between two bounded contexts for the sake of a shared word. Owning it here is what lets every module declare its nets without depending on any other, and what lets one conformance test (:func:assert_nets_complete) hold them all to it.

DurableSubscriberNet dataclass

A durable subscriber's reconciliation net, or a documented exemption.

Exactly one of backfill / exempt_reason is set. Neither is not an option: a subscriber with no net and no stated reason is one whose derived state silently goes missing, which is the failure this declaration exists to make impossible to introduce by omission.

Attributes:

Name	Type	Description
subscriber_id	str	The stable durable-subscriber id (as registered on the :class:jasil.jobs.registry.JobHandlerRegistry).
backfill	Callable[[], None] | None	The scheduled, argument-free backfill that re-derives anything the create-path handler missed, or None when the subscriber is exempt (its derived state is transient / self-healing).
exempt_reason	str | None	Why no backfill is required, when backfill is None. Must be set for exempt subscribers and unset otherwise.

Source code in jasil/jobs/reconciliation.py

@dataclass(frozen=True)
class DurableSubscriberNet:
    """A durable subscriber's reconciliation net, or a documented exemption.

    Exactly one of ``backfill`` / ``exempt_reason`` is set. Neither is not an
    option: a subscriber with no net and no stated reason is one whose derived
    state silently goes missing, which is the failure this declaration exists to
    make impossible to introduce by omission.

    Attributes:
        subscriber_id: The stable durable-subscriber id (as registered on the
            :class:`jasil.jobs.registry.JobHandlerRegistry`).
        backfill: The scheduled, argument-free backfill that re-derives anything
            the create-path handler missed, or ``None`` when the subscriber is
            exempt (its derived state is transient / self-healing).
        exempt_reason: Why no backfill is required, when ``backfill`` is ``None``.
            Must be set for exempt subscribers and unset otherwise.
    """

    subscriber_id: str
    backfill: Callable[[], None] | None
    exempt_reason: str | None = None

    def __post_init__(self) -> None:
        """
        Reject a declaration that sets both fields, or neither.

        Returns:
            None.

        Raises:
            ValueError: When ``backfill`` and ``exempt_reason`` are both set or
                both unset.
        """
        if self.backfill is not None and self.exempt_reason is not None:
            raise ValueError(
                f"durable subscriber {self.subscriber_id!r} declares both a backfill and an exemption; "
                "an exemption means there is nothing to run"
            )
        if self.backfill is None and self.exempt_reason is None:
            raise ValueError(
                f"durable subscriber {self.subscriber_id!r} declares no reconciliation net; "
                "give it a backfill, or an exempt_reason saying why its derived state needs none"
            )

__post_init__

__post_init__()

Reject a declaration that sets both fields, or neither.

Returns:

Type	Description
None	None.

Raises:

Type	Description
ValueError	When backfill and exempt_reason are both set or both unset.

Source code in jasil/jobs/reconciliation.py

def __post_init__(self) -> None:
    """
    Reject a declaration that sets both fields, or neither.

    Returns:
        None.

    Raises:
        ValueError: When ``backfill`` and ``exempt_reason`` are both set or
            both unset.
    """
    if self.backfill is not None and self.exempt_reason is not None:
        raise ValueError(
            f"durable subscriber {self.subscriber_id!r} declares both a backfill and an exemption; "
            "an exemption means there is nothing to run"
        )
    if self.backfill is None and self.exempt_reason is None:
        raise ValueError(
            f"durable subscriber {self.subscriber_id!r} declares no reconciliation net; "
            "give it a backfill, or an exempt_reason saying why its derived state needs none"
        )

assert_nets_complete

assert_nets_complete(nets, *, registry)

Fail unless every registered durable subscriber declares a net.

Parameters:

Name	Type	Description	Default
nets	Iterable[DurableSubscriberNet]	The reconciliation nets declared across every module.	required
registry	JobHandlerRegistry	The registry holding the durable subscribers to check.	required

Returns:

Type	Description
None	None.

Raises:

Type	Description
AssertionError	When a registered subscriber declares no net. This is a host's conformance test in one call, where an assertion is the idiom.

Source code in jasil/jobs/reconciliation.py

def assert_nets_complete(
    nets: Iterable[DurableSubscriberNet],
    *,
    registry: JobHandlerRegistry,
) -> None:
    """
    Fail unless every registered durable subscriber declares a net.

    Args:
        nets: The reconciliation nets declared across every module.
        registry: The registry holding the durable subscribers to check.

    Returns:
        None.

    Raises:
        AssertionError: When a registered subscriber declares no net. This is a
            host's conformance test in one call, where an assertion is the idiom.
    """
    missing = undeclared_subscribers(nets, registry=registry)
    if missing:
        raise AssertionError(
            "durable subscribers with no declared reconciliation net: "
            + ", ".join(sorted(missing))
            + " — declare a DurableSubscriberNet with a backfill, or an exempt_reason"
        )

undeclared_subscribers

undeclared_subscribers(nets, *, registry)

Return the registered subscriber ids that no net accounts for.

Parameters:

Name	Type	Description	Default
nets	Iterable[DurableSubscriberNet]	The reconciliation nets declared across every module.	required
registry	JobHandlerRegistry	The registry holding the durable subscribers to check — normally the process-wide one, with every subscriber module imported.	required

Returns:

Type	Description
frozenset[str]	The subscriber ids present in the registry but absent from nets.
frozenset[str]	Empty when every durable subscriber is accounted for.

Source code in jasil/jobs/reconciliation.py

def undeclared_subscribers(
    nets: Iterable[DurableSubscriberNet],
    *,
    registry: JobHandlerRegistry,
) -> frozenset[str]:
    """
    Return the registered subscriber ids that no net accounts for.

    Args:
        nets: The reconciliation nets declared across every module.
        registry: The registry holding the durable subscribers to check —
            normally the process-wide one, with every subscriber module imported.

    Returns:
        The subscriber ids present in the registry but absent from ``nets``.
        Empty when every durable subscriber is accounted for.
    """
    return registry.subscriber_ids() - {net.subscriber_id for net in nets}

jasil.jobs.runner

The job runner — claims durable jobs and executes their subscribers.

Records outcomes to processing_jobs only (the per-subscriber execution state), never to event_log (which stays a one-row-per-event publication log). A successful run marks the job completed; a failure reschedules it with backoff, or dead-letters it once the attempt ceiling is reached.

ClaimedJob dataclass

A detached snapshot of a claimed job, safe to use after its session closes.

Attributes:

Name	Type	Description
id	str	The job id.
event_id	str	The originating envelope event_id.
event_type	str	The domain-event channel.
subscriber_id	str	The durable subscriber to run.
source	str	Where the originating event came from.
payload	dict	The domain payload.
metadata	dict | None	Correlation context, if any.
attempts	int	The attempt number this run represents.
timestamp	str	ISO-8601 enqueue time, used to rebuild the event envelope.
schema_version	int	The payload-shape version carried from the envelope, so the handler can upgrade or refuse a payload written by another build.

Source code in jasil/jobs/runner.py

@dataclass(frozen=True)
class ClaimedJob:
    """A detached snapshot of a claimed job, safe to use after its session closes.

    Attributes:
        id: The job id.
        event_id: The originating envelope event_id.
        event_type: The domain-event channel.
        subscriber_id: The durable subscriber to run.
        source: Where the originating event came from.
        payload: The domain payload.
        metadata: Correlation context, if any.
        attempts: The attempt number this run represents.
        timestamp: ISO-8601 enqueue time, used to rebuild the event envelope.
        schema_version: The payload-shape version carried from the envelope, so
            the handler can upgrade or refuse a payload written by another build.
    """

    id: str
    event_id: str
    event_type: str
    subscriber_id: str
    source: str
    payload: dict
    metadata: dict | None
    attempts: int
    timestamp: str
    schema_version: int

JobRunner

Claims a batch of due jobs and runs each one's registered subscriber.

Source code in jasil/jobs/runner.py

class JobRunner:
    """Claims a batch of due jobs and runs each one's registered subscriber."""

    def __init__(
        self,
        *,
        registry: JobHandlerRegistry,
        clock: ClockProvider,
        session_factory: Callable[[], Session],
        worker_id: str,
        lease_seconds: int,
        batch_size: int,
        backoff_base_seconds: float,
        backoff_max_seconds: float,
    ) -> None:
        self._registry = registry
        self._clock = clock
        self._session_factory = session_factory
        self._worker_id = worker_id
        self._lease_seconds = lease_seconds
        self._batch_size = batch_size
        self._backoff_base_seconds = backoff_base_seconds
        self._backoff_max_seconds = backoff_max_seconds

    def run_once(self) -> int:
        """
        Claim and process one batch of due jobs.

        Returns:
            The number of jobs processed in this batch (0 when none were due).
        """
        now = self._clock.now()
        with self._session_factory() as db:
            claimed = jobs_crud.claim_jobs(
                worker_id=self._worker_id,
                limit=self._batch_size,
                lease_seconds=self._lease_seconds,
                now=now,
                db=db,
            )
            snapshots = [self._snapshot(job) for job in claimed]
        for snapshot in snapshots:
            # A DB error finishing one job must not abort the rest of the batch;
            # its lease simply expires and the reaper requeues it.
            try:
                self._run_job(snapshot)
            except Exception as error:
                logger.error(
                    "Durable job could not be finalized",
                    exc_info=error,
                    extra={"job_id": snapshot.id, "subscriber": snapshot.subscriber_id},
                )
        return len(snapshots)

    def reap_once(self) -> int:
        """
        Requeue (or dead-letter) jobs whose lease expired.

        Returns:
            The number of jobs reclaimed.
        """
        now = self._clock.now()
        with self._session_factory() as db:
            return jobs_crud.reclaim_expired_leases(now=now, db=db)

    def _run_job(self, job: ClaimedJob) -> None:
        handler = self._registry.get(job.subscriber_id)
        event = self._event_from(job)
        try:
            if handler is None:
                raise LookupError(f"no durable handler registered for subscriber_id {job.subscriber_id!r}")
            handler(event)
        except Exception as error:
            self._fail(job, error)
            return
        now = self._clock.now()
        with self._session_factory() as db:
            jobs_crud.mark_job_completed(job.id, now=now, db=db)

    def _fail(self, job: ClaimedJob, error: Exception) -> None:
        now = self._clock.now()
        with self._session_factory() as db:
            status = jobs_crud.mark_job_failed(
                job.id,
                str(error),
                base_seconds=self._backoff_base_seconds,
                max_seconds=self._backoff_max_seconds,
                now=now,
                db=db,
            )
        level = logging.ERROR if status == jobs_crud.STATUS_DEAD_LETTER else logging.WARNING
        logger.log(
            level,
            "Durable job failed",
            exc_info=error,
            extra={
                "job_id": job.id,
                "subscriber": job.subscriber_id,
                "event_type": job.event_type,
                "attempts": job.attempts,
                "job_status": status or "unknown",
            },
        )

    def _event_from(self, job: ClaimedJob) -> Event:
        return Event(
            event_id=job.event_id,
            event_type=job.event_type,
            source=job.source,
            timestamp=job.timestamp,
            payload=job.payload,
            metadata=job.metadata or {},
            retry_count=job.attempts,
            schema_version=job.schema_version,
        )

    def _snapshot(self, job: ProcessingJob) -> ClaimedJob:
        return ClaimedJob(
            id=job.id,
            event_id=job.event_id,
            event_type=job.event_type,
            subscriber_id=job.subscriber_id,
            source=job.source,
            payload=dict(job.payload),
            metadata=dict(job.job_metadata) if job.job_metadata else None,
            attempts=job.attempts,
            timestamp=as_utc(job.created_at).isoformat(),
            schema_version=job.schema_version,
        )

reap_once

reap_once()

Requeue (or dead-letter) jobs whose lease expired.

Returns:

Type	Description
int	The number of jobs reclaimed.

Source code in jasil/jobs/runner.py

def reap_once(self) -> int:
    """
    Requeue (or dead-letter) jobs whose lease expired.

    Returns:
        The number of jobs reclaimed.
    """
    now = self._clock.now()
    with self._session_factory() as db:
        return jobs_crud.reclaim_expired_leases(now=now, db=db)

run_once

run_once()

Claim and process one batch of due jobs.

Returns:

Type	Description
int	The number of jobs processed in this batch (0 when none were due).

Source code in jasil/jobs/runner.py

def run_once(self) -> int:
    """
    Claim and process one batch of due jobs.

    Returns:
        The number of jobs processed in this batch (0 when none were due).
    """
    now = self._clock.now()
    with self._session_factory() as db:
        claimed = jobs_crud.claim_jobs(
            worker_id=self._worker_id,
            limit=self._batch_size,
            lease_seconds=self._lease_seconds,
            now=now,
            db=db,
        )
        snapshots = [self._snapshot(job) for job in claimed]
    for snapshot in snapshots:
        # A DB error finishing one job must not abort the rest of the batch;
        # its lease simply expires and the reaper requeues it.
        try:
            self._run_job(snapshot)
        except Exception as error:
            logger.error(
                "Durable job could not be finalized",
                exc_info=error,
                extra={"job_id": snapshot.id, "subscriber": snapshot.subscriber_id},
            )
    return len(snapshots)

jasil.jobs.service

Wiring for durable job processing — worker lifecycle and scheduled maintenance.

Consumed by main (start/stop the in-process worker) and the scheduler (relay the outbox into jobs; reap expired leases). The relay, reaper, and worker all run on every process and coordinate through SELECT ... FOR UPDATE SKIP LOCKED plus the idempotent job fan-out (dedup on event_id + subscriber_id), so replicas scale horizontally without a single-runner lock — duplicate work is skipped or deduplicated rather than serialized. All of this is inert unless JOBS_ENABLED is set.

build_runner

build_runner()

Build a :class:JobRunner from settings and the active platform.

Returns:

Type	Description
JobRunner	A runner wired to the durable-subscriber registry, the platform clock,
JobRunner	and the main-database session factory.

Source code in jasil/jobs/service.py

def build_runner() -> JobRunner:
    """
    Build a :class:`JobRunner` from settings and the active platform.

    Returns:
        A runner wired to the durable-subscriber registry, the platform clock,
        and the main-database session factory.
    """
    jobs = get_settings().jobs
    platform = platform_runtime.get_active_platform()
    return JobRunner(
        registry=jobs_registry.registry,
        clock=platform.clock,
        session_factory=jasil_orm.get_sessionmaker(),
        worker_id=_worker_id(),
        lease_seconds=jobs.lease_seconds,
        batch_size=jobs.batch_size,
        backoff_base_seconds=jobs.backoff_base_seconds,
        backoff_max_seconds=jobs.backoff_max_seconds,
    )

reap_expired_jobs_scheduled

reap_expired_jobs_scheduled()

Requeue or dead-letter jobs with expired leases.

Runs on every replica; reclaim_expired_leases uses FOR UPDATE SKIP LOCKED so concurrent reapers reclaim disjoint rows.

Source code in jasil/jobs/service.py

def reap_expired_jobs_scheduled() -> None:
    """Requeue or dead-letter jobs with expired leases.

    Runs on every replica; ``reclaim_expired_leases`` uses ``FOR UPDATE SKIP
    LOCKED`` so concurrent reapers reclaim disjoint rows.
    """
    platform = platform_runtime.get_active_platform()
    with jasil_orm.get_sessionmaker()() as db:
        reclaimed = jobs_crud.reclaim_expired_leases(now=platform.clock.now(), db=db)
    if reclaimed:
        logger.info(f"Reaped {reclaimed} expired job lease(s)")

relay_outbox_scheduled

relay_outbox_scheduled()

Relay the outbox into per-subscriber jobs.

Runs on every replica; FOR UPDATE SKIP LOCKED gives each relayer a disjoint batch and the idempotent fan-out dedups any overlap, so no single-runner lock is needed.

Source code in jasil/jobs/service.py

def relay_outbox_scheduled() -> None:
    """Relay the outbox into per-subscriber jobs.

    Runs on every replica; ``FOR UPDATE SKIP LOCKED`` gives each relayer a
    disjoint batch and the idempotent fan-out dedups any overlap, so no
    single-runner lock is needed.
    """
    jobs = get_settings().jobs
    platform = platform_runtime.get_active_platform()
    for _ in range(_MAX_RELAY_BATCHES):
        relayed = jobs_relay.relay_outbox_once(
            registry=jobs_registry.registry,
            clock=platform.clock,
            session_factory=jasil_orm.get_sessionmaker(),
            max_attempts=jobs.max_attempts,
            batch_size=jobs.batch_size,
        )
        if relayed == 0:
            break

schedule_job_maintenance

schedule_job_maintenance(scheduler)

Register the recurring relay and reaper jobs on the scheduler.

Parameters:

Name	Type	Description	Default
scheduler	AsyncIOScheduler	The application scheduler to register the jobs on.	required

Returns:

Type	Description
None	None.

Source code in jasil/jobs/service.py

def schedule_job_maintenance(scheduler: AsyncIOScheduler) -> None:
    """
    Register the recurring relay and reaper jobs on the scheduler.

    Args:
        scheduler: The application scheduler to register the jobs on.

    Returns:
        None.
    """
    scheduler.add_job(
        relay_outbox_scheduled,
        "interval",
        seconds=_RELAY_INTERVAL_SECONDS,
        id=_RELAY_JOB_ID,
        replace_existing=True,
    )
    scheduler.add_job(
        reap_expired_jobs_scheduled,
        "interval",
        seconds=_REAP_INTERVAL_SECONDS,
        id=_REAP_JOB_ID,
        replace_existing=True,
    )
    logger.info("Scheduled durable-job outbox relay and lease reaper")

start_job_worker

start_job_worker()

Start the in-process job worker (idempotent).

Source code in jasil/jobs/service.py

def start_job_worker() -> None:
    """Start the in-process job worker (idempotent)."""
    global _worker
    if _worker is not None:
        return
    _worker = BackgroundWorker(build_runner(), poll_interval_seconds=get_settings().jobs.poll_interval_seconds)
    _worker.start()
    logger.info("Durable job worker started")

stop_job_worker

stop_job_worker()

Stop the in-process job worker if it is running.

Source code in jasil/jobs/service.py

def stop_job_worker() -> None:
    """Stop the in-process job worker if it is running."""
    global _worker
    if _worker is None:
        return
    _worker.stop()
    _worker = None
    logger.info("Durable job worker stopped")

Retention

jasil.retention

Scheduled retention pruning for the substrate's append-only bookkeeping tables.

The event-log and durable-job tables are append-only: every event writes an event_log row, the relay stamps event_outbox rows, and each (event, subscriber) pair produces a processing_jobs row. Left alone they grow without bound, fastest in whichever subsystem publishes most. This module prunes rows past their retention window on a schedule (one window for the event_log trail, another for the durable-job tables — each configured and disabled independently), deleting only what is safe to lose:

• every event_log row (best-effort, safe-to-lose observability trail),
• relayed event_outbox rows (already fanned out into jobs), and
• completed processing_jobs rows.

In-flight and human-actionable rows are never touched: unrelayed outbox rows (pending relay), pending / claimed jobs (in-flight work), and dead_letter jobs (kept for operator review). It runs single-runner across replicas via the platform LockProvider and is inert when both windows are <= 0 (retention disabled — keep every row forever).

prune_expired_records

prune_expired_records()

Prune substrate bookkeeping rows older than their retention windows.

Scheduled daily (and once at startup). Each window is applied independently: EVENT_LOG_RETENTION_DAYS gates the event_log trail and JOBS_RETENTION_DAYS gates the durable-job tables. No-ops when both are disabled (<= 0) or when another replica already holds the prune lock.

Returns:

Type	Description
None	None.

Source code in jasil/retention.py

def prune_expired_records() -> None:
    """Prune substrate bookkeeping rows older than their retention windows.

    Scheduled daily (and once at startup). Each window is applied independently:
    ``EVENT_LOG_RETENTION_DAYS`` gates the event_log trail and
    ``JOBS_RETENTION_DAYS`` gates the durable-job tables. No-ops when both are
    disabled (``<= 0``) or when another replica already holds the prune lock.

    Returns:
        None.
    """
    settings = get_settings()
    event_log_days = settings.event_log.retention_days
    jobs_days = settings.jobs.retention_days
    if event_log_days <= 0 and jobs_days <= 0:
        return

    platform = platform_runtime.get_active_platform()
    with platform.lock.try_acquire(_PRUNE_LOCK_NAME) as acquired:
        if not acquired:
            logger.debug("Retention prune: another replica holds the lock; skipping")
            return
        _run_prune(platform.clock.now(), event_log_days, jobs_days)

jasil.pruning

Bounded batch deletes shared by the substrate's prunable tables.

Every table :mod:jasil.retention prunes (event_log, relayed event_outbox rows, completed processing_jobs) is deleted the same way, and only the model and the filter differ. The batching is what keeps each delete transaction short, so a prune pass never holds locks on a hot table long enough to block the relay or the worker.

bounded_delete

bounded_delete(
    model, *conditions, db, batch_size=PRUNE_BATCH_SIZE
)

Delete the rows matching conditions in bounded, committed batches.

Parameters:

Name	Type	Description	Default
model	type[Any]	Mapped class to delete from; must expose an id column.	required
*conditions	ColumnExpressionArgument[bool]	Filters selecting the rows that are safe to prune.	()
db	Session	Active database session.	required
batch_size	int	Maximum rows deleted per batch.	PRUNE_BATCH_SIZE

Returns:

Type	Description
int	The total number of rows deleted.

Source code in jasil/pruning.py

def bounded_delete(
    model: type[Any],
    *conditions: ColumnExpressionArgument[bool],
    db: Session,
    batch_size: int = PRUNE_BATCH_SIZE,
) -> int:
    """
    Delete the rows matching ``conditions`` in bounded, committed batches.

    Args:
        model: Mapped class to delete from; must expose an ``id`` column.
        *conditions: Filters selecting the rows that are safe to prune.
        db: Active database session.
        batch_size: Maximum rows deleted per batch.

    Returns:
        The total number of rows deleted.
    """
    total = 0
    for _ in range(PRUNE_MAX_BATCHES):
        id_stmt = select(model.id).where(*conditions).limit(batch_size)
        if supports_skip_locked(db.bind):  # pragma: no cover - server-side locking, not exercised on SQLite
            # Concurrent prunes step over each other's claimed page rather
            # than blocking on it.
            id_stmt = id_stmt.with_for_update(skip_locked=True)
        ids = list(db.execute(id_stmt).scalars().all())
        if not ids:
            break
        db.execute(delete(model).where(model.id.in_(ids)))
        db.commit()
        total += len(ids)
        if len(ids) < batch_size:
            break
    return total

Migrations

jasil.migrations

Alembic migrations for JASIL's companion tables.

JASIL maps its tables into the host's declarative registry (see :mod:jasil.orm), so its tables and the host's live in one database. To let JASIL evolve its schema without owning the host's Alembic history, these migrations run on their own version table (jasil_alembic_version) and are scoped to JASIL's own tables — the host's tables are never touched.

Requires the optional jasil[migrations] extra (Alembic). import jasil never pulls this in; import it explicitly::

import jasil.orm as jasil_orm
from jasil import migrations

jasil_orm.map_models(Base)                  # the metadata must exist first
migrations.upgrade(engine)                  # create/upgrade JASIL's tables
# migrations.stamp(engine)                  # existing DB already at head
# migrations.verify_schema_current(engine)  # fail fast if not migrated

Hosts that prefer a single, unified Alembic history can instead point their own env.py at their Base.metadata (the base passed to :func:jasil.orm.map_models) and add this package's versions directory to their version_locations — but the self-contained runner here needs no host wiring.

db_revision

db_revision(engine)

Return the JASIL migration revision currently recorded in engine.

Source code in jasil/migrations/__init__.py

def db_revision(engine: "Engine") -> str | None:
    """Return the JASIL migration revision currently recorded in ``engine``."""
    _require_alembic()
    from alembic.runtime.migration import MigrationContext

    with engine.connect() as connection:
        context = MigrationContext.configure(connection, opts={"version_table": VERSION_TABLE})
        return context.get_current_revision()

downgrade

downgrade(engine, revision)

Downgrade JASIL's tables to revision ("base" drops them all).

Source code in jasil/migrations/__init__.py

def downgrade(engine: "Engine", revision: str) -> None:
    """Downgrade JASIL's tables to ``revision`` (``"base"`` drops them all)."""
    _run(engine, "downgrade", revision)

head_revision

head_revision()

Return the newest revision shipped in this package.

Source code in jasil/migrations/__init__.py

def head_revision() -> str | None:
    """Return the newest revision shipped in this package."""
    _require_alembic()
    from alembic.script import ScriptDirectory

    return ScriptDirectory.from_config(_config()).get_current_head()

jasil_include_object

jasil_include_object(
    obj, name, type_, reflected, compare_to
)

Alembic include_object hook scoping operations to JASIL's tables.

Excludes every host table sharing the registry, so JASIL's migrations and autogenerate never add, drop, or diff them.

Source code in jasil/migrations/__init__.py

def jasil_include_object(obj: Any, name: str | None, type_: str, reflected: bool, compare_to: Any) -> bool:
    """Alembic ``include_object`` hook scoping operations to JASIL's tables.

    Excludes every host table sharing the registry, so JASIL's migrations and
    autogenerate never add, drop, or diff them.
    """
    from jasil.orm import jasil_table_names

    if type_ == "table":
        return name in jasil_table_names()
    return True

stamp

stamp(engine, revision='head')

Record revision without running migrations.

Use on an existing deployment whose JASIL tables were created with Base.metadata.create_all (or an older release): stamping marks it as being at head so future :func:upgrade calls apply only new revisions.

Source code in jasil/migrations/__init__.py

def stamp(engine: "Engine", revision: str = "head") -> None:
    """Record ``revision`` without running migrations.

    Use on an existing deployment whose JASIL tables were created with
    ``Base.metadata.create_all`` (or an older release): stamping marks it as
    being at head so future :func:`upgrade` calls apply only new revisions.
    """
    _run(engine, "stamp", revision)

upgrade

upgrade(engine, revision='head')

Create or upgrade JASIL's tables to revision (default head).

Source code in jasil/migrations/__init__.py

def upgrade(engine: "Engine", revision: str = "head") -> None:
    """Create or upgrade JASIL's tables to ``revision`` (default ``head``)."""
    _run(engine, "upgrade", revision)

verify_schema_current

verify_schema_current(engine)

Raise if the database is not migrated to the packaged head revision.

A fail-fast startup check: call it after configuring the engine to catch a forgotten upgrade before the first query does.

Raises:

Type	Description
RuntimeError	If the recorded revision differs from the packaged head.

Source code in jasil/migrations/__init__.py

def verify_schema_current(engine: "Engine") -> None:
    """Raise if the database is not migrated to the packaged head revision.

    A fail-fast startup check: call it after configuring the engine to catch a
    forgotten upgrade before the first query does.

    Raises:
        RuntimeError: If the recorded revision differs from the packaged head.
    """
    head = head_revision()
    current = db_revision(engine)
    if current != head:
        raise RuntimeError(
            "JASIL database schema is out of date "
            f"(database revision={current!r}, expected head={head!r}). "
            "Run jasil.migrations.upgrade(engine) at deploy time, or "
            "jasil.migrations.stamp(engine) if the tables already exist."
        )

Async bridge

jasil.async_bridge

Bridge for dispatching async work onto the main event loop from sync code.

Synchronous code cannot await. Two situations in this backend need to run genuinely async I/O from a synchronous context:

• a synchronous FastAPI route, which Starlette runs on a threadpool worker thread (no running event loop of its own), and
• an in-process event-bus subscriber, which runs inline on whatever thread called publish (a request thread, the scheduler, or a durable-job worker).

When such code must perform async I/O — pushing a message over a live connection is the motivating case — it hands the coroutine to :func:dispatch, which schedules it on the main event loop captured at application startup via :func:asyncio.run_coroutine_threadsafe. This keeps the async work on the one loop that owns the resources (e.g. the websocket connections) while letting the producing code stay fully synchronous.

The main loop is a process-wide handle (like the platform handle in :mod:jasil.runtime), so it lives in a module-level slot set once from the lifespan startup and cleared on shutdown. This module imports nothing from the domain layer or any backend, so it is safe for any module to depend on.

capture_running_loop

capture_running_loop()

Capture the currently running loop as the main loop.

Convenience for the lifespan startup: keeps the asyncio detail here rather than in main. Must be called from within a running event loop.

Returns:

Type	Description
None	None.

Source code in jasil/async_bridge.py

def capture_running_loop() -> None:
    """Capture the currently running loop as the main loop.

    Convenience for the lifespan startup: keeps the ``asyncio`` detail here rather
    than in ``main``. Must be called from within a running event loop.

    Returns:
        None.
    """
    set_main_loop(asyncio.get_running_loop())

dispatch

dispatch(coro)

Schedule coro on the main event loop from any thread; fire-and-forget.

Thread-safe. Intended for synchronous callers (sync routes, in-process subscribers) that need to run async I/O without awaiting it. Failures raised by the coroutine are logged (never surfaced to the caller), so a delivery failure — e.g. a dropped websocket — cannot break the synchronous work that triggered it.

Parameters:

Name	Type	Description	Default
coro	Coroutine[Any, Any, Any]	The coroutine to run on the main loop.	required

Returns:

Name	Type	Description
The	Future[Any] | None	class:concurrent.futures.Future tracking the scheduled coroutine,
	Future[Any] | None	or None when no running main loop is available. In the None case
	Future[Any] | None	the coroutine is closed so it does not emit a "coroutine was never
	Future[Any] | None	awaited" warning.

Source code in jasil/async_bridge.py

def dispatch(coro: Coroutine[Any, Any, Any]) -> Future[Any] | None:
    """Schedule ``coro`` on the main event loop from any thread; fire-and-forget.

    Thread-safe. Intended for synchronous callers (sync routes, in-process
    subscribers) that need to run async I/O without awaiting it. Failures raised
    by the coroutine are logged (never surfaced to the caller), so a delivery
    failure — e.g. a dropped websocket — cannot break the synchronous work that
    triggered it.

    Args:
        coro: The coroutine to run on the main loop.

    Returns:
        The :class:`concurrent.futures.Future` tracking the scheduled coroutine,
        or ``None`` when no running main loop is available. In the ``None`` case
        the coroutine is closed so it does not emit a "coroutine was never
        awaited" warning.
    """
    loop = _main_loop
    if loop is None or loop.is_closed():
        logger.warning("async_bridge.dispatch called with no running main loop; dropping coroutine")
        coro.close()
        return None

    future = asyncio.run_coroutine_threadsafe(coro, loop)

    def _log_failure(completed: Future[Any]) -> None:
        # Runs on the loop thread once the coroutine finishes. Surface any error
        # to the log; a cancelled future has no exception to report.
        if completed.cancelled():
            return
        error = completed.exception()
        if error is None:
            return
        logger.error(
            f"async_bridge dispatched coroutine failed: {type(error).__name__}",
            exc_info=error if isinstance(error, Exception) else None,
        )

    future.add_done_callback(_log_failure)
    return future

get_main_loop

get_main_loop()

Return the captured main event loop, or None if it is not set.

Returns:

Type	Description
AbstractEventLoop | None	The main event loop when the application is running, else None
AbstractEventLoop | None	(e.g. in unit tests, or before startup / after shutdown).

Source code in jasil/async_bridge.py

def get_main_loop() -> asyncio.AbstractEventLoop | None:
    """Return the captured main event loop, or ``None`` if it is not set.

    Returns:
        The main event loop when the application is running, else ``None``
        (e.g. in unit tests, or before startup / after shutdown).
    """
    return _main_loop

set_main_loop

set_main_loop(loop)

Record (or clear) the main event loop.

Called once from the lifespan startup with the running loop, and again on shutdown with None.

Parameters:

Name	Type	Description	Default
loop	AbstractEventLoop | None	The running main event loop, or None to clear it.	required

Returns:

Type	Description
None	None.

Source code in jasil/async_bridge.py

def set_main_loop(loop: asyncio.AbstractEventLoop | None) -> None:
    """Record (or clear) the main event loop.

    Called once from the lifespan startup with the running loop, and again on
    shutdown with ``None``.

    Args:
        loop: The running main event loop, or ``None`` to clear it.

    Returns:
        None.
    """
    global _main_loop  # single process-wide handle, set once at startup
    _main_loop = loop