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

Composable provider interfaces and complete platform capabilities.

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

Implementations may additionally offer record_queued_in(event, db), which stages the queued row on the caller's session so it commits or rolls back with the outbox row. It is optional and deliberately not part of this protocol (SemVer); the publisher uses it when present and otherwise falls back to :meth:record_queued.

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

    Implementations may additionally offer ``record_queued_in(event, db)``, which
    stages the ``queued`` row on the caller's session so it commits or rolls back
    with the outbox row. It is optional and deliberately not part of this
    protocol (SemVer); the publisher uses it when present and otherwise falls
    back to :meth:`record_queued`.
    """

    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]: ...

MailError

Bases: RuntimeError

A message could not be handed to the mail server.

The message carries a failure category and, when there is one, the SMTP status code — never an address, a subject, a body or server-supplied text, because a job runner persists str(error) and logs it.

Attributes:

Name	Type	Description
status		The SMTP status code, or None when the failure had none (unreachable server, TLS failure, deadline).

Source code in jasil/providers.py

class MailError(RuntimeError):
    """A message could not be handed to the mail server.

    The message carries a failure category and, when there is one, the SMTP
    status code — never an address, a subject, a body or server-supplied text,
    because a job runner persists ``str(error)`` and logs it.

    Attributes:
        status: The SMTP status code, or ``None`` when the failure had none
            (unreachable server, TLS failure, deadline).
    """

    def __init__(self, message: str, *, status: int | None = None) -> None:
        super().__init__(message)
        self.status = status

    @property
    def permanent(self) -> bool:
        """True for a 5xx reply: the server will refuse a retry the same way."""
        return self.status is not None and 500 <= self.status < 600

permanent property

permanent

True for a 5xx reply: the server will refuse a retry the same way.

MailMessage dataclass

One outbound email, already built by the host.

Attributes:

Name	Type	Description
to	Sequence[str]	Recipient addresses, each a bare addr or "Name <addr>".
subject	str	Subject line. Must not contain CR or LF.
text	str | None	Plain-text body. At least one of text / html is required.
html	str | None	HTML body. With text as well, the message is multipart/alternative.
reply_to	str | None	Optional single Reply-To address.

Source code in jasil/providers.py

@dataclass(frozen=True)
class MailMessage:
    """One outbound email, already built by the host.

    Attributes:
        to: Recipient addresses, each a bare ``addr`` or ``"Name <addr>"``.
        subject: Subject line. Must not contain CR or LF.
        text: Plain-text body. At least one of ``text`` / ``html`` is required.
        html: HTML body. With ``text`` as well, the message is
            ``multipart/alternative``.
        reply_to: Optional single ``Reply-To`` address.
    """

    to: Sequence[str]
    subject: str
    text: str | None = None
    html: str | None = None
    reply_to: str | None = None

    def __hash__(self) -> int:
        # ``to`` is usually a list, which the generated hash cannot digest.
        recipients = (self.to,) if isinstance(self.to, str) else tuple(self.to)
        return hash((recipients, self.subject, self.text, self.html, self.reply_to))

MailNotConfiguredError

Bases: MailError

Mail was sent while no backend is configured.

Source code in jasil/providers.py

class MailNotConfiguredError(MailError):
    """Mail was sent while no backend is configured."""

MailProvider

Bases: Protocol

Outbound email — hand a built message to a mail server.

Unlike geocoding, mail is not best-effort enrichment: a silently dropped message is a bug, and a raised exception is what makes a durable job retry.

send returns the Message-ID the backend assigned, angle brackets included, exactly as it appears in the header.

Raises:

Type	Description
MailError	On any delivery failure (:class:MailNotConfiguredError when no backend is configured).
ValueError	On an invalid message — no recipient, no body, a malformed address, or CR/LF in a header value. A programming error, not a delivery error.

Source code in jasil/providers.py

@runtime_checkable
class MailProvider(Protocol):
    """Outbound email — hand a built message to a mail server.

    Unlike geocoding, mail is not best-effort enrichment: a silently dropped
    message is a bug, and a raised exception is what makes a durable job retry.

    ``send`` returns the ``Message-ID`` the backend assigned, angle brackets
    included, exactly as it appears in the header.

    Raises:
        MailError: On any delivery failure (:class:`MailNotConfiguredError` when
            no backend is configured).
        ValueError: On an invalid message — no recipient, no body, a malformed
            address, or CR/LF in a header value. A programming error, not a
            delivery error.
    """

    def send(self, message: MailMessage) -> str: ...

ObjectStat dataclass

Metadata available without reading an object's content.

content_type and etag are optional because portable local filesystems expose neither. An ETag is an opaque backend validator, not necessarily a digest of the object bytes.

Source code in jasil/providers.py

@dataclass(frozen=True)
class ObjectStat:
    """Metadata available without reading an object's content.

    ``content_type`` and ``etag`` are optional because portable local
    filesystems expose neither. An ETag is an opaque backend validator, not
    necessarily a digest of the object bytes.
    """

    size: int
    modified_epoch: float
    content_type: str | None = None
    etag: str | None = None

PartRef dataclass

Backend-validated reference to one uploaded part.

validator is an opaque equality token. It may be a digest, an object storage ETag, or another backend value; callers must not interpret it.

Source code in jasil/providers.py

@dataclass(frozen=True)
class PartRef:
    """Backend-validated reference to one uploaded part.

    ``validator`` is an opaque equality token. It may be a digest, an object
    storage ETag, or another backend value; callers must not interpret it.
    """

    part_number: int
    size: int
    validator: str

ResumableUploads

Bases: Protocol

Durable multipart upload lifecycle and abandoned-session cleanup.

upload_part reads source once without seeking or closing it. The required size must exactly match the source; short and long sources raise :class:ValueError without committing that part. A failed attempt may have consumed the source, so retry with a newly opened source and the same session and part number. Limits reported by :class:UploadSession are JASIL's built-in portability limits, shared by every built-in backend so callers never branch on the selected backend.

Source code in jasil/providers.py

@runtime_checkable
class ResumableUploads(Protocol):
    """Durable multipart upload lifecycle and abandoned-session cleanup.

    ``upload_part`` reads ``source`` once without seeking or closing it. The
    required ``size`` must exactly match the source; short and long sources raise
    :class:`ValueError` without committing that part. A failed attempt may have
    consumed the source, so retry with a newly opened source and the same session
    and part number. Limits reported by :class:`UploadSession` are JASIL's
    built-in portability limits, shared by every built-in backend so callers
    never branch on the selected backend.
    """

    def begin_upload(
        self,
        area: str,
        key: str,
        *,
        max_bytes: int | None = None,
        content_type: str | None = None,
    ) -> UploadSession: ...
    def upload_part(
        self,
        session: UploadSession,
        part_number: int,
        source: BinaryIO,
        *,
        size: int,
    ) -> PartRef: ...
    def complete_upload(self, session: UploadSession, parts: Sequence[PartRef]) -> int: ...
    def abort_upload(self, session: UploadSession) -> None: ...
    def cleanup_uploads(self, *, older_than_epoch: float) -> int: ...

ServeFile dataclass

Serve a local file directly, using sendfile or a reverse-proxy mapping.

Source code in jasil/providers.py

@dataclass(frozen=True)
class ServeFile:
    """Serve a local file directly, using sendfile or a reverse-proxy mapping."""

    path: Path

ServeRedirect dataclass

Redirect the client to a backend-generated object URL.

Source code in jasil/providers.py

@dataclass(frozen=True)
class ServeRedirect:
    """Redirect the client to a backend-generated object URL."""

    url: str

ServeStream dataclass

Proxy an object through the host using a read-once binary stream.

Source code in jasil/providers.py

@dataclass(frozen=True)
class ServeStream:
    """Proxy an object through the host using a read-once binary stream."""

    stream: BinaryIO

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

StorageBackendUnavailableError

Bases: RuntimeError

Raised when a storage backend cannot complete an operation.

Source code in jasil/providers.py

class StorageBackendUnavailableError(RuntimeError):
    """Raised when a storage backend cannot complete an operation."""

StorageDelivery

Bases: Protocol

Backend-efficient delivery without coupling to a web framework.

Source code in jasil/providers.py

@runtime_checkable
class StorageDelivery(Protocol):
    """Backend-efficient delivery without coupling to a web framework."""

    def serve(
        self,
        area: str,
        key: str,
        *,
        download_as: str | None = None,
        content_type: str | None = None,
        expires_in: int = 3600,
    ) -> ServePlan: ...
    def url(
        self,
        area: str,
        key: str,
        expires_in: int = 3600,
        *,
        download_as: str | None = None,
        content_type: str | None = None,
    ) -> str: ...

StorageManagement

Bases: Protocol

Object-set mutation and backend readiness operations.

Source code in jasil/providers.py

@runtime_checkable
class StorageManagement(Protocol):
    """Object-set mutation and backend readiness operations."""

    def delete_prefix(self, area: str, prefix: str) -> int: ...
    def copy(self, src_area: str, src_key: str, dst_area: str, dst_key: str) -> None: ...
    def check_writable(self) -> None: ...

StorageObjects

Bases: Protocol

Whole-object persistence, metadata, deletion, and enumeration.

Source code in jasil/providers.py

@runtime_checkable
class StorageObjects(Protocol):
    """Whole-object persistence, metadata, deletion, and enumeration."""

    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 stat(self, area: str, key: str) -> ObjectStat | 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 iter_objects(self, area: str, prefix: str = "") -> Iterator[tuple[str, float]]: ...

StorageProvider

Bases: StorageObjects, StorageStreams, StorageDelivery, StorageManagement, ResumableUploads, Protocol

Complete object-storage capability assembled by :class:Platform.

An area is a domain-owned namespace (e.g. "avatars", "exports") so one backend serves every subsystem. 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.

save and get are the simple whole-object path for small blobs. save_stream and open_stream are the bounded-memory path for large objects. Streaming writes are atomic: exceeding max_bytes raises :class:StorageSizeLimitError and never exposes a partial new object. Streaming reads are read-once and non-seekable on every backend. offset and length select a byte range without requiring the returned stream to seek; a missing object raises :class:FileNotFoundError.

Resumable writes use begin_upload, one or more upload_part calls, then complete_upload or abort_upload. Sessions are portable across processes sharing the configured backend and report their part constraints. cleanup_uploads aborts sessions initiated before a caller-owned cutoff.

list_keys is the convenient materialized listing for small namespaces. iter_objects lazily yields (key, modified_epoch) for reconciliation jobs that may scan millions of objects and need an age guard.

serve selects the backend's efficient delivery primitive without importing a web framework: local storage returns :class:ServeFile, S3 returns :class:ServeRedirect, and a backend with neither capability may return :class:ServeStream. It verifies that the object exists first.

stat returns size and modification time plus backend metadata when available. copy keeps object bytes inside the backend. delete_prefix deletes one non-empty key root and its slash-delimited descendants, not adjacent lexical prefixes ("pkg/1" never selects "pkg/10").

check_writable performs a write-and-delete readiness probe and raises :class:StorageBackendUnavailableError when the configured store cannot persist objects.

expires_in is best-effort, and the one place the backends genuinely differ: s3:// returns a presigned URL that honours expiry and response header overrides, while local:// returns a plain path for the host's own web server and cannot enforce any of them. Do not rely on these controls unless you know the deployment uses object storage; on local disk they are host policy. The local backend logs a warning the first time it is handed a control it cannot honour.

Source code in jasil/providers.py

@runtime_checkable
class StorageProvider(
    StorageObjects,
    StorageStreams,
    StorageDelivery,
    StorageManagement,
    ResumableUploads,
    Protocol,
):
    """Complete object-storage capability assembled by :class:`Platform`.

    An area is a domain-owned namespace (e.g. ``"avatars"``, ``"exports"``) so
    one backend serves every subsystem. 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.

    ``save`` and ``get`` are the simple whole-object path for small blobs.
    ``save_stream`` and ``open_stream`` are the bounded-memory path for large
    objects. Streaming writes are atomic: exceeding ``max_bytes`` raises
    :class:`StorageSizeLimitError` and never exposes a partial new object.
    Streaming reads are read-once and non-seekable on every backend. ``offset``
    and ``length`` select a byte range without requiring the returned stream to
    seek; a missing object raises :class:`FileNotFoundError`.

    Resumable writes use ``begin_upload``, one or more ``upload_part`` calls,
    then ``complete_upload`` or ``abort_upload``. Sessions are portable across
    processes sharing the configured backend and report their part constraints.
    ``cleanup_uploads`` aborts sessions initiated before a caller-owned cutoff.

    ``list_keys`` is the convenient materialized listing for small namespaces.
    ``iter_objects`` lazily yields ``(key, modified_epoch)`` for reconciliation
    jobs that may scan millions of objects and need an age guard.

    ``serve`` selects the backend's efficient delivery primitive without
    importing a web framework: local storage returns :class:`ServeFile`, S3
    returns :class:`ServeRedirect`, and a backend with neither capability may
    return :class:`ServeStream`. It verifies that the object exists first.

    ``stat`` returns size and modification time plus backend metadata when
    available. ``copy`` keeps object bytes inside the backend. ``delete_prefix``
    deletes one non-empty key root and its slash-delimited descendants, not
    adjacent lexical prefixes (``"pkg/1"`` never selects ``"pkg/10"``).

    ``check_writable`` performs a write-and-delete readiness probe and raises
    :class:`StorageBackendUnavailableError` when the configured store cannot
    persist objects.

    ``expires_in`` is **best-effort, and the one place the backends genuinely
    differ**: ``s3://`` returns a presigned URL that honours expiry and response
    header overrides, while ``local://`` returns a plain path for the host's own
    web server and cannot enforce any of them. Do not rely on these controls
    unless you know the deployment uses object storage; on local disk they are
    host policy. The local backend logs a warning the first time it is handed a
    control it cannot honour.
    """

StorageSizeLimitError

Bases: ValueError

Raised when a streaming write exceeds its configured byte limit.

Source code in jasil/providers.py

class StorageSizeLimitError(ValueError):
    """Raised when a streaming write exceeds its configured byte limit."""

StorageStreams

Bases: Protocol

Bounded-memory object reads and writes.

Source code in jasil/providers.py

@runtime_checkable
class StorageStreams(Protocol):
    """Bounded-memory object reads and writes."""

    def save_stream(
        self,
        area: str,
        key: str,
        source: BinaryIO,
        *,
        max_bytes: int | None = None,
        content_type: str | None = None,
    ) -> int: ...
    def open_stream(
        self,
        area: str,
        key: str,
        *,
        offset: int = 0,
        length: int | None = None,
    ) -> BinaryIO: ...

StorageUploadSessionError

Bases: RuntimeError

Raised when a resumable upload session or part reference is not active.

Source code in jasil/providers.py

class StorageUploadSessionError(RuntimeError):
    """Raised when a resumable upload session or part reference is not active."""

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

UploadSession dataclass

Durable handle and portable limits for one resumable upload.

Source code in jasil/providers.py

@dataclass(frozen=True)
class UploadSession:
    """Durable handle and portable limits for one resumable upload."""

    area: str
    key: str
    session_id: str
    max_bytes: int | None
    min_part_size: int
    max_part_size: int
    max_parts: int

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.

payload and metadata are stored and transmitted verbatim: both are written to event_outbox and processing_jobs, to event_log when the trail is enabled, and serialized onto the Redis stream on the redis:// bus — and metadata is additionally logged in full when a best-effort subscriber fails. Neither is size-capped. So carry identifiers, not secrets and not blobs; see the events documentation for the full contract.

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.

    ``payload`` and ``metadata`` are stored and transmitted verbatim: both are
    written to ``event_outbox`` and ``processing_jobs``, to ``event_log`` when the
    trail is enabled, and serialized onto the Redis stream on the ``redis://``
    bus — and ``metadata`` is additionally logged in full when a best-effort
    subscriber fails. Neither is size-capped. So carry identifiers, not secrets
    and not blobs; see the events documentation for the full contract.

    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:
        # Debug, not info: an upgrade is the versioning system working as designed,
        # and this fires once per event — draining an outbox backlog after a deploy
        # would otherwise write one info line per row into the host's log budget.
        logger.debug(
            "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. Plain :func:publish commits the outbox row on its own, so a crash between the domain commit and the outbox write can drop an event; :func:publish_committing and :func:publish_on_commit stage it in the caller's transaction instead. Every subscriber should still have a reconciliation net — a backfill or sweeper that re-derives missed work. "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)
            # Which route an event took is the first thing anyone asks when it
            # appears not to have been processed, and nothing else records it.
            logger.debug("Staged %s (%s) in the outbox for durable delivery", event_type, event.event_id)
        else:
            _dispatch_event(event)
    except Exception as err:
        logger.error("Failed to publish event %s: %s", 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 _no_metadata
    if db is not None and _durable_delivery_enabled(event_type):
        # Atomic path: stage every outbox row inside the caller's transaction.
        _stage_batch(event_type, payloads, source, metadata_of, schema_version, db)
        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)

publish_many_on_commit

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

Publish many same-type events tied to db's commit.

The batch counterpart of :func:publish_on_commit, with the same delivery routing and the same guarantee: best-effort events dispatch, in order, only after the session's outermost transaction commits. An empty sequence is a no-op; it neither stages anything nor starts a transaction.

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.	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
schema_version	int	Payload schema version.	INITIAL_SCHEMA_VERSION

Returns:

Type	Description
None	None.

Source code in jasil/publisher.py

def publish_many_on_commit(
    event_type: str,
    payloads: Sequence[dict],
    *,
    source: str,
    metadata_for: Callable[[dict], dict | None] | None = None,
    db: Any,
    schema_version: int = INITIAL_SCHEMA_VERSION,
) -> None:
    """Publish many same-type events tied to ``db``'s commit.

    The batch counterpart of :func:`publish_on_commit`, with the same delivery
    routing and the same guarantee: best-effort events dispatch, in order, only
    after the session's outermost transaction commits. An empty sequence is a
    no-op; it neither stages anything nor starts a transaction.

    Args:
        event_type: The domain-owned channel shared by every event in the batch.
        payloads: One payload per event.
        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).
        schema_version: Payload schema version.

    Returns:
        None.
    """
    if db is None:
        raise ValueError("publish_on_commit/publish_many_on_commit require the producer's session as `db`")
    if not isinstance(db, Session):
        raise TypeError(
            f"publish_on_commit/publish_many_on_commit need a synchronous sqlalchemy.orm.Session, got {type(db).__name__}"
        )
    if not payloads:
        return
    metadata_of = metadata_for if metadata_for is not None else _no_metadata
    if _durable_delivery_enabled(event_type):
        _stage_batch(event_type, payloads, source, metadata_of, schema_version, db)
        return
    for payload in payloads:
        try:
            event = _mint(event_type, payload, source, metadata_of(payload), schema_version)
            run_after_commit(db, partial(_dispatch_event, event))
        except Exception as err:
            logger.error("Failed to publish event %s: %s", event_type, err, exc_info=err)

publish_on_commit

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

Publish a domain event tied to db's commit, without owning the commit.

The caller stages the event and whoever owns the transaction commits whenever it likes, so the publishing function composes with other writes in one atomic unit of work.

• Durable delivery enabled (durable jobs on + a durable subscriber for event_type): the outbox row is staged on db immediately, uncommitted, and commits (or rolls back) atomically with the caller's domain rows. A staging failure propagates to the caller.
• Otherwise (best-effort bus path): the event is minted now (so the ambient request id is captured) and held on the session. It is dispatched on the bus only after db's outermost transaction commits, in publish order. An event published inside a SAVEPOINT that is rolled back, or in a transaction that is rolled back or closed, is discarded and never dispatched. Dispatch failures are logged and swallowed.

"Outermost" is the session's outermost transaction. If the session joined an external transaction (join_transaction_mode="create_savepoint"), its commit() is that session's outermost commit even though the enclosing transaction may still roll back.

Publishing starts the session's transaction if none is active. Sessions are synchronous sqlalchemy.orm.Session instances; each carries its own pending events, so concurrent sessions never see each other's.

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 change).	required
schema_version	int	Payload schema version.	INITIAL_SCHEMA_VERSION

Returns:

Type	Description
None	None.

Source code in jasil/publisher.py

def publish_on_commit(
    event_type: str,
    payload: dict,
    *,
    source: str,
    metadata: dict | None = None,
    db: Any,
    schema_version: int = INITIAL_SCHEMA_VERSION,
) -> None:
    """Publish a domain event tied to ``db``'s commit, without owning the commit.

    The caller stages the event and whoever owns the transaction commits whenever
    it likes, so the publishing function composes with other writes in one atomic
    unit of work.

    * **Durable delivery enabled** (durable jobs on + a durable subscriber for
      ``event_type``): the outbox row is staged on ``db`` immediately, uncommitted,
      and commits (or rolls back) atomically with the caller's domain rows. A
      staging failure propagates to the caller.
    * **Otherwise** (best-effort bus path): the event is minted now (so the ambient
      request id is captured) and held on the session. It is dispatched on the bus
      only after ``db``'s **outermost** transaction commits, in publish order.
      An event published inside a SAVEPOINT that is rolled back, or in a
      transaction that is rolled back or closed, is discarded and never
      dispatched. Dispatch failures are logged and swallowed.

    "Outermost" is the session's outermost transaction. If the session joined an
    external transaction (``join_transaction_mode="create_savepoint"``), its
    ``commit()`` is that session's outermost commit even though the enclosing
    transaction may still roll back.

    Publishing starts the session's transaction if none is active. Sessions are
    synchronous ``sqlalchemy.orm.Session`` instances; each carries its own pending
    events, so concurrent sessions never see each other's.

    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 change).
        schema_version: Payload schema version.

    Returns:
        None.
    """
    publish_many_on_commit(
        event_type,
        [payload],
        source=source,
        metadata_for=lambda _payload: metadata,
        db=db,
        schema_version=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. This is why
                    # ``metadata`` must never carry a secret.
                    "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), the configured number of web workers, and whether durable jobs run inside the API processes or in standalone worker processes (JobWorkerMode).
• What does that shape imply? DeploymentTopology.multi_process (two processes can reach the same capability) and DeploymentTopology.multi_node (they may be on different machines). resolve_topology is the one place that knows which inputs imply which fact, so a new kind of process is a single line there and every consistency rule picks it up.

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).
job_worker_mode	JobWorkerMode	Where durable-job consumers run.

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).
        job_worker_mode: Where durable-job consumers run.
    """

    profile: DeploymentProfile
    web_workers: int
    job_worker_mode: JobWorkerMode

    @property
    def multi_process_causes(self) -> tuple[str, ...]:
        """What makes this topology multi-process; empty for a single process.

        Each entry is a short phrase naming the setting behind it, so a refusal
        can tell an operator *why* their deployment counts as multi-process.
        """
        causes: list[str] = []
        if self.profile is DeploymentProfile.DISTRIBUTED:
            causes.append("profile=distributed runs replicas on separate nodes")
        if self.web_workers > 1:
            causes.append(f"web_workers={self.web_workers} runs several API processes")
        if self.job_worker_mode is JobWorkerMode.STANDALONE:
            causes.append("job_worker_mode=standalone runs job workers in their own processes")
        return tuple(causes)

    @property
    def multi_process_reason(self) -> str:
        """:attr:`multi_process_causes` joined into one phrase for a refusal message."""
        return "; ".join(self.multi_process_causes)

    @property
    def multi_process(self) -> bool:
        """Whether two or more processes can reach the same capability.

        True for the ``distributed`` profile, for more than one web worker, and
        whenever durable-job consumers run as standalone processes. Process-local
        memory, an in-process lock or a recording mail backend cannot be shared
        between them.
        """
        return bool(self.multi_process_causes)

    @property
    def multi_node(self) -> bool:
        """Whether those processes may run on different machines.

        Only the ``distributed`` profile promises that. Several processes of a
        ``local`` deployment share one host, and with it one disk.
        """
        return self.profile is DeploymentProfile.DISTRIBUTED

multi_node property

multi_node

Whether those processes may run on different machines.

Only the distributed profile promises that. Several processes of a local deployment share one host, and with it one disk.

multi_process property

multi_process

Whether two or more processes can reach the same capability.

True for the distributed profile, for more than one web worker, and whenever durable-job consumers run as standalone processes. Process-local memory, an in-process lock or a recording mail backend cannot be shared between them.

multi_process_causes property

multi_process_causes

What makes this topology multi-process; empty for a single process.

Each entry is a short phrase naming the setting behind it, so a refusal can tell an operator why their deployment counts as multi-process.

multi_process_reason property

multi_process_reason

:attr:multi_process_causes joined into one phrase for a refusal message.

JobWorkerMode

Bases: StrEnum

Where durable-job consumers run.

A fact about the deployment's shape, not a count: it does not change when workers are scaled, and every process of a deployment loads the same value, so the API replicas and the workers agree on the shape without coordinating.

Attributes:

Name	Type	Description
IN_PROCESS		Each API process starts its own consumer with jasil.jobs.service.start_job_worker. The default.
STANDALONE		Consumers are dedicated processes started with jasil.jobs.service.run_job_worker; API processes only schedule the relay and the reaper. Every such process is another process that reaches every capability.

Source code in jasil/profile.py

class JobWorkerMode(StrEnum):
    """Where durable-job consumers run.

    A fact about the deployment's shape, not a count: it does not change when
    workers are scaled, and every process of a deployment loads the same value, so
    the API replicas and the workers agree on the shape without coordinating.

    Attributes:
        IN_PROCESS: Each API process starts its own consumer with
            ``jasil.jobs.service.start_job_worker``. The default.
        STANDALONE: Consumers are dedicated processes started with
            ``jasil.jobs.service.run_job_worker``; API processes only schedule
            the relay and the reaper. Every such process is another process that
            reaches every capability.
    """

    IN_PROCESS = "in_process"
    STANDALONE = "standalone"

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_job_worker_mode

parse_job_worker_mode(value)

Parse a raw job-worker-mode value into a JobWorkerMode.

Parameters:

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

Returns:

Type	Description
JobWorkerMode	The parsed mode; JobWorkerMode.IN_PROCESS when unset or empty.

Raises:

Type	Description
ValueError	When the value is a non-empty, unrecognized mode name. Raising (rather than defaulting) prevents a typo like standalon from silently running in_process and hiding every worker process from the consistency checks.

Source code in jasil/profile.py

def parse_job_worker_mode(value: str | JobWorkerMode | None) -> JobWorkerMode:
    """Parse a raw job-worker-mode value into a ``JobWorkerMode``.

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

    Returns:
        The parsed mode; ``JobWorkerMode.IN_PROCESS`` when unset or empty.

    Raises:
        ValueError: When the value is a non-empty, unrecognized mode name.
            Raising (rather than defaulting) prevents a typo like ``standalon``
            from silently running ``in_process`` and hiding every worker process
            from the consistency checks.
    """
    if isinstance(value, JobWorkerMode):
        return value
    if value is None:
        return JobWorkerMode.IN_PROCESS
    normalized = value.strip().lower()
    if not normalized:
        return JobWorkerMode.IN_PROCESS
    try:
        return JobWorkerMode(normalized)
    except ValueError as err:
        valid = ", ".join(mode.value for mode in JobWorkerMode)
        raise ValueError(f"Invalid job worker mode '{value}'. Valid values: {valid}.") from err

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, job_worker_mode)

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

Parameters:

Name	Type	Description	Default
profile	DeploymentProfile	The deployment profile.	required
web_workers	int	Configured web worker count.	required
job_worker_mode	JobWorkerMode	Where durable-job consumers run.	required

Returns:

Type	Description
DeploymentTopology	The resolved DeploymentTopology.

Source code in jasil/profile.py

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

    Args:
        profile: The deployment profile.
        web_workers: Configured web worker count.
        job_worker_mode: Where durable-job consumers run.

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

Capability report

jasil.capabilities

Startup capability report and deployment-consistency checks.

Renders how each infrastructure capability (state, storage, events, lock, mail, 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.

Every rule asks one question of a :class:~jasil.profile.DeploymentTopology rather than re-deriving the deployment's shape from raw settings: can two processes reach this capability (multi_process), or two machines (multi_node)? What makes a topology multi-process — the distributed profile, more than one web worker, or standalone job workers — is decided in one place, :func:~jasil.profile.resolve_topology, so a new kind of process needs no change here.

Five 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 is multi-process; 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 when the topology is multi-node (the distributed profile), where replicas run on separate nodes with no shared disk. Several processes of a local deployment share 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 when the topology is multi-process, or every process would run every scheduled job. Validated by :func:check_lock_consistency.
• Mail must not resolve to the recording memory:// backend when the topology is multi-process: every process would keep its own outbox and no message would ever be delivered. Validated by :func:check_mail_consistency.
• Job workers may not be declared standalone while durable jobs are disabled: the declaration would make every process stricter for consumers that cannot exist. Validated by :func:check_job_worker_consistency.

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

:func:check_deployment_consistency applies all five 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"job_worker_mode={self.topology.job_worker_mode.value}, "
            f"multi_process={self.topology.multi_process}, "
            f"multi_node={self.topology.multi_node})"
        )
        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"job_worker_mode={self.topology.job_worker_mode.value}, "
        f"multi_process={self.topology.multi_process}, "
        f"multi_node={self.topology.multi_node})"
    )
    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 / mail / 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 / mail / 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(
            "mail",
            _mail_backend_of(settings.mail.uri, secure=settings.mail.secure),
            "mail.uri" if settings.mail.uri else "not configured",
        ),
        CapabilityRow("clock", "system", "always the system clock"),
    )
    return CapabilityReport(topology=topology_of(settings), 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 five 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 five 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.
    """
    topology = topology_of(settings)
    return [
        *check_state_consistency(
            topology=topology,
            state_sources=(
                StateSource(label="state_uri", uri=settings.resolved_state_uri),
                StateSource(label="events_uri", uri=settings.resolved_events_uri),
            ),
        ),
        *check_storage_consistency(
            topology=topology,
            storage_uri=settings.resolved_storage_uri,
            storage_label="storage_uri",
        ),
        *check_lock_consistency(
            topology=topology,
            lock_uri=settings.resolved_lock_uri,
            lock_label="lock_uri",
        ),
        *check_mail_consistency(
            topology=topology,
            mail_uri=settings.mail.uri,
            mail_label="mail.uri",
        ),
        *check_job_worker_consistency(
            topology=topology,
            jobs_enabled=settings.jobs.enabled,
        ),
    ]

check_job_worker_consistency

check_job_worker_consistency(*, topology, jobs_enabled)

Return a fatal issue when standalone job workers are declared with jobs off.

job_worker_mode=standalone tells every process that other processes consume durable jobs, which makes the other rules stricter. With jobs.enabled false the jobs layer is dormant, so no such consumer can exist and the declaration would only impose constraints nothing justifies.

Parameters:

Name	Type	Description	Default
topology	DeploymentTopology	The deployment topology.	required
jobs_enabled	bool	Whether the durable-job pipeline is enabled.	required

Returns:

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

Source code in jasil/capabilities.py

def check_job_worker_consistency(
    *,
    topology: DeploymentTopology,
    jobs_enabled: bool,
) -> list[str]:
    """Return a fatal issue when standalone job workers are declared with jobs off.

    ``job_worker_mode=standalone`` tells every process that other processes
    consume durable jobs, which makes the other rules stricter. With
    ``jobs.enabled`` false the jobs layer is dormant, so no such consumer can
    exist and the declaration would only impose constraints nothing justifies.

    Args:
        topology: The deployment topology.
        jobs_enabled: Whether the durable-job pipeline is enabled.

    Returns:
        A single-item issue list when contradictory; empty otherwise.
    """
    if topology.job_worker_mode is not JobWorkerMode.STANDALONE or jobs_enabled:
        return []
    return [
        "job_worker_mode=standalone declares standalone job workers, but jobs.enabled is false, "
        "so the durable-job pipeline is dormant and no worker can consume anything. Enable "
        "jobs, or leave job_worker_mode at in_process."
    ]

check_lock_consistency

check_lock_consistency(*, topology, 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 the topology is multi-process, 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:// under distributed, so this trips on an explicit lock_uri="noop://" override, or on a local default paired with several processes. A single-process deployment has nothing to coordinate, so noop stays valid there.

Parameters:

Name	Type	Description	Default
topology	DeploymentTopology	The deployment topology.	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(
    *,
    topology: DeploymentTopology,
    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 the topology is multi-process, 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://`` under ``distributed``, so this
    trips on an explicit ``lock_uri="noop://"`` override, or on a ``local``
    default paired with several processes. A single-process deployment has
    nothing to coordinate, so ``noop`` stays valid there.

    Args:
        topology: The deployment topology.
        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 topology.profile is DeploymentProfile.CUSTOM or not topology.multi_process:
        return []
    if not lock_uri.strip().lower().startswith("noop://"):
        return []
    return [
        f"{lock_label} resolves to an in-process no-op lock, but this deployment runs multiple "
        f"processes ({topology.multi_process_reason}) that would each run scheduled jobs. Point it at "
        f"the shared database lock (postgres-advisory://)."
    ]

check_mail_consistency

check_mail_consistency(*, topology, mail_uri, mail_label)

Return a fatal issue when a multi-process deployment records mail in memory.

memory:// keeps each message in the sending process's outbox and never delivers it. In a single process that is a test fixture; in a multi-process deployment — including a standalone job worker, whose handlers are where mail is sent — it is production mail silently going nowhere. Unset mail is not flagged: it raises on every send, so it cannot drop a message unnoticed.

Parameters:

Name	Type	Description	Default
topology	DeploymentTopology	The deployment topology.	required
mail_uri	str | None	The configured mail URI, or None when mail is unset.	required
mail_label	str	The setting backing mail (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_mail_consistency(
    *,
    topology: DeploymentTopology,
    mail_uri: str | None,
    mail_label: str,
) -> list[str]:
    """Return a fatal issue when a multi-process deployment records mail in memory.

    ``memory://`` keeps each message in the sending process's outbox and never
    delivers it. In a single process that is a test fixture; in a multi-process
    deployment — including a standalone job worker, whose handlers are where mail
    is sent — it is production mail silently going nowhere. Unset mail is not
    flagged: it raises on every send, so it cannot drop a message unnoticed.

    Args:
        topology: The deployment topology.
        mail_uri: The configured mail URI, or ``None`` when mail is unset.
        mail_label: The setting backing mail (for the message).

    Returns:
        A single-item issue list when misconfigured; empty otherwise. The
        ``custom`` profile is exempt — it promises no defaults.
    """
    if topology.profile is DeploymentProfile.CUSTOM or not mail_uri or not topology.multi_process:
        return []
    # ``build_platform`` accepts only the literal ``scheme://`` form, so this is
    # the same test the builder applies.
    if mail_uri.partition("://")[0].lower() != "memory":
        return []
    return [
        f"{mail_label} resolves to the in-memory mail recorder, but this deployment runs multiple "
        f"processes ({topology.multi_process_reason}), and no message would ever be delivered. Point it "
        f"at a relay (smtp:// or smtps://), or leave it unset."
    ]

check_state_consistency

check_state_consistency(*, topology, state_sources)

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

A multi-process deployment that 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
topology	DeploymentTopology	The deployment topology.	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(
    *,
    topology: DeploymentTopology,
    state_sources: Sequence[StateSource],
) -> list[str]:
    """Return fatal issues for cross-process backends (empty when consistent).

    A multi-process deployment that 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:
        topology: The deployment topology.
        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 topology.profile is DeploymentProfile.CUSTOM or not topology.multi_process:
        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 this deployment runs multiple processes "
            f"({topology.multi_process_reason}) that need a backend shared across them. Point it at "
            f"Redis (redis://...) or run a single process under the local profile."
        )
    return issues

check_storage_consistency

check_storage_consistency(
    *, topology, storage_uri, storage_label
)

Return a fatal issue when multi-node 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. Several processes of a local deployment share one host disk, so local storage stays valid there and is not flagged.

Parameters:

Name	Type	Description	Default
topology	DeploymentTopology	The deployment topology.	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(
    *,
    topology: DeploymentTopology,
    storage_uri: str,
    storage_label: str,
) -> list[str]:
    """Return a fatal issue when multi-node 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. Several processes
    of a ``local`` deployment share one host disk, so local storage stays valid
    there and is not flagged.

    Args:
        topology: The deployment topology.
        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 not topology.multi_node:
        return []
    if not storage_uri.strip().lower().startswith("local://"):
        return []
    return [
        f"{storage_label} resolves to the local filesystem, but "
        f"profile={topology.profile.value} runs replicas on separate nodes that do "
        f"not share a disk. Point it at object storage (s3://bucket/...)."
    ]

topology_of

topology_of(settings)

Return the deployment topology settings describe.

The one place that turns settings into a topology, so the report, every rule and the job-worker entry points agree on what shape a deployment has.

Parameters:

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

Returns:

Type	Description
DeploymentTopology	The resolved topology.

Source code in jasil/capabilities.py

def topology_of(settings: JasilSettings) -> DeploymentTopology:
    """Return the deployment topology ``settings`` describe.

    The one place that turns settings into a topology, so the report, every rule
    and the job-worker entry points agree on what shape a deployment has.

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

    Returns:
        The resolved topology.
    """
    return resolve_topology(
        profile=settings.profile,
        web_workers=settings.web_workers,
        job_worker_mode=settings.job_worker_mode,
    )

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:MailSettings, :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.
job_worker_mode	JobWorkerMode	Where durable-job consumers run: in_process (each API process calls start_job_worker) or standalone (dedicated processes call run_job_worker). Like web_workers it describes the deployment's shape, and every process of the deployment must load the same value: a standalone worker is one more process that reaches every capability, and the consistency checks can only see it if the API processes know about it too.
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.
mail	MailSettings	Outbound-mail 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.
        job_worker_mode: Where durable-job consumers run: ``in_process`` (each API
            process calls ``start_job_worker``) or ``standalone`` (dedicated
            processes call ``run_job_worker``). Like ``web_workers`` it describes
            the deployment's shape, and **every process of the deployment must
            load the same value**: a standalone worker is one more process that
            reaches every capability, and the consistency checks can only see it
            if the API processes know about it too.
        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.
        mail: Outbound-mail backend configuration.
        network: Outbound-egress configuration.
    """

    profile: DeploymentProfile = DeploymentProfile.LOCAL
    web_workers: int = 1
    job_worker_mode: JobWorkerMode = JobWorkerMode.IN_PROCESS
    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)
    mail: MailSettings = field(default_factory=MailSettings)
    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 worker may go silent before the reaper may reclaim the jobs it holds. A running worker renews its leases on every heartbeat, so this is not a cap on handler runtime (see max_runtime_seconds). Must be at least three heartbeat_interval_seconds.
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_runtime_seconds	int	The longest a single handler is kept leased. Past it the job is abandoned to the reaper, the handler is told to stop (jasil.jobs.lease_lost), and the rest of the worker's batch is released to run elsewhere. Must not be below lease_seconds.
record_attempts	bool	Write one job_attempts row per attempt to run a job: which worker ran it, when, for how long, and how it ended. On by default; False skips every attempt write (job processing is unaffected) and the attempt-derived admin fields read None.
heartbeat_interval_seconds	float	Interval between worker heartbeats, each of which renews held leases and refreshes the worker-registry row.
max_attempts	int	Attempts before a job is dead-lettered.
retention_days	int	Age at which relayed outbox rows, completed jobs, their attempt history, and stopped and stale worker rows are pruned. A job that is pending, claimed or dead-lettered keeps its whole history. <= 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 worker may go silent before the reaper may
            reclaim the jobs it holds. A running worker renews its leases on every
            heartbeat, so this is *not* a cap on handler runtime (see
            ``max_runtime_seconds``). Must be at least three
            ``heartbeat_interval_seconds``.
        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_runtime_seconds: The longest a single handler is kept leased. Past it
            the job is abandoned to the reaper, the handler is told to stop
            (``jasil.jobs.lease_lost``), and the rest of the worker's batch is
            released to run elsewhere. Must not be below ``lease_seconds``.
        record_attempts: Write one ``job_attempts`` row per attempt to run a job:
            which worker ran it, when, for how long, and how it ended. On by
            default; ``False`` skips every attempt write (job processing is
            unaffected) and the attempt-derived admin fields read ``None``.
        heartbeat_interval_seconds: Interval between worker heartbeats, each of
            which renews held leases and refreshes the worker-registry row.
        max_attempts: Attempts before a job is dead-lettered.
        retention_days: Age at which relayed outbox rows, completed jobs, their
            attempt history, and stopped and stale worker rows are pruned. A job
            that is pending, claimed or dead-lettered keeps its whole history.
            ``<= 0`` disables pruning.
    """

    enabled: bool = False
    lease_seconds: int = 300
    max_runtime_seconds: int = 1800
    record_attempts: bool = True
    batch_size: int = 20
    backoff_base_seconds: int = 60
    backoff_max_seconds: int = 3600
    poll_interval_seconds: float = 5.0
    heartbeat_interval_seconds: float = 15.0
    max_attempts: int = 5
    retention_days: int = 30

MailSettings dataclass

Outbound-mail backend configuration.

Attributes:

Name	Type	Description
uri	str | None	memory:// (tests), smtp://host[:port] (STARTTLS, default port 587) or smtps://host[:port] (implicit TLS, default port 465). Unset disables mail: sending raises :class:~jasil.providers.MailNotConfiguredError. Credentials never belong in the URI. The host is trusted operator configuration and is not checked against the SSRF address denylist (a relay is normally on a private network): never derive it from user input. memory:// is refused in a multi-process deployment (see enforce_deployment_consistency), where no message would ever be delivered.
from_address	str	Sender identity, "addr" or "Name <addr>". Required for the SMTP schemes. Independent of username.
secure	bool	True (the default) requires verified TLS. False sends over a cleartext connection and is accepted only for smtp://, only when the host resolves exclusively to private, loopback or link-local addresses (re-checked on every send), and never together with username / password. That check resolves the host while the platform is built, so it needs a working resolver at startup.
username	str | None	SMTP login, when the server requires authentication.
password	str | None	SMTP password. Required with username, and the reverse.
ehlo_hostname	str | None	Name announced in EHLO. Defaults to this machine's hostname; set a fully qualified one for a relay that rejects a bare name.

Source code in jasil/settings.py

@dataclass(frozen=True)
class MailSettings:
    """Outbound-mail backend configuration.

    Attributes:
        uri: ``memory://`` (tests), ``smtp://host[:port]`` (STARTTLS, default
            port 587) or ``smtps://host[:port]`` (implicit TLS, default port
            465). Unset disables mail: sending raises
            :class:`~jasil.providers.MailNotConfiguredError`. Credentials never
            belong in the URI. The host is trusted operator configuration and
            is not checked against the SSRF address denylist (a relay is
            normally on a private network): never derive it from user input.
            ``memory://`` is refused in a multi-process deployment (see
            ``enforce_deployment_consistency``), where no message would ever be
            delivered.
        from_address: Sender identity, ``"addr"`` or ``"Name <addr>"``. Required
            for the SMTP schemes. Independent of ``username``.
        secure: ``True`` (the default) requires verified TLS. ``False`` sends
            over a cleartext connection and is accepted only for ``smtp://``,
            only when the host resolves exclusively to private, loopback or
            link-local addresses (re-checked on every send), and never together with ``username`` / ``password``.
            That check resolves the host while the platform is built, so it
            needs a working resolver at startup.
        username: SMTP login, when the server requires authentication.
        password: SMTP password. Required with ``username``, and the reverse.
        ehlo_hostname: Name announced in ``EHLO``. Defaults to this machine's
            hostname; set a fully qualified one for a relay that rejects a bare
            name.
    """

    uri: str | None = None
    from_address: str = ""
    secure: bool = True
    username: str | None = None
    password: str | None = field(default=None, repr=False)
    ehlo_hostname: str | None = None

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. Prefer a CIDR: a hostname entry exempts every address that name resolves to, including a cloud metadata endpoint.

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. Prefer a CIDR: a *hostname* entry
            exempts every address that name resolves to, including a cloud
            metadata endpoint.
    """

    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, job_workers, and job_attempts — 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 model module is imported. The model modules bind their classes to the active base at import time, so importing one first is a configuration error and raises. A host that would rather not own a base may call map_models() with no argument and use JASIL's convenience :data:Base.

Two modules reach a model directly and therefore inherit that ordering constraint: :mod:jasil.jobs.crud and :mod:jasil.event_log.crud. Import them inside a function, or only after map_models has run. Every other public entry point — :mod:jasil.publisher, :mod:jasil.retention, :mod:jasil.jobs.service, :mod:jasil.container, :mod:jasil.deps, :mod:jasil.testing — defers its model imports and is safe to import from anywhere, at any point in the host's import graph.

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 importing any JASIL model module (omit the base to use jasil.orm.Base). "
            "You are seeing this because something imported jasil.jobs.crud, jasil.event_log.crud, "
            "or a model module directly at import time — move that import inside a function, or map "
            "first. jasil.publisher, jasil.retention, jasil.jobs.service, jasil.container and "
            "jasil.deps are always safe to import at module scope."
        )
    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.
mail	MailProvider	Outbound-mail provider. Always present — when mail is unconfigured this is a backend whose send raises :class:~jasil.providers.MailNotConfiguredError, so a message is never dropped silently.

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.
        mail: Outbound-mail provider. Always present — when mail is unconfigured
            this is a backend whose ``send`` raises
            :class:`~jasil.providers.MailNotConfiguredError`, so a message is
            never dropped silently.
    """

    profile: DeploymentProfile
    state: StateProvider
    storage: StorageProvider
    events: EventBusProvider
    lock: LockProvider
    clock: ClockProvider
    geocoding: GeocodingProvider = field(default_factory=NullGeocoding)
    recorder: EventRecorder | None = None
    mail: MailProvider = field(default_factory=NullMail)

    def close(self) -> None:
        """Release what the platform owns: the bus consumer and the Redis clients.

        Call once on shutdown, after the host has stopped producing events. Only
        the two capabilities that hold something beyond the process are affected —
        the rest are pure or borrow the host's engine, which the host closes.

        Two things it deliberately does *not* stop, because the platform does not
        own them: the durable-job worker and its scheduled maintenance (see
        ``jasil.jobs.service.stop_job_worker``), and the database engine.

        Safe to call more than once, and never raises: a failure to shut a
        connection down must not mask whatever prompted the shutdown.
        """
        try:
            self.events.stop()
        except Exception as error:
            logger.warning("Failed to stop the event bus during shutdown: %r", error)
        redis_clients.close_shared_clients()

close

close()

Release what the platform owns: the bus consumer and the Redis clients.

Call once on shutdown, after the host has stopped producing events. Only the two capabilities that hold something beyond the process are affected — the rest are pure or borrow the host's engine, which the host closes.

Two things it deliberately does not stop, because the platform does not own them: the durable-job worker and its scheduled maintenance (see jasil.jobs.service.stop_job_worker), and the database engine.

Safe to call more than once, and never raises: a failure to shut a connection down must not mask whatever prompted the shutdown.

Source code in jasil/container.py

def close(self) -> None:
    """Release what the platform owns: the bus consumer and the Redis clients.

    Call once on shutdown, after the host has stopped producing events. Only
    the two capabilities that hold something beyond the process are affected —
    the rest are pure or borrow the host's engine, which the host closes.

    Two things it deliberately does *not* stop, because the platform does not
    own them: the durable-job worker and its scheduled maintenance (see
    ``jasil.jobs.service.stop_job_worker``), and the database engine.

    Safe to call more than once, and never raises: a failure to shut a
    connection down must not mask whatever prompted the shutdown.
    """
    try:
        self.events.stop()
    except Exception as error:
        logger.warning("Failed to stop the event bus during shutdown: %r", error)
    redis_clients.close_shared_clients()

build_platform

build_platform(settings=None, *, clock=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
clock	ClockProvider | None	Time source to wire into the platform and the backends that stamp time (mail Date). Defaults to :class:SystemClock; a test helper passes a controllable one.	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, when the resolved wiring contradicts the deployment topology, or when mail is configured incorrectly.
RuntimeError	When a selected Redis backend cannot be reached.

Source code in jasil/container.py

def build_platform(settings: JasilSettings | None = None, *, clock: ClockProvider | 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``.
        clock: Time source to wire into the platform and the backends that stamp
            time (mail ``Date``). Defaults to :class:`SystemClock`; a test
            helper passes a controllable one.

    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, when the resolved
            wiring contradicts the deployment topology, or when ``mail`` is
            configured incorrectly.
        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)
    clock = clock if clock is not None else SystemClock()
    # Built first: it does no I/O beyond a DNS lookup (``secure=False`` only), so a
    # mistyped mail setting fails before any Redis client is connected and left open.
    mail = _build_mail(settings, clock)
    return Platform(
        profile=profile,
        state=_build_state(settings),
        storage=_build_storage(settings),
        events=_build_events(settings, recorder),
        lock=_build_lock(settings),
        clock=clock,
        geocoding=_build_geocoding(settings),
        mail=mail,
        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

is_platform_active

is_platform_active()

Return whether a platform has been published.

Returns:

Type	Description
bool	True once :func:set_active_platform has run and before :func:reset.

Source code in jasil/runtime.py

def is_platform_active() -> bool:
    """Return whether a platform has been published.

    Returns:
        True once :func:`set_active_platform` has run and before :func:`reset`.
    """
    return _active_platform is not None

reset

reset()

Unpublish the active platform.

For tests that build a platform per case; production publishes once at startup and never resets. This does not close the platform — call :meth:jasil.container.Platform.close first if it owns live connections.

Source code in jasil/runtime.py

def reset() -> None:
    """Unpublish the active platform.

    For tests that build a platform per case; production publishes once at
    startup and never resets. This does not close the platform — call
    :meth:`jasil.container.Platform.close` first if it owns live connections.
    """
    global _active_platform
    _active_platform = None

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

Shutdown

jasil.lifecycle

Ordered shutdown of everything JASIL owns in a process.

Two things outlive a request and have to be wound down deliberately: the durable-job worker thread, and the platform's own resources — the event-bus consumer thread and the shared Redis clients. The order matters. The worker runs subscribers, and a subscriber that publishes needs the bus still up, so the worker stops first.

They are two calls underneath because :mod:jasil.container must not import :mod:jasil.jobs: the jobs layer imports APScheduler at module scope, and a deployment that never enables durable jobs has to be able to build a platform without the jobs extra installed at all. This module is the seam that composes them, so the ordering lives in one place instead of in every host::

import jasil.lifecycle as jasil_lifecycle

@asynccontextmanager
async def lifespan(app: FastAPI):
    ...
    yield
    jasil_lifecycle.shutdown()

What it deliberately leaves alone is what the host owns: the APScheduler instance passed to schedule_job_maintenance, and the database engine behind the session factory. JASIL never created either, so it does not close them.

shutdown

shutdown()

Stop the durable-job worker, release the platform, and unpublish it.

Idempotent, and safe to call when nothing was ever started — a process that only ever published events has no worker and may have no platform.

Never raises. Shutdown runs while something else is often already going wrong, and a failure to release a connection must not mask it; each step logs and continues instead.

Returns:

Type	Description
None	None.

Source code in jasil/lifecycle.py

def shutdown() -> None:
    """Stop the durable-job worker, release the platform, and unpublish it.

    Idempotent, and safe to call when nothing was ever started — a process that
    only ever published events has no worker and may have no platform.

    Never raises. Shutdown runs while something else is often already going
    wrong, and a failure to release a connection must not mask it; each step
    logs and continues instead.

    Returns:
        None.
    """
    _stop_job_worker()
    _close_platform()
    platform_runtime.reset()

FastAPI dependencies

jasil.deps

FastAPI dependencies exposing the platform providers to routes and handlers.

These thin dependencies resolve the assembled Platform so routes depend on providers (via Depends) rather than importing backends.

Two places are consulted, in order:

1. request.app.state.platform, when the host attached one. Use this to give a single process more than one platform — two apps mounted together, or a test client — since it is scoped to the app rather than to the process.
2. The process-wide platform published by jasil.runtime.set_active_platform, which is what the quick start does and what every non-request caller (the scheduler, the durable-job worker, a background thread) already resolves.

So a host that followed the quick start needs no extra wiring here, and one that wants per-app isolation gets it by setting app.state.platform at startup.

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 get_platform(request).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 get_platform(request).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 get_platform(request).lock

get_mail

get_mail(request)

Return the outbound-mail provider.

Source code in jasil/deps.py

def get_mail(request: Request) -> MailProvider:
    """Return the outbound-mail provider."""
    return get_platform(request).mail

get_platform

get_platform(request)

Return the Platform for this request.

Parameters:

Name	Type	Description	Default
request	Request	The incoming request, used to reach app.state.	required

Returns:

Type	Description
Platform	The app-scoped platform when the host attached one, else the
Platform	process-wide platform.

Raises:

Type	Description
RuntimeError	When neither has been published, i.e. startup never ran.

Source code in jasil/deps.py

def get_platform(request: Request) -> Platform:
    """Return the ``Platform`` for this request.

    Args:
        request: The incoming request, used to reach ``app.state``.

    Returns:
        The app-scoped platform when the host attached one, else the
        process-wide platform.

    Raises:
        RuntimeError: When neither has been published, i.e. startup never ran.
    """
    platform: Platform | None = getattr(request.app.state, "platform", None)
    if platform is not None:
        return platform
    return platform_runtime.get_active_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 get_platform(request).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 get_platform(request).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)
        self._queues: dict[str, str] = {}

    def register(
        self,
        event_type: str,
        subscriber_id: str,
        handler: JobHandler,
        *,
        queue: str = DEFAULT_QUEUE,
    ) -> 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.
            queue: Durable-job queue assigned to this subscriber's fan-out.

        Returns:
            None.

        Raises:
            ValueError: When an identifier is invalid for the 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)
        validate_queue_name(queue)
        self._handlers[subscriber_id] = handler
        self._queues[subscriber_id] = queue
        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 queue_for(self, subscriber_id: str) -> str:
        """Return the queue assigned to a registered subscriber.

        Raises:
            LookupError: When registry state has no queue for ``subscriber_id``.
                Registration always stores the handler and queue together, so
                this indicates inconsistent process-local state and must not be
                hidden by routing the job to another queue.
        """
        try:
            return self._queues[subscriber_id]
        except KeyError as error:
            raise LookupError(f"no queue registered for subscriber_id {subscriber_id!r}") from error

    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 subscriber_ids_for_queues(self, queues: Iterable[str]) -> tuple[str, ...]:
        """Return registered subscriber ids assigned to the selected queues."""
        selected = normalize_queue_selector(queues)
        if selected is None:
            return tuple(self._handlers)
        selected_set = frozenset(selected)
        return tuple(subscriber_id for subscriber_id, queue in self._queues.items() if queue in selected_set)

    def queue_assignments(self) -> dict[str, str]:
        """Return a copy of the ``{subscriber_id: queue}`` assignments."""
        return dict(self._queues)

    def clear(self) -> None:
        """Remove every registration (used to reset state between tests)."""
        self._handlers.clear()
        self._by_event_type.clear()
        self._queues.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()
    self._queues.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)

queue_assignments

queue_assignments()

Return a copy of the {subscriber_id: queue} assignments.

Source code in jasil/jobs/registry.py

def queue_assignments(self) -> dict[str, str]:
    """Return a copy of the ``{subscriber_id: queue}`` assignments."""
    return dict(self._queues)

queue_for

queue_for(subscriber_id)

Return the queue assigned to a registered subscriber.

Raises:

Type	Description
LookupError	When registry state has no queue for subscriber_id. Registration always stores the handler and queue together, so this indicates inconsistent process-local state and must not be hidden by routing the job to another queue.

Source code in jasil/jobs/registry.py

def queue_for(self, subscriber_id: str) -> str:
    """Return the queue assigned to a registered subscriber.

    Raises:
        LookupError: When registry state has no queue for ``subscriber_id``.
            Registration always stores the handler and queue together, so
            this indicates inconsistent process-local state and must not be
            hidden by routing the job to another queue.
    """
    try:
        return self._queues[subscriber_id]
    except KeyError as error:
        raise LookupError(f"no queue registered for subscriber_id {subscriber_id!r}") from error

register

register(
    event_type,
    subscriber_id,
    handler,
    *,
    queue=DEFAULT_QUEUE,
)

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
queue	str	Durable-job queue assigned to this subscriber's fan-out.	DEFAULT_QUEUE

Returns:

Type	Description
None	None.

Raises:

Type	Description
ValueError	When an identifier is invalid for the 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,
    *,
    queue: str = DEFAULT_QUEUE,
) -> 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.
        queue: Durable-job queue assigned to this subscriber's fan-out.

    Returns:
        None.

    Raises:
        ValueError: When an identifier is invalid for the 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)
    validate_queue_name(queue)
    self._handlers[subscriber_id] = handler
    self._queues[subscriber_id] = queue
    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)

subscriber_ids_for_queues

subscriber_ids_for_queues(queues)

Return registered subscriber ids assigned to the selected queues.

Source code in jasil/jobs/registry.py

def subscriber_ids_for_queues(self, queues: Iterable[str]) -> tuple[str, ...]:
    """Return registered subscriber ids assigned to the selected queues."""
    selected = normalize_queue_selector(queues)
    if selected is None:
        return tuple(self._handlers)
    selected_set = frozenset(selected)
    return tuple(subscriber_id for subscriber_id, queue in self._queues.items() if queue in selected_set)

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

normalize_queue_selector

normalize_queue_selector(queues)

Validate an optional non-empty queue allowlist and remove duplicates.

Source code in jasil/jobs/registry.py

def normalize_queue_selector(queues: Iterable[str] | None) -> tuple[str, ...] | None:
    """Validate an optional non-empty queue allowlist and remove duplicates."""
    if queues is None:
        return None
    values = (queues,) if isinstance(queues, str) else tuple(queues)
    if not values:
        raise ValueError("queues must be a non-empty allowlist when provided")
    return tuple(dict.fromkeys(validate_queue_name(queue) for queue in values))

validate_queue_name

validate_queue_name(queue)

Validate and return one portable durable-job queue identifier.

Queue names are deliberately lowercase ASCII so equality and ordering have the same meaning under SQLite, PostgreSQL, and MySQL collations.

Parameters:

Name	Type	Description	Default
queue	str	Queue identifier to validate.	required

Returns:

Type	Description
str	The unchanged queue name.

Raises:

Type	Description
ValueError	When the queue is empty, overlong, or contains unsupported characters.

Source code in jasil/jobs/registry.py

def validate_queue_name(queue: str) -> str:
    """Validate and return one portable durable-job queue identifier.

    Queue names are deliberately lowercase ASCII so equality and ordering have
    the same meaning under SQLite, PostgreSQL, and MySQL collations.

    Args:
        queue: Queue identifier to validate.

    Returns:
        The unchanged queue name.

    Raises:
        ValueError: When the queue is empty, overlong, or contains unsupported
            characters.
    """
    if not isinstance(queue, str):
        raise ValueError("queue must be a string")
    check_length(queue, field="queue", limit=MAX_QUEUE_NAME_LENGTH)
    if _QUEUE_NAME_PATTERN.fullmatch(queue) is None:
        raise ValueError(
            "queue must start with a lowercase letter or digit and contain only a-z, 0-9, '.', '_', or '-'"
        )
    return queue

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.
queue	str	Named queue from which the job was claimed.

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.
        queue: Named queue from which the job was claimed.
    """

    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
    queue: str

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,
        max_runtime_seconds: float,
        heartbeat_interval_seconds: float,
        batch_size: int,
        backoff_base_seconds: float,
        backoff_max_seconds: float,
        queues: Iterable[str] | None = None,
        record_attempts: bool = True,
    ) -> None:
        self._record_attempts = record_attempts
        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
        self._queues = normalize_queue_selector(queues)
        self._legacy_queues_backfilled = False
        self._queue_cursor: str | None = None
        self._heartbeat_interval_seconds = heartbeat_interval_seconds
        self._leases = LeaseKeeper(
            session_factory=session_factory,
            clock=clock,
            worker_id=worker_id,
            lease_seconds=lease_seconds,
            max_runtime_seconds=max_runtime_seconds,
            interval_seconds=heartbeat_interval_seconds,
            record_attempts=record_attempts,
        )

    @property
    def leases(self) -> LeaseKeeper:
        """The record of what this worker holds, renewed by the worker heartbeat."""
        return self._leases

    @property
    def heartbeat_interval_seconds(self) -> float:
        """How often the worker heartbeat renews leases and refreshes telemetry."""
        return self._heartbeat_interval_seconds

    @property
    def worker_id(self) -> str:
        """Restart-unique worker and lease-holder identity."""
        return self._worker_id

    @property
    def clock(self) -> ClockProvider:
        """Clock shared by claims and worker telemetry."""
        return self._clock

    @property
    def queues(self) -> tuple[str, ...] | None:
        """Normalized queue allowlist, or None when all queues are selected."""
        return self._queues

    def run_once(self, stop: threading.Event | None = None) -> int:
        """
        Claim and process one batch of due jobs.

        Args:
            stop: When set between jobs, the jobs not yet started are released
                back to the queue instead of being run, so a graceful shutdown
                does not hold them for the rest of the batch.

        Returns:
            The number of jobs processed in this batch (0 when none were due).
        """
        now = self._clock.now()
        self._backfill_legacy_queues()
        default_subscribers, excluded_subscribers = self._legacy_routing()
        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,
                queues=self._queues,
                queue_cursor=self._queue_cursor,
                default_queue_subscribers=default_subscribers,
                excluded_default_queue_subscribers=excluded_subscribers,
                record_attempts=self._record_attempts,
            )
            snapshots = [self._snapshot(job) for job in claimed]
        if not snapshots:
            return 0
        self._queue_cursor = self._cursor_for(snapshots[-1].queue)
        logger.debug("Claimed %d durable job(s) as %s", len(snapshots), self._worker_id)
        self._leases.hold({snapshot.id: snapshot.attempts for snapshot in snapshots})
        try:
            for index, snapshot in enumerate(snapshots):
                # A DB error finishing one job must not abort the rest of the batch;
                # its lease simply expires and the reaper requeues it.
                if stop is not None and stop.is_set():
                    self._leases.release_waiting()
                    break
                try:
                    # The first job's lease was taken a moment ago; every later one
                    # has been waiting, so it is renewed (and confirmed held) first.
                    if index and not self._still_held(snapshot):
                        continue
                    if self._run_job(snapshot):
                        # This worker lost its claim or overran its cap: it is in
                        # trouble, so the rest of the batch goes back to the queue
                        # for a worker that is not.
                        self._leases.release_waiting()
                        break
                except Exception as error:
                    logger.error(
                        "Durable job could not be finalized",
                        exc_info=error,
                        extra={"job_id": snapshot.id, "subscriber": snapshot.subscriber_id},
                    )
        finally:
            self._leases.clear()
        return len(snapshots)

    @staticmethod
    def _cursor_for(queue: str) -> str | None:
        """The rotation cursor for a claimed row's queue, or None when unusable.

        The name comes from the database, which may hold a row written by a host
        or an older tool. A name that is not a valid queue must restart the
        rotation rather than fail every later claim.
        """
        try:
            return validate_queue_name(queue)
        except ValueError:
            logger.warning("Durable job has a non-conforming queue name; restarting queue rotation")
            return None

    def _legacy_routing(self) -> tuple[tuple[str, ...] | None, tuple[str, ...] | None]:
        """Subscribers whose ``default``-queue rows this worker adopts or leaves.

        Read from the registry on every claim, so a subscriber registered after
        the runner was built is routed like the rest.
        """
        if self._queues is None:
            return None, None
        selected = self._registry.subscriber_ids_for_queues(self._queues)
        return selected, tuple(sorted(self._registry.subscriber_ids().difference(selected)))

    def _backfill_legacy_queues(self) -> None:
        """Retag pending ``default`` rows to their subscribers' queues, once.

        Best-effort: a failure is retried on the next batch and never stops a claim.
        """
        if self._legacy_queues_backfilled:
            return
        try:
            with self._session_factory() as db:
                moved = jobs_crud.backfill_legacy_queues(self._registry.queue_assignments(), db=db)
        except Exception as error:
            logger.warning("Durable job queue backfill failed; will retry", exc_info=error)
            return
        self._legacy_queues_backfilled = True
        if moved:
            logger.info("Moved %d pending durable job(s) to their subscribers' queues", moved)

    def _still_held(self, job: ClaimedJob) -> bool:
        """Renew the held leases; False when ``job`` is no longer this worker's.

        The batch was claimed under one lease, but jobs run in turn. A job the
        reaper already requeued belongs to another worker now and must not be run
        here as well. The same renewal routine runs on the worker heartbeat, so a
        waiting job normally has a fresh lease already; this is the confirmation
        that it is still held.
        """
        try:
            renewed = self._leases.renew()
        except Exception as error:
            # Not run: the lease is unknown, and the reaper requeues it.
            logger.error(
                "Durable job lease could not be renewed; skipping",
                exc_info=error,
                extra=self._context(job),
            )
            self._leases.drop(job.id)
            return False
        if job.id in renewed:
            return True
        logger.warning(
            "Durable job lease was lost before its turn; skipping",
            extra=self._context(job),
        )
        return False

    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) -> bool:
        """Run one job and record its outcome.

        Returns:
            True when the job's lease was lost while it ran, abandoned past
            ``max_runtime_seconds`` or fenced after failed renewals.
        """
        lost = self._leases.begin(job.id)
        try:
            self._mark_started(job)
            lost_claim = self._execute(job, lost)
        finally:
            # After the outcome is written, so the heartbeat keeps renewing the
            # lease through the completion write.
            self._leases.finish(job.id)
        return lost_claim or lost.is_set()

    def _mark_started(self, job: ClaimedJob) -> None:
        """Stamp when the handler began. Best-effort: history must not stop the job."""
        if not self._record_attempts:
            return
        try:
            with self._session_factory() as db:
                jobs_crud.mark_attempt_started(
                    job.id, worker_id=self._worker_id, attempt=job.attempts, now=self._clock.now(), db=db
                )
        except Exception as error:
            logger.warning("Durable job start could not be recorded", exc_info=error, extra=self._context(job))

    def _execute(self, job: ClaimedJob, lost: threading.Event) -> bool:
        """Run the handler and write its outcome.

        Returns:
            True when the outcome write found the claim no longer owned. This, not
            a renewal seeing the row finished, is what says the claim was lost once
            the handler has returned.
        """
        handler = self._registry.get(job.subscriber_id)
        event = self._event_from(job)
        began = self._clock.monotonic()
        try:
            if handler is None:
                raise LookupError(f"no durable handler registered for subscriber_id {job.subscriber_id!r}")
            with job_lease(lost):
                handler(event)
        except Exception as error:
            self._leases.finalize(job.id)
            return self._fail(job, error, duration_ms=self._elapsed_ms(began))
        self._leases.finalize(job.id)
        duration_ms = self._elapsed_ms(began)
        now = self._clock.now()
        with self._session_factory() as db:
            completed = jobs_crud.mark_job_completed(
                job.id,
                worker_id=self._worker_id,
                attempt=job.attempts,
                now=now,
                db=db,
                duration_ms=duration_ms,
                record_attempt=self._record_attempts,
            )
        if not completed:
            logger.warning(
                "Durable job completion skipped because claim ownership was lost",
                extra={**self._context(job), "duration_ms": duration_ms},
            )
        return not completed

    def _elapsed_ms(self, began: float) -> int:
        return max(0, round((self._clock.monotonic() - began) * 1000))

    def _context(self, job: ClaimedJob) -> dict[str, object]:
        """The structured fields every job-outcome log record carries."""
        return {
            "job_id": job.id,
            "subscriber": job.subscriber_id,
            "queue": job.queue,
            "attempts": job.attempts,
            "worker_id": self._worker_id,
        }

    def _fail(self, job: ClaimedJob, error: Exception, *, duration_ms: int | None = None) -> bool:
        """Record a failure; True when the claim was no longer owned."""
        now = self._clock.now()
        with self._session_factory() as db:
            status = jobs_crud.mark_job_failed(
                job.id,
                str(error),
                worker_id=self._worker_id,
                attempt=job.attempts,
                base_seconds=self._backoff_base_seconds,
                max_seconds=self._backoff_max_seconds,
                now=now,
                db=db,
                duration_ms=duration_ms,
                record_attempt=self._record_attempts,
            )
        if not status:
            logger.warning(
                "Durable job failure skipped because claim ownership was lost",
                exc_info=error,
                extra={**self._context(job), "duration_ms": duration_ms},
            )
            return True
        level = logging.ERROR if status == jobs_crud.STATUS_DEAD_LETTER else logging.WARNING
        logger.log(
            level,
            "Durable job failed",
            exc_info=error,
            extra={
                **self._context(job),
                "event_type": job.event_type,
                "job_status": status,
                "duration_ms": duration_ms,
            },
        )
        return False

    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,
            queue=job.queue,
        )

clock property

clock

Clock shared by claims and worker telemetry.

heartbeat_interval_seconds property

heartbeat_interval_seconds

How often the worker heartbeat renews leases and refreshes telemetry.

leases property

leases

The record of what this worker holds, renewed by the worker heartbeat.

queues property

queues

Normalized queue allowlist, or None when all queues are selected.

worker_id property

worker_id

Restart-unique worker and lease-holder identity.

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(stop=None)

Claim and process one batch of due jobs.

Parameters:

Name	Type	Description	Default
stop	Event | None	When set between jobs, the jobs not yet started are released back to the queue instead of being run, so a graceful shutdown does not hold them for the rest of the batch.	None

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, stop: threading.Event | None = None) -> int:
    """
    Claim and process one batch of due jobs.

    Args:
        stop: When set between jobs, the jobs not yet started are released
            back to the queue instead of being run, so a graceful shutdown
            does not hold them for the rest of the batch.

    Returns:
        The number of jobs processed in this batch (0 when none were due).
    """
    now = self._clock.now()
    self._backfill_legacy_queues()
    default_subscribers, excluded_subscribers = self._legacy_routing()
    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,
            queues=self._queues,
            queue_cursor=self._queue_cursor,
            default_queue_subscribers=default_subscribers,
            excluded_default_queue_subscribers=excluded_subscribers,
            record_attempts=self._record_attempts,
        )
        snapshots = [self._snapshot(job) for job in claimed]
    if not snapshots:
        return 0
    self._queue_cursor = self._cursor_for(snapshots[-1].queue)
    logger.debug("Claimed %d durable job(s) as %s", len(snapshots), self._worker_id)
    self._leases.hold({snapshot.id: snapshot.attempts for snapshot in snapshots})
    try:
        for index, snapshot in enumerate(snapshots):
            # A DB error finishing one job must not abort the rest of the batch;
            # its lease simply expires and the reaper requeues it.
            if stop is not None and stop.is_set():
                self._leases.release_waiting()
                break
            try:
                # The first job's lease was taken a moment ago; every later one
                # has been waiting, so it is renewed (and confirmed held) first.
                if index and not self._still_held(snapshot):
                    continue
                if self._run_job(snapshot):
                    # This worker lost its claim or overran its cap: it is in
                    # trouble, so the rest of the batch goes back to the queue
                    # for a worker that is not.
                    self._leases.release_waiting()
                    break
            except Exception as error:
                logger.error(
                    "Durable job could not be finalized",
                    exc_info=error,
                    extra={"job_id": snapshot.id, "subscriber": snapshot.subscriber_id},
                )
    finally:
        self._leases.clear()
    return len(snapshots)

jasil.jobs.service

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

Consumed by application startup (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 compare-and-set claims and 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 durable jobs are enabled.

build_runner

build_runner(*, queues=None)

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.

Parameters:

Name	Type	Description	Default
queues	Iterable[str] | None	Optional non-empty queue allowlist. Omit it to consume all queues.	None

Raises:

Type	Description
ValueError	When jobs.heartbeat_interval_seconds, jobs.lease_seconds and jobs.max_runtime_seconds cannot keep a lease alive (see :func:_validate_job_timing).

Source code in jasil/jobs/service.py

def build_runner(*, queues: Iterable[str] | None = None) -> "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.

    Args:
        queues: Optional non-empty queue allowlist. Omit it to consume all queues.

    Raises:
        ValueError: When ``jobs.heartbeat_interval_seconds``,
            ``jobs.lease_seconds`` and ``jobs.max_runtime_seconds`` cannot keep a
            lease alive (see :func:`_validate_job_timing`).
    """
    from jasil.jobs.runner import JobRunner

    jobs = get_settings().jobs
    _validate_job_timing(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,
        max_runtime_seconds=jobs.max_runtime_seconds,
        heartbeat_interval_seconds=jobs.heartbeat_interval_seconds,
        batch_size=jobs.batch_size,
        backoff_base_seconds=jobs.backoff_base_seconds,
        backoff_max_seconds=jobs.backoff_max_seconds,
        queues=queues,
        record_attempts=jobs.record_attempts,
    )

reap_expired_jobs_scheduled

reap_expired_jobs_scheduled()

Requeue or dead-letter jobs with expired leases.

Runs on every replica; reclaim_expired_leases combines FOR UPDATE SKIP LOCKED with a compare-and-set on status, so concurrent reapers reclaim disjoint rows and a loser never overwrites the winner's requeue.

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`` combines ``FOR UPDATE SKIP
    LOCKED`` with a compare-and-set on ``status``, so concurrent reapers reclaim
    disjoint rows and a loser never overwrites the winner's requeue.
    """
    import jasil.jobs.crud as jobs_crud

    platform = platform_runtime.get_active_platform()
    now = platform.clock.now()
    with jasil_orm.get_sessionmaker()() as db:
        # Read before reclaiming, which clears the holder: a crash should name the
        # worker, not just say that one happened.
        holders = jobs_crud.expired_lease_holders(now=now, db=db)
        reclaimed = jobs_crud.reclaim_expired_leases(now=now, db=db)
    if reclaimed:
        # A warning, not an info: a lease only expires because the worker holding
        # it died or went silent, so this is the symptom of something else going wrong.
        logger.warning(
            "Reaped %d expired job lease(s); a worker died mid-job or went silent",
            reclaimed,
            extra={"workers": holders},
        )

relay_outbox_scheduled

relay_outbox_scheduled()

Relay the outbox into per-subscriber jobs.

Runs on every replica; each pass claims its batch with FOR UPDATE SKIP LOCKED held across the fan-out, and the idempotent fan-out dedups any overlap where the dialect cannot lock, 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; each pass claims its batch with ``FOR UPDATE SKIP
    LOCKED`` held across the fan-out, and the idempotent fan-out dedups any
    overlap where the dialect cannot lock, so no single-runner lock is needed.
    """
    import jasil.jobs.relay as jobs_relay

    jobs = get_settings().jobs
    platform = platform_runtime.get_active_platform()
    total = 0
    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,
        )
        total += relayed
        if relayed == 0:
            break
    if total:
        # Debug, not info: this runs every few seconds, and "an event was
        # published but never processed" is the question it exists to answer.
        logger.debug("Relayed %d outbox row(s) into durable jobs", total)

run_job_worker

run_job_worker(
    *,
    queues=None,
    role=None,
    label=None,
    metadata=None,
    stop=None,
)

Run a blocking standalone durable-job worker.

This is the supported process entrypoint for distributed deployments. The host configures JASIL, registers its durable subscribers, and calls this function without importing runner or CRUD internals. Register every durable subscriber first, including those on queues this worker does not select: the worker re-reads the registry on every claim, but retags legacy default rows from it once, on its first batch. Signal handling remains host-owned: pass an event that the host sets during graceful shutdown. This function blocks until that event is set; run it only as a dedicated worker process entrypoint, never on an API request or event-loop thread.

Parameters:

Name	Type	Description	Default
queues	Iterable[str] | None	Optional non-empty queue allowlist. Omit it to consume all queues.	None
role	str | None	Optional host-supplied worker role.	None
label	str | None	Optional host-supplied operator label.	None
metadata	dict[str, Any] | None	Optional host-supplied neutral metadata.	None
stop	Event | None	Event ending the loop; a new unset event is used when omitted.	None

Raises:

Type	Description
RuntimeError	When job_worker_mode is not standalone, or when the configured database is SQLite, which supports only a single-process topology.
ValueError	When an explicit queue allowlist is empty or invalid.

Source code in jasil/jobs/service.py

def run_job_worker(
    *,
    queues: Iterable[str] | None = None,
    role: str | None = None,
    label: str | None = None,
    metadata: dict[str, Any] | None = None,
    stop: threading.Event | None = None,
) -> None:
    """Run a blocking standalone durable-job worker.

    This is the supported process entrypoint for distributed deployments. The
    host configures JASIL, registers its durable subscribers, and calls this
    function without importing runner or CRUD internals. Register every durable
    subscriber first, including those on queues this worker does not select: the
    worker re-reads the registry on every claim, but retags legacy ``default`` rows
    from it once, on its first batch. Signal handling remains
    host-owned: pass an event that the host sets during graceful shutdown. This
    function blocks until that event is set; run it only as a dedicated worker
    process entrypoint, never on an API request or event-loop thread.

    Args:
        queues: Optional non-empty queue allowlist. Omit it to consume all queues.
        role: Optional host-supplied worker role.
        label: Optional host-supplied operator label.
        metadata: Optional host-supplied neutral metadata.
        stop: Event ending the loop; a new unset event is used when omitted.

    Raises:
        RuntimeError: When ``job_worker_mode`` is not ``standalone``, or when the
            configured database is SQLite, which supports only a single-process
            topology.
        ValueError: When an explicit queue allowlist is empty or invalid.
    """
    _ensure_job_topology(JobWorkerMode.STANDALONE, entry_point="run_job_worker", other="start_job_worker")
    runner = build_runner(queues=queues)
    from jasil.jobs.worker import run_worker

    jobs = get_settings().jobs
    heartbeat = _build_heartbeat(runner, role=role, label=label, metadata=metadata)
    run_worker(
        runner,
        poll_interval_seconds=jobs.poll_interval_seconds,
        stop=stop or threading.Event(),
        heartbeat=heartbeat,
    )

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(*, role=None, label=None, metadata=None)

Start the in-process job worker (idempotent).

Belongs to job_worker_mode=in_process, the default: each API process runs its own consumer. A deployment with dedicated worker processes declares standalone and uses :func:run_job_worker instead.

Parameters:

Name	Type	Description	Default
role	str | None	Optional host-supplied worker role.	None
label	str | None	Optional host-supplied operator label.	None
metadata	dict[str, Any] | None	Optional host-supplied neutral metadata.	None

Raises:

Type	Description
RuntimeError	When job_worker_mode is not in_process, or when the configured database is SQLite and the topology is multi-process.

Source code in jasil/jobs/service.py

def start_job_worker(
    *,
    role: str | None = None,
    label: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> None:
    """Start the in-process job worker (idempotent).

    Belongs to ``job_worker_mode=in_process``, the default: each API process runs
    its own consumer. A deployment with dedicated worker processes declares
    ``standalone`` and uses :func:`run_job_worker` instead.

    Args:
        role: Optional host-supplied worker role.
        label: Optional host-supplied operator label.
        metadata: Optional host-supplied neutral metadata.

    Raises:
        RuntimeError: When ``job_worker_mode`` is not ``in_process``, or when the
            configured database is SQLite and the topology is multi-process.
    """
    global _worker
    if _worker is not None:
        if _worker.is_alive:
            return
        _worker = None
    _ensure_job_topology(JobWorkerMode.IN_PROCESS, entry_point="start_job_worker", other="run_job_worker")
    from jasil.jobs.worker import BackgroundWorker

    # The loop logs its own start and stop, from the thread that actually runs it.
    runner = build_runner()
    jobs = get_settings().jobs
    heartbeat = _build_heartbeat(runner, role=role, label=label, metadata=metadata)
    _worker = BackgroundWorker(runner, poll_interval_seconds=jobs.poll_interval_seconds, heartbeat=heartbeat)
    _worker.start()

stop_job_worker

stop_job_worker()

Stop the in-process worker from synchronous shutdown code.

Source code in jasil/jobs/service.py

def stop_job_worker() -> None:
    """Stop the in-process worker from synchronous shutdown code."""
    global _worker
    if _worker is None:
        return
    worker = _worker
    if worker.stop() and _worker is worker:
        _worker = None

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, plus stopped/stale worker telemetry.

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.

Call it yourself, or register it with :func:schedule_retention_maintenance. Each window is applied independently: JasilSettings.event_log.retention_days gates the event_log trail and JasilSettings.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.

    Call it yourself, or register it with
    :func:`schedule_retention_maintenance`. Each window is applied
    independently: ``JasilSettings.event_log.retention_days`` gates the event_log
    trail and ``JasilSettings.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)

schedule_retention_maintenance

schedule_retention_maintenance(
    scheduler, *, run_at_startup=True
)

Register the recurring retention prune on the host's scheduler.

The counterpart of :func:jasil.jobs.service.schedule_job_maintenance, kept separate because retention also prunes the event_log — so it applies to a deployment that never enabled durable jobs.

Register it on every replica. The prune takes the platform's coordination lock, so only one of them does the work per pass.

Parameters:

Name	Type	Description	Default
scheduler	AsyncIOScheduler	The application scheduler to register the job on.	required
run_at_startup	bool	Run one pass as soon as the scheduler starts. On by default because a daily interval otherwise means a process that is redeployed daily never prunes at all.	True

Returns:

Type	Description
None	None.

Source code in jasil/retention.py

def schedule_retention_maintenance(scheduler: "AsyncIOScheduler", *, run_at_startup: bool = True) -> None:
    """
    Register the recurring retention prune on the host's scheduler.

    The counterpart of :func:`jasil.jobs.service.schedule_job_maintenance`, kept
    separate because retention also prunes the event_log — so it applies to a
    deployment that never enabled durable jobs.

    Register it on every replica. The prune takes the platform's coordination
    lock, so only one of them does the work per pass.

    Args:
        scheduler: The application scheduler to register the job on.
        run_at_startup: Run one pass as soon as the scheduler starts. On by
            default because a daily interval otherwise means a process that is
            redeployed daily never prunes at all.

    Returns:
        None.
    """
    starts_now = {"next_run_time": datetime.now(UTC)} if run_at_startup else {}
    scheduler.add_job(
        prune_expired_records,
        "interval",
        hours=_PRUNE_INTERVAL_HOURS,
        id=_PRUNE_JOB_ID,
        replace_existing=True,
        **starts_now,
    )
    logger.info("Scheduled JASIL retention pruning")

Admin and dashboards

jasil.admin

The operator-facing surface: dashboard aggregates and dead-letter replay.

The event-log and durable-job tables are what an operator needs to see, and the queries that aggregate them live in :mod:jasil.event_log.crud and :mod:jasil.jobs.crud. Neither is a good thing for a host to wire a route to: both reach a model at import time, so importing either before :func:jasil.orm.map_models has run raises, and both take a session they then commit — which, handed a session that already carries an admin request's own uncommitted work, commits that too.

This module is the stable seam in front of them. It is safe to import from anywhere in the host's import graph (the model imports are deferred into the functions), and nothing here accepts a session: each call opens a short-lived one of its own, so a host cannot hand JASIL the transaction its request is in the middle of::

import jasil.admin as jasil_admin

@router.get("/admin/jobs")
def jobs_dashboard() -> jasil_admin.JobsSummary:
    return jasil_admin.get_jobs_summary()

@router.post("/admin/jobs/{job_id}/replay")
def replay(job_id: str) -> jasil_admin.JobReplayResult:
    return jasil_admin.replay_dead_letter_job(job_id)

The response schemas are re-exported here so a host can type its routes without importing from an internal module. Authentication and authorization are the host's: these functions expose operational data and one state-changing action, and JASIL has no notion of who is calling.

DeadLetterJob

Bases: BaseModel

A dead-lettered job, shown for inspection and replay.

Attributes:

Name	Type	Description
id	str	The job id (used to replay it).
event_id	str	The originating envelope event_id.
event_type	str	The event channel.
subscriber_id	str	The subscriber that failed.
source	str	Where the originating event came from.
attempts	int	Attempts made before dead-lettering.
max_attempts	int	The attempt ceiling that was reached.
last_error	str | None	The final failure reason.
created_at	datetime	When the job was enqueued.
updated_at	datetime	When the job was dead-lettered.
completed_at	datetime | None	When the job reached its terminal state.
last_worker_id	str | None	The worker instance that ran its final attempt; None when attempts are not recorded or the history was pruned.

Source code in jasil/jobs/schema.py

class DeadLetterJob(BaseModel):
    """
    A dead-lettered job, shown for inspection and replay.

    Attributes:
        id: The job id (used to replay it).
        event_id: The originating envelope event_id.
        event_type: The event channel.
        subscriber_id: The subscriber that failed.
        source: Where the originating event came from.
        attempts: Attempts made before dead-lettering.
        max_attempts: The attempt ceiling that was reached.
        last_error: The final failure reason.
        created_at: When the job was enqueued.
        updated_at: When the job was dead-lettered.
        completed_at: When the job reached its terminal state.
        last_worker_id: The worker instance that ran its final attempt; ``None``
            when attempts are not recorded or the history was pruned.
    """

    model_config = ConfigDict(from_attributes=True)

    id: str
    event_id: str
    event_type: str
    subscriber_id: str
    source: str
    attempts: int
    max_attempts: int
    last_error: str | None
    created_at: datetime
    updated_at: datetime
    completed_at: datetime | None
    last_worker_id: str | None = None

DurationBuckets

Bases: BaseModel

How long completed attempts took, in fixed ranges.

The boundaries are part of this schema, so two deployments' dashboards are comparable. Counts are non-cumulative and sum to the completed attempts in the window. Hosts needing other boundaries or true percentiles compute them from the duration_ms log field.

Attributes:

Name	Type	Description
under_100ms	int	< 100 ms.
under_1s	int	100 ms up to < 1 s.
under_10s	int	1 s up to < 10 s.
under_1m	int	10 s up to < 60 s.
under_5m	int	60 s up to < 300 s.
over_5m	int	>= 300 s.

Source code in jasil/jobs/schema.py

class DurationBuckets(BaseModel):
    """
    How long completed attempts took, in fixed ranges.

    The boundaries are part of this schema, so two deployments' dashboards are
    comparable. Counts are non-cumulative and sum to the completed attempts in
    the window. Hosts needing other boundaries or true percentiles compute them
    from the ``duration_ms`` log field.

    Attributes:
        under_100ms: ``< 100 ms``.
        under_1s: ``100 ms`` up to ``< 1 s``.
        under_10s: ``1 s`` up to ``< 10 s``.
        under_1m: ``10 s`` up to ``< 60 s``.
        under_5m: ``60 s`` up to ``< 300 s``.
        over_5m: ``>= 300 s``.
    """

    under_100ms: int = 0
    under_1s: int = 0
    under_10s: int = 0
    under_1m: int = 0
    under_5m: int = 0
    over_5m: int = 0

EventJobs

Bases: BaseModel

Every subscriber's job for one event, each with its attempts.

Answers "which worker processed this event, for each subscriber?". An event's event_log row only records publication, so this is where its outcome lives.

Source code in jasil/jobs/schema.py

class EventJobs(BaseModel):
    """Every subscriber's job for one event, each with its attempts.

    Answers "which worker processed this event, for each subscriber?". An event's
    ``event_log`` row only records publication, so this is where its outcome lives.
    """

    event_id: str
    jobs: list[JobDetail]
    attempts_recorded: bool = True

EventLogFailure

Bases: BaseModel

A single failed or dead-lettered event for inspection.

Attributes:

Name	Type	Description
id	str	The event_id.
event_type	str	The domain-event channel.
event_source	str	Where the event originated.
handler_name	str | None	The subscriber(s) that processed the event.
error_message	str | None	The failure reason.
retry_count	int	Processing attempts so far.
event_metadata	dict[str, Any] | None	Correlation context (request_id, plus any host-defined keys).
created_at	datetime	When the event was published.
completed_at	datetime | None	When processing finished.

Source code in jasil/event_log/schema.py

class EventLogFailure(BaseModel):
    """
    A single failed or dead-lettered event for inspection.

    Attributes:
        id: The event_id.
        event_type: The domain-event channel.
        event_source: Where the event originated.
        handler_name: The subscriber(s) that processed the event.
        error_message: The failure reason.
        retry_count: Processing attempts so far.
        event_metadata: Correlation context (request_id, plus any host-defined keys).
        created_at: When the event was published.
        completed_at: When processing finished.
    """

    model_config = ConfigDict(from_attributes=True)

    id: str
    event_type: str
    event_source: str
    handler_name: str | None
    error_message: str | None
    retry_count: int
    event_metadata: dict[str, Any] | None
    created_at: datetime
    completed_at: datetime | None

EventLogPending

Bases: BaseModel

A group of not-yet-finished events and its oldest age.

Attributes:

Name	Type	Description
event_type	str	The domain-event channel.
status	str	The pending state (published or processing).
count	int	Number of events in this group.
oldest_seconds	float | None	Age of the oldest event in the group, in seconds.

Source code in jasil/event_log/schema.py

class EventLogPending(BaseModel):
    """
    A group of not-yet-finished events and its oldest age.

    Attributes:
        event_type: The domain-event channel.
        status: The pending state (published or processing).
        count: Number of events in this group.
        oldest_seconds: Age of the oldest event in the group, in seconds.
    """

    event_type: str
    status: str
    count: int
    oldest_seconds: float | None

EventLogSummary

Bases: BaseModel

The full admin-dashboard payload, aggregated from event_log.

Attributes:

Name	Type	Description
window_hours	int	The look-back window applied to the aggregates.
total_events	int	Total events recorded within the window.
by_type	list[EventTypeStats]	Per-event-type throughput/outcome/latency stats.
pending	list[EventLogPending]	Not-yet-finished event groups and their oldest age.
recent_failures	list[EventLogFailure]	The most recent failed/dead-lettered events.

Source code in jasil/event_log/schema.py

class EventLogSummary(BaseModel):
    """
    The full admin-dashboard payload, aggregated from event_log.

    Attributes:
        window_hours: The look-back window applied to the aggregates.
        total_events: Total events recorded within the window.
        by_type: Per-event-type throughput/outcome/latency stats.
        pending: Not-yet-finished event groups and their oldest age.
        recent_failures: The most recent failed/dead-lettered events.
    """

    window_hours: int
    total_events: int
    by_type: list[EventTypeStats]
    pending: list[EventLogPending]
    recent_failures: list[EventLogFailure]

EventTypeStats

Bases: BaseModel

Per-event-type throughput, outcome, and latency counts.

Attributes:

Name	Type	Description
event_type	str	The domain-event channel.
total	int	Total events of this type in the window.
published	int	Count still in the published state.
queued	int	Count handed to the durable job queue (terminal in event_log; execution is tracked per-subscriber in the Jobs dashboard).
processing	int	Count currently processing.
completed	int	Count that finished successfully.
failed	int	Count that failed.
dead_letter	int	Count moved to dead-letter.
avg_processing_time_ms	float | None	Mean handler time, or None when unmeasured.
max_processing_time_ms	int | None	Slowest handler time, or None.

Source code in jasil/event_log/schema.py

class EventTypeStats(BaseModel):
    """
    Per-event-type throughput, outcome, and latency counts.

    Attributes:
        event_type: The domain-event channel.
        total: Total events of this type in the window.
        published: Count still in the published state.
        queued: Count handed to the durable job queue (terminal in event_log;
            execution is tracked per-subscriber in the Jobs dashboard).
        processing: Count currently processing.
        completed: Count that finished successfully.
        failed: Count that failed.
        dead_letter: Count moved to dead-letter.
        avg_processing_time_ms: Mean handler time, or None when unmeasured.
        max_processing_time_ms: Slowest handler time, or None.
    """

    event_type: str
    total: int
    published: int
    queued: int
    processing: int
    completed: int
    failed: int
    dead_letter: int
    avg_processing_time_ms: float | None
    max_processing_time_ms: int | None

JobAttemptInfo

Bases: BaseModel

One attempt to run a durable job.

Attributes:

Name	Type	Description
id	str	The attempt id.
job_id	str	The job it ran.
event_id	str	The originating event.
event_type	str	The event channel.
subscriber_id	str	The subscriber that ran.
queue	str	The queue the job was claimed from.
attempt	int	The attempt number at claim.
worker_id	str	The worker instance that claimed it.
claimed_at	datetime	When the worker took the lease.
started_at	datetime | None	When the handler began; None if it never ran.
finished_at	datetime | None	When the attempt ended; None while it is open.
outcome	str | None	One of completed, failed, dead_letter, lease_expired, superseded, released, abandoned; None while open.
duration_ms	int | None	Handler wall time.
error	str | None	The failure reason, if it failed.

Source code in jasil/jobs/schema.py

class JobAttemptInfo(BaseModel):
    """
    One attempt to run a durable job.

    Attributes:
        id: The attempt id.
        job_id: The job it ran.
        event_id: The originating event.
        event_type: The event channel.
        subscriber_id: The subscriber that ran.
        queue: The queue the job was claimed from.
        attempt: The attempt number at claim.
        worker_id: The worker instance that claimed it.
        claimed_at: When the worker took the lease.
        started_at: When the handler began; ``None`` if it never ran.
        finished_at: When the attempt ended; ``None`` while it is open.
        outcome: One of ``completed``, ``failed``, ``dead_letter``,
            ``lease_expired``, ``superseded``, ``released``, ``abandoned``;
            ``None`` while open.
        duration_ms: Handler wall time.
        error: The failure reason, if it failed.
    """

    model_config = ConfigDict(from_attributes=True)

    id: str
    job_id: str
    event_id: str
    event_type: str
    subscriber_id: str
    queue: str
    attempt: int
    worker_id: str
    claimed_at: datetime
    started_at: datetime | None
    finished_at: datetime | None
    outcome: str | None
    duration_ms: int | None
    error: str | None

JobDetail

Bases: BaseModel

One job and its attempt history, newest attempt first.

Attributes:

Name	Type	Description
id	str	The job id.
event_id	str	The originating event.
event_type	str	The event channel.
subscriber_id	str	The durable subscriber.
queue	str	The named queue.
source	str	Where the event came from.
status	str	pending, claimed, completed or dead_letter.
attempts	int	Attempts made so far.
max_attempts	int	The attempt ceiling.
locked_by	str | None	The worker holding it now, while claimed.
last_error	str | None	The most recent failure reason.
available_at	datetime	Earliest instant it may be claimed.
lease_expires_at	datetime | None	When the current lease expires, while claimed.
created_at	datetime	When it was enqueued.
updated_at	datetime	When it last changed state.
completed_at	datetime | None	When it reached a terminal state.
history	list[JobAttemptInfo]	Its attempts, newest first, bounded by the caller's limit.
attempts_recorded	bool	Whether this process records attempts. False means history may be empty because nothing was written, not because nothing ran.

Source code in jasil/jobs/schema.py

class JobDetail(BaseModel):
    """
    One job and its attempt history, newest attempt first.

    Attributes:
        id: The job id.
        event_id: The originating event.
        event_type: The event channel.
        subscriber_id: The durable subscriber.
        queue: The named queue.
        source: Where the event came from.
        status: ``pending``, ``claimed``, ``completed`` or ``dead_letter``.
        attempts: Attempts made so far.
        max_attempts: The attempt ceiling.
        locked_by: The worker holding it now, while claimed.
        last_error: The most recent failure reason.
        available_at: Earliest instant it may be claimed.
        lease_expires_at: When the current lease expires, while claimed.
        created_at: When it was enqueued.
        updated_at: When it last changed state.
        completed_at: When it reached a terminal state.
        history: Its attempts, newest first, bounded by the caller's limit.
        attempts_recorded: Whether this process records attempts. ``False`` means
            ``history`` may be empty because nothing was written, not because
            nothing ran.
    """

    model_config = ConfigDict(from_attributes=True)

    id: str
    event_id: str
    event_type: str
    subscriber_id: str
    queue: str
    source: str
    status: str
    attempts: int
    max_attempts: int
    locked_by: str | None
    last_error: str | None
    available_at: datetime
    lease_expires_at: datetime | None
    created_at: datetime
    updated_at: datetime
    completed_at: datetime | None
    history: list[JobAttemptInfo] = Field(default_factory=list)
    attempts_recorded: bool = True

JobQueueStats

Bases: BaseModel

Windowed job counts and all-time current backlog age for one named queue.

Source code in jasil/jobs/schema.py

class JobQueueStats(BaseModel):
    """Windowed job counts and all-time current backlog age for one named queue."""

    queue: str
    total: int
    pending: int
    claimed: int
    completed: int
    dead_letter: int
    oldest_pending_seconds: float | None
    avg_duration_ms: float | None = None
    max_duration_ms: int | None = None
    duration_buckets: DurationBuckets | None = None
    superseded: int | None = None

JobReplayResult

Bases: BaseModel

The outcome of replaying a dead-lettered job.

Attributes:

Name	Type	Description
replayed	bool	True when the job was requeued for a fresh run.

Source code in jasil/jobs/schema.py

class JobReplayResult(BaseModel):
    """
    The outcome of replaying a dead-lettered job.

    Attributes:
        replayed: True when the job was requeued for a fresh run.
    """

    replayed: bool

JobSubscriberStats

Bases: BaseModel

Per-subscriber job counts by status within the window.

Attributes:

Name	Type	Description
subscriber_id	str	The durable subscriber.
event_type	str	The event channel it reacts to.
total	int	Total jobs for this subscriber in the window.
pending	int	Count waiting to be claimed (includes backoff).
claimed	int	Count currently leased by a worker.
completed	int	Count that finished successfully.
dead_letter	int	Count that exhausted retries.
avg_duration_ms	float | None	Mean handler time of attempts that completed in the window; None when none did or attempts are not recorded.
max_duration_ms	int | None	Slowest such attempt.
duration_buckets	DurationBuckets | None	The same attempts in fixed duration ranges; None when attempts are not recorded.
superseded	int | None	Attempts that ended in the window by finishing after their job was taken (the handler ran twice); None when attempts are not recorded.

Source code in jasil/jobs/schema.py

class JobSubscriberStats(BaseModel):
    """
    Per-subscriber job counts by status within the window.

    Attributes:
        subscriber_id: The durable subscriber.
        event_type: The event channel it reacts to.
        total: Total jobs for this subscriber in the window.
        pending: Count waiting to be claimed (includes backoff).
        claimed: Count currently leased by a worker.
        completed: Count that finished successfully.
        dead_letter: Count that exhausted retries.
        avg_duration_ms: Mean handler time of attempts that completed in the
            window; ``None`` when none did or attempts are not recorded.
        max_duration_ms: Slowest such attempt.
        duration_buckets: The same attempts in fixed duration ranges; ``None``
            when attempts are not recorded.
        superseded: Attempts that ended in the window by finishing after their
            job was taken (the handler ran twice); ``None`` when attempts are not
            recorded.
    """

    subscriber_id: str
    event_type: str
    total: int
    pending: int
    claimed: int
    completed: int
    dead_letter: int
    avg_duration_ms: float | None = None
    max_duration_ms: int | None = None
    duration_buckets: DurationBuckets | None = None
    superseded: int | None = None

JobsSummary

Bases: BaseModel

The durable-jobs admin-dashboard payload.

Attributes:

Name	Type	Description
window_hours	int	The look-back window applied to the counts.
total_jobs	int	Total jobs enqueued within the window.
pending	int	Window count waiting to be claimed.
claimed	int	Window count currently leased.
completed	int	Window count finished successfully.
dead_letter	int	Window count that exhausted retries.
oldest_pending_seconds	float | None	Age of the oldest unfinished job, in seconds.
by_subscriber	list[JobSubscriberStats]	Per-subscriber breakdown within the window.
by_queue	list[JobQueueStats]	Per-queue breakdown within the window.
recent_dead_letter	list[DeadLetterJob]	The current dead-letter queue contents (most recent first).

Source code in jasil/jobs/schema.py

class JobsSummary(BaseModel):
    """
    The durable-jobs admin-dashboard payload.

    Attributes:
        window_hours: The look-back window applied to the counts.
        total_jobs: Total jobs enqueued within the window.
        pending: Window count waiting to be claimed.
        claimed: Window count currently leased.
        completed: Window count finished successfully.
        dead_letter: Window count that exhausted retries.
        oldest_pending_seconds: Age of the oldest unfinished job, in seconds.
        by_subscriber: Per-subscriber breakdown within the window.
        by_queue: Per-queue breakdown within the window.
        recent_dead_letter: The current dead-letter queue contents (most recent first).
    """

    window_hours: int
    total_jobs: int
    pending: int
    claimed: int
    completed: int
    dead_letter: int
    oldest_pending_seconds: float | None
    by_subscriber: list[JobSubscriberStats]
    by_queue: list[JobQueueStats] = Field(default_factory=list)
    recent_dead_letter: list[DeadLetterJob]

WorkerAttempts

Bases: BaseModel

One bounded page of a worker instance's attempts, newest first.

Source code in jasil/jobs/schema.py

class WorkerAttempts(BaseModel):
    """One bounded page of a worker instance's attempts, newest first."""

    instance_id: str
    attempts: list[JobAttemptInfo]
    next_cursor: str | None = None
    attempts_recorded: bool = True

WorkerInfo

Bases: BaseModel

Operator-facing state for one restart-unique worker instance.

The completed .. abandoned counts are attempts that ended within the summary's window, and are None when attempts are not recorded.

Source code in jasil/jobs/schema.py

class WorkerInfo(BaseModel):
    """Operator-facing state for one restart-unique worker instance.

    The ``completed`` .. ``abandoned`` counts are attempts that *ended* within
    the summary's window, and are ``None`` when attempts are not recorded.
    """

    instance_id: str
    started_at: datetime
    last_heartbeat_at: datetime
    stopped_at: datetime | None
    queues: list[str] | None
    role: str | None
    label: str | None
    metadata: dict[str, Any] | None
    active_claimed_jobs: int
    status: WorkerStatus
    completed: int | None = None
    failed: int | None = None
    dead_lettered: int | None = None
    lease_expired: int | None = None
    superseded: int | None = None
    released: int | None = None
    abandoned: int | None = None
    last_finished_at: datetime | None = None

WorkersSummary

Bases: BaseModel

Global worker totals and one bounded page of retained telemetry.

Source code in jasil/jobs/schema.py

class WorkersSummary(BaseModel):
    """Global worker totals and one bounded page of retained telemetry."""

    stale_after_seconds: float
    window_hours: int = 24
    claimed_by_stale_workers: int = 0
    total_workers: int
    running: int
    stale: int
    stopped: int
    workers: list[WorkerInfo]
    next_cursor: str | None = None

get_event_jobs

get_event_jobs(event_id, *, attempt_limit=20)

Return every subscriber's job for one event, each with its attempts.

Answers "which worker processed this event, for each subscriber?". An event's event_log row records its publication only; a durable event's outcome is the state of these jobs.

Parameters:

Name	Type	Description	Default
event_id	str	The event id.	required
attempt_limit	int	Most attempts to return per job, from 1 through 500.	20

Returns:

Type	Description
EventJobs	The event's jobs (at most 200), oldest first; empty when the event has
EventJobs	none.

Raises:

Type	Description
ValueError	When attempt_limit is out of range.

Source code in jasil/admin.py

def get_event_jobs(event_id: str, *, attempt_limit: int = 20) -> EventJobs:
    """Return every subscriber's job for one event, each with its attempts.

    Answers "which worker processed this event, for each subscriber?". An event's
    ``event_log`` row records its publication only; a durable event's outcome is
    the state of these jobs.

    Args:
        event_id: The event id.
        attempt_limit: Most attempts to return per job, from 1 through 500.

    Returns:
        The event's jobs (at most 200), oldest first; empty when the event has
        none.

    Raises:
        ValueError: When ``attempt_limit`` is out of range.
    """
    import jasil.jobs._attempts as attempts

    with jasil_orm.get_sessionmaker()() as db:
        return attempts.get_event_jobs(
            event_id,
            db=db,
            attempt_limit=attempt_limit,
            record_attempts=get_settings().jobs.record_attempts,
        )

get_event_log_summary

get_event_log_summary(*, hours=24, failure_limit=20)

Aggregate event_log into the observability dashboard payload.

Parameters:

Name	Type	Description	Default
hours	int	Look-back window for the throughput, outcome, and latency stats.	24
failure_limit	int	Maximum recent failures to return.	20

Returns:

Type	Description
EventLogSummary	Per-event-type statistics, the not-yet-finished groups and their oldest
EventLogSummary	age, and the most recent failures.

Raises:

Type	Description
RuntimeError	When jasil.orm.configure_sessionmaker has not run.

Source code in jasil/admin.py

def get_event_log_summary(*, hours: int = 24, failure_limit: int = 20) -> EventLogSummary:
    """Aggregate ``event_log`` into the observability dashboard payload.

    Args:
        hours: Look-back window for the throughput, outcome, and latency stats.
        failure_limit: Maximum recent failures to return.

    Returns:
        Per-event-type statistics, the not-yet-finished groups and their oldest
        age, and the most recent failures.

    Raises:
        RuntimeError: When ``jasil.orm.configure_sessionmaker`` has not run.
    """
    import jasil.event_log.crud as event_log_crud

    with jasil_orm.get_sessionmaker()() as db:
        return event_log_crud.get_event_log_summary(db, hours=hours, failure_limit=failure_limit)

get_job

get_job(job_id, *, attempt_limit=50)

Return one job and its attempt history, newest attempt first.

Answers "what happened to this job?": which worker ran each attempt, how long it took, how it ended, and why it failed.

Parameters:

Name	Type	Description	Default
job_id	str	The job id (as shown on a dead letter).	required
attempt_limit	int	Most attempts to return, from 1 through 500.	50

Returns:

Type	Description
JobDetail | None	The job with its history, or None when no such job exists. With
JobDetail | None	jobs.record_attempts off history holds only what was recorded
JobDetail | None	earlier and attempts_recorded is False.

Raises:

Type	Description
ValueError	When attempt_limit is out of range.

Source code in jasil/admin.py

def get_job(job_id: str, *, attempt_limit: int = 50) -> JobDetail | None:
    """Return one job and its attempt history, newest attempt first.

    Answers "what happened to this job?": which worker ran each attempt, how long
    it took, how it ended, and why it failed.

    Args:
        job_id: The job id (as shown on a dead letter).
        attempt_limit: Most attempts to return, from 1 through 500.

    Returns:
        The job with its history, or ``None`` when no such job exists. With
        ``jobs.record_attempts`` off ``history`` holds only what was recorded
        earlier and ``attempts_recorded`` is ``False``.

    Raises:
        ValueError: When ``attempt_limit`` is out of range.
    """
    import jasil.jobs._attempts as attempts

    with jasil_orm.get_sessionmaker()() as db:
        return attempts.get_job_detail(
            job_id,
            db=db,
            attempt_limit=attempt_limit,
            record_attempts=get_settings().jobs.record_attempts,
        )

get_jobs_summary

get_jobs_summary(*, hours=24, dead_letter_limit=50)

Aggregate processing_jobs into the durable-jobs dashboard payload.

Parameters:

Name	Type	Description	Default
hours	int	Look-back window for the status and per-subscriber counts.	24
dead_letter_limit	int	Maximum dead-letter jobs to return for inspection.	50

Returns:

Type	Description
JobsSummary	Window counts, the per-subscriber breakdown, the age of the oldest
JobsSummary	unfinished job, and the current dead-letter queue.

Raises:

Type	Description
RuntimeError	When jasil.orm.configure_sessionmaker has not run.

Source code in jasil/admin.py

def get_jobs_summary(*, hours: int = 24, dead_letter_limit: int = 50) -> JobsSummary:
    """Aggregate ``processing_jobs`` into the durable-jobs dashboard payload.

    Args:
        hours: Look-back window for the status and per-subscriber counts.
        dead_letter_limit: Maximum dead-letter jobs to return for inspection.

    Returns:
        Window counts, the per-subscriber breakdown, the age of the oldest
        unfinished job, and the current dead-letter queue.

    Raises:
        RuntimeError: When ``jasil.orm.configure_sessionmaker`` has not run.
    """
    import jasil.jobs.crud as jobs_crud

    with jasil_orm.get_sessionmaker()() as db:
        return jobs_crud.get_jobs_summary(
            db,
            hours=hours,
            dead_letter_limit=dead_letter_limit,
            record_attempts=get_settings().jobs.record_attempts,
        )

get_worker_attempts

get_worker_attempts(instance_id, *, limit=100, cursor=None)

Return one page of a worker instance's attempts, newest first.

Answers "what has this worker been doing, and how did it go?".

Parameters:

Name	Type	Description	Default
instance_id	str	The worker instance id (WorkerInfo.instance_id).	required
limit	int	Page size from 1 through 500.	100
cursor	str | None	Opaque cursor returned as next_cursor by the previous page.	None

Returns:

Type	Description
WorkerAttempts	The page of attempts; empty for an unknown worker.

Raises:

Type	Description
ValueError	When limit is out of range or cursor is not one this function returned.

Source code in jasil/admin.py

def get_worker_attempts(
    instance_id: str,
    *,
    limit: int = 100,
    cursor: str | None = None,
) -> WorkerAttempts:
    """Return one page of a worker instance's attempts, newest first.

    Answers "what has this worker been doing, and how did it go?".

    Args:
        instance_id: The worker instance id (``WorkerInfo.instance_id``).
        limit: Page size from 1 through 500.
        cursor: Opaque cursor returned as ``next_cursor`` by the previous page.

    Returns:
        The page of attempts; empty for an unknown worker.

    Raises:
        ValueError: When ``limit`` is out of range or ``cursor`` is not one
            this function returned.
    """
    import jasil.jobs._attempts as attempts

    with jasil_orm.get_sessionmaker()() as db:
        return attempts.get_worker_attempts(
            instance_id,
            db=db,
            limit=limit,
            cursor=cursor,
            record_attempts=get_settings().jobs.record_attempts,
        )

get_workers_summary

get_workers_summary(
    *,
    stale_after_seconds=None,
    limit=100,
    cursor=None,
    hours=24,
)

Return global worker totals and one bounded telemetry page.

Status is operator telemetry only: JASIL does not map it onto a host health endpoint or choose an alert policy.

Parameters:

Name	Type	Description	Default
stale_after_seconds	float | None	Maximum heartbeat age still considered running. Defaults to three configured heartbeat intervals.	None
limit	int	Page size from 1 through 500.	100
cursor	str | None	Opaque cursor returned as next_cursor by the previous page.	None
hours	int	Look-back window for each worker's attempt counts (attempts that ended within it) and last_finished_at.	24

Returns:

Type	Description
WorkersSummary	Worker counts and retained worker-instance details. The attempt counts
WorkersSummary	are None when this process does not record attempts, and
WorkersSummary	claimed_by_stale_workers counts jobs waiting for the reaper behind a
WorkersSummary	worker that stopped beating.

Source code in jasil/admin.py

def get_workers_summary(
    *,
    stale_after_seconds: float | None = None,
    limit: int = 100,
    cursor: str | None = None,
    hours: int = 24,
) -> WorkersSummary:
    """Return global worker totals and one bounded telemetry page.

    Status is operator telemetry only: JASIL does not map it onto a host health
    endpoint or choose an alert policy.

    Args:
        stale_after_seconds: Maximum heartbeat age still considered running.
            Defaults to three configured heartbeat intervals.
        limit: Page size from 1 through 500.
        cursor: Opaque cursor returned as ``next_cursor`` by the previous page.
        hours: Look-back window for each worker's attempt counts (attempts that
            ended within it) and ``last_finished_at``.

    Returns:
        Worker counts and retained worker-instance details. The attempt counts
        are ``None`` when this process does not record attempts, and
        ``claimed_by_stale_workers`` counts jobs waiting for the reaper behind a
        worker that stopped beating.
    """
    import jasil.jobs._worker_registry as worker_registry

    effective_stale_after = (
        stale_after_seconds if stale_after_seconds is not None else get_settings().jobs.heartbeat_interval_seconds * 3
    )
    now = platform_runtime.get_active_platform().clock.now()
    with jasil_orm.get_sessionmaker()() as db:
        return worker_registry.get_workers_summary(
            now=now,
            stale_after_seconds=effective_stale_after,
            db=db,
            limit=limit,
            cursor=cursor,
            hours=hours,
            record_attempts=get_settings().jobs.record_attempts,
        )

replay_dead_letter_job

replay_dead_letter_job(job_id)

Requeue a dead-lettered job for a fresh run with a full attempt budget.

Only a job currently in dead_letter is affected, so replaying one twice is harmless: the second call finds it pending and reports False.

Parameters:

Name	Type	Description	Default
job_id	str	The job to replay.	required

Returns:

Type	Description
JobReplayResult	Whether a dead-letter job was requeued.

Raises:

Type	Description
RuntimeError	When no platform has been published, or when jasil.orm.configure_sessionmaker has not run.

Source code in jasil/admin.py

def replay_dead_letter_job(job_id: str) -> JobReplayResult:
    """Requeue a dead-lettered job for a fresh run with a full attempt budget.

    Only a job currently in ``dead_letter`` is affected, so replaying one twice
    is harmless: the second call finds it ``pending`` and reports ``False``.

    Args:
        job_id: The job to replay.

    Returns:
        Whether a dead-letter job was requeued.

    Raises:
        RuntimeError: When no platform has been published, or when
            ``jasil.orm.configure_sessionmaker`` has not run.
    """
    import jasil.jobs.crud as jobs_crud

    now = platform_runtime.get_active_platform().clock.now()
    with jasil_orm.get_sessionmaker()() as db:
        replayed = jobs_crud.replay_dead_letter_job(job_id, now=now, db=db)
    return JobReplayResult(replayed=replayed)

Test helpers

jasil.testing

Test helpers for applications that embed JASIL.

JASIL installs several things process-wide — settings, a correlation provider, a session factory, the active platform, memoized Redis clients, the durable subscriber registry — and a test suite has to put every one of them back between cases or state leaks from one test into the next. This module is the shared version of the fixture every host would otherwise write.

Nothing here is imported by the library itself, and nothing here needs an optional extra::

import jasil.testing as jasil_testing

@pytest.fixture(autouse=True)
def _jasil(tmp_path):
    platform = jasil_testing.install_test_platform(tmp_path)
    yield platform
    jasil_testing.reset_all()

FixedClock dataclass

A ClockProvider whose time only moves when a test moves it.

Lease expiry, retry backoff, and retention windows are all measured against the platform clock, so injecting this is what lets a suite exercise them without sleeping.

monotonic advances in lockstep with now rather than tracking the real clock, so code measuring an elapsed duration sees the time the test moved.

Attributes:

Name	Type	Description
moment	datetime	The instant :meth:now returns.

Source code in jasil/testing.py

@dataclass
class FixedClock:
    """A ``ClockProvider`` whose time only moves when a test moves it.

    Lease expiry, retry backoff, and retention windows are all measured against
    the platform clock, so injecting this is what lets a suite exercise them
    without sleeping.

    ``monotonic`` advances in lockstep with ``now`` rather than tracking the real
    clock, so code measuring an elapsed duration sees the time the test moved.

    Attributes:
        moment: The instant :meth:`now` returns.
    """

    moment: datetime = DEFAULT_NOW

    def now(self) -> datetime:
        """Return the current instant."""
        return self.moment

    def monotonic(self) -> float:
        """Return a monotonic reading derived from :attr:`moment`."""
        return self.moment.timestamp()

    def advance(self, seconds: float) -> None:
        """Move time forward.

        Args:
            seconds: How far to advance. Negative values move time backwards,
                which is occasionally what a clock-skew test wants.
        """
        self.moment += timedelta(seconds=seconds)

advance

advance(seconds)

Move time forward.

Parameters:

Name	Type	Description	Default
seconds	float	How far to advance. Negative values move time backwards, which is occasionally what a clock-skew test wants.	required

Source code in jasil/testing.py

def advance(self, seconds: float) -> None:
    """Move time forward.

    Args:
        seconds: How far to advance. Negative values move time backwards,
            which is occasionally what a clock-skew test wants.
    """
    self.moment += timedelta(seconds=seconds)

monotonic

monotonic()

Return a monotonic reading derived from :attr:moment.

Source code in jasil/testing.py

def monotonic(self) -> float:
    """Return a monotonic reading derived from :attr:`moment`."""
    return self.moment.timestamp()

now

now()

Return the current instant.

Source code in jasil/testing.py

def now(self) -> datetime:
    """Return the current instant."""
    return self.moment

MemoryMail

MailProvider that appends each message to :attr:outbox.

Messages are validated and given a real Message-ID by the same builder the SMTP backend uses, so a test fails on the same invalid message production would reject.

Parameters:

Name	Type	Description	Default
clock	ClockProvider	Time source for the Date header.	required
from_address	str	Sender identity; a placeholder when empty.	''

Source code in jasil/backends/mail_memory.py

class MemoryMail:
    """``MailProvider`` that appends each message to :attr:`outbox`.

    Messages are validated and given a real ``Message-ID`` by the same builder
    the SMTP backend uses, so a test fails on the same invalid message
    production would reject.

    Args:
        clock: Time source for the ``Date`` header.
        from_address: Sender identity; a placeholder when empty.
    """

    def __init__(self, clock: ClockProvider, from_address: str = "") -> None:
        self._clock = clock
        self._from_address = from_address or _DEFAULT_SENDER
        self._lock = threading.Lock()
        self.outbox: list[MailMessage] = []
        #: The built MIME message for each ``outbox`` entry, same order, so a
        #: test can assert on the generated headers and ``Message-ID``.
        self.built: list[EmailMessage] = []

    def send(self, message: MailMessage) -> str:
        """Validate ``message``, record it, and return its ``Message-ID``.

        Raises:
            ValueError: When the message is invalid.
            MailError: When the durable job sending it has lost its lease, so a
                worker that no longer owns the job records nothing, as
                :class:`~jasil.backends.mail_smtp.SmtpMail` would send nothing.
        """
        if lease_lost():
            raise MailError("Job lease lost; send abandoned")
        mime = build_mime(message, sender=self._from_address, now=self._clock.now())
        # Recipients copied as sent: the caller's list may be reused or mutated
        # after ``send`` returns, which would rewrite what the test asserts on. A
        # shallow copy keeps the type, so ``outbox == [message]`` still holds.
        recorded = replace(message, to=copy.copy(message.to))
        with self._lock:
            self.outbox.append(recorded)
            self.built.append(mime)
        return str(mime["Message-ID"])

    def clear(self) -> None:
        """Forget every recorded message, e.g. between two phases of a test."""
        with self._lock:
            self.outbox.clear()
            self.built.clear()

clear

clear()

Forget every recorded message, e.g. between two phases of a test.

Source code in jasil/backends/mail_memory.py

def clear(self) -> None:
    """Forget every recorded message, e.g. between two phases of a test."""
    with self._lock:
        self.outbox.clear()
        self.built.clear()

send

send(message)

Validate message, record it, and return its Message-ID.

Raises:

Type	Description
ValueError	When the message is invalid.
MailError	When the durable job sending it has lost its lease, so a worker that no longer owns the job records nothing, as :class:~jasil.backends.mail_smtp.SmtpMail would send nothing.

Source code in jasil/backends/mail_memory.py

def send(self, message: MailMessage) -> str:
    """Validate ``message``, record it, and return its ``Message-ID``.

    Raises:
        ValueError: When the message is invalid.
        MailError: When the durable job sending it has lost its lease, so a
            worker that no longer owns the job records nothing, as
            :class:`~jasil.backends.mail_smtp.SmtpMail` would send nothing.
    """
    if lease_lost():
        raise MailError("Job lease lost; send abandoned")
    mime = build_mime(message, sender=self._from_address, now=self._clock.now())
    # Recipients copied as sent: the caller's list may be reused or mutated
    # after ``send`` returns, which would rewrite what the test asserts on. A
    # shallow copy keeps the type, so ``outbox == [message]`` still holds.
    recorded = replace(message, to=copy.copy(message.to))
    with self._lock:
        self.outbox.append(recorded)
        self.built.append(mime)
    return str(mime["Message-ID"])

install_test_platform

install_test_platform(
    data_dir, *, clock=None, settings=None
)

Build an all-in-process platform, publish it, and return it.

The local profile already resolves every capability to a process-local backend, so this needs no infrastructure. What it adds is the part that is easy to get wrong: rooting local storage inside the test's temporary directory instead of the working directory, substituting a controllable clock, and publishing the result — a platform that is built but never passed to set_active_platform makes every publish raise.

Parameters:

Name	Type	Description	Default
data_dir	str | Path	Directory the local storage backend writes under. Pass pytest's tmp_path so nothing survives the test.	required
clock	FixedClock | None	Time source to install. Defaults to a fresh :class:FixedClock.	None
settings	JasilSettings | None	Base configuration to build from. Defaults to an all-defaults :class:~jasil.settings.JasilSettings; data_dir is applied on top either way. It is also installed via jasil.settings.configure so code reading the settings directly agrees with the platform. When mail.uri is unset (or empty) it becomes memory://, so sent messages land in the outbox :func:memory_mail returns — unless the settings describe a multi-process deployment that enforces consistency, where memory:// would fail the build; mail then stays unset and sending raises.	None

Returns:

Type	Description
Platform	The published :class:~jasil.container.Platform, carrying clock.

Source code in jasil/testing.py

def install_test_platform(
    data_dir: str | Path,
    *,
    clock: FixedClock | None = None,
    settings: jasil_settings.JasilSettings | None = None,
) -> Platform:
    """Build an all-in-process platform, publish it, and return it.

    The ``local`` profile already resolves every capability to a process-local
    backend, so this needs no infrastructure. What it adds is the part that is
    easy to get wrong: rooting local storage inside the test's temporary
    directory instead of the working directory, substituting a controllable
    clock, and *publishing* the result — a platform that is built but never
    passed to ``set_active_platform`` makes every ``publish`` raise.

    Args:
        data_dir: Directory the local storage backend writes under. Pass pytest's
            ``tmp_path`` so nothing survives the test.
        clock: Time source to install. Defaults to a fresh :class:`FixedClock`.
        settings: Base configuration to build from. Defaults to an all-defaults
            :class:`~jasil.settings.JasilSettings`; ``data_dir`` is applied on top
            either way. It is also installed via ``jasil.settings.configure`` so
            code reading the settings directly agrees with the platform. When
            ``mail.uri`` is unset (or empty) it becomes ``memory://``, so sent
            messages land in the outbox :func:`memory_mail` returns — unless the
            settings describe a multi-process deployment that enforces
            consistency, where ``memory://`` would fail the build; mail then
            stays unset and sending raises.

    Returns:
        The published :class:`~jasil.container.Platform`, carrying ``clock``.
    """
    configured = replace(settings or jasil_settings.JasilSettings(), data_dir=str(data_dir))
    if not configured.mail.uri:
        # Unconfigured mail would raise on every send; a test wants it recorded.
        recording = replace(configured, mail=replace(configured.mail, uri="memory://", secure=True))
        refused = capabilities.check_mail_consistency(
            topology=capabilities.topology_of(recording),
            mail_uri=recording.mail.uri,
            mail_label="mail.uri",
        )
        if not refused or not recording.enforce_deployment_consistency:
            configured = recording
    jasil_settings.configure(configured)
    platform = build_platform(configured, clock=clock or FixedClock())
    platform_runtime.set_active_platform(platform)
    return platform

memory_mail

memory_mail(platform)

Return the recording mail backend of a test platform, typed.

Platform.mail is a :class:~jasil.providers.MailProvider, which has no outbox; this narrows it for a type checker and fails loudly when the platform was built with a real backend.

Parameters:

Name	Type	Description	Default
platform	Platform	The platform from :func:install_test_platform.	required

Returns:

Type	Description
MemoryMail	The MemoryMail whose outbox holds every message sent.

Raises:

Type	Description
TypeError	When platform.mail is not the in-memory backend.

Source code in jasil/testing.py

def memory_mail(platform: Platform) -> MemoryMail:
    """Return the recording mail backend of a test platform, typed.

    ``Platform.mail`` is a :class:`~jasil.providers.MailProvider`, which has no
    ``outbox``; this narrows it for a type checker and fails loudly when the
    platform was built with a real backend.

    Args:
        platform: The platform from :func:`install_test_platform`.

    Returns:
        The ``MemoryMail`` whose ``outbox`` holds every message sent.

    Raises:
        TypeError: When ``platform.mail`` is not the in-memory backend.
    """
    mail = platform.mail
    if not isinstance(mail, MemoryMail):
        raise TypeError("platform.mail is not the in-memory backend; configure mail.uri as memory://")
    return mail

reset_all

reset_all()

Clear every process-wide slot JASIL installs, except the ORM mapping.

Covers the settings, the correlation provider, the published platform, the durable subscriber registry, and the memoized Redis clients. Call it in fixture teardown; leaving any one of them set is how a passing test starts depending on the one before it.

The declarative base is deliberately left alone. JASIL's model modules capture it at import time, so clearing it would strand every model already imported — and a test module that imports one at module scope cannot re-import it. Map once for the whole session and leave it mapped. Call :func:jasil.orm.reset yourself only if you genuinely need to remap.

Does not close the platform: call :meth:jasil.container.Platform.close first if this process opened real connections.

Source code in jasil/testing.py

def reset_all() -> None:
    """Clear every process-wide slot JASIL installs, except the ORM mapping.

    Covers the settings, the correlation provider, the published platform, the
    durable subscriber registry, and the memoized Redis clients. Call it in
    fixture teardown; leaving any one of them set is how a passing test starts
    depending on the one before it.

    **The declarative base is deliberately left alone.** JASIL's model modules
    capture it at import time, so clearing it would strand every model already
    imported — and a test module that imports one at module scope cannot re-import
    it. Map once for the whole session and leave it mapped. Call
    :func:`jasil.orm.reset` yourself only if you genuinely need to remap.

    Does not close the platform: call :meth:`jasil.container.Platform.close`
    first if this process opened real connections.
    """
    jasil_settings.reset()
    correlation.reset()
    platform_runtime.reset()
    jobs_registry.registry.clear()
    redis_clients.reset_shared_clients()

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.adopt_existing_schema(engine)   # validate/record legacy tables, if any
migrations.upgrade(engine)                  # create/upgrade JASIL's tables
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.

SchemaCompatibilityError

Bases: RuntimeError

An unversioned JASIL schema cannot safely be adopted.

Source code in jasil/migrations/__init__.py

class SchemaCompatibilityError(RuntimeError):
    """An unversioned JASIL schema cannot safely be adopted."""

adopt_existing_schema

adopt_existing_schema(engine)

Validate and adopt complete, unversioned JASIL tables.

The tables are matched exactly against the head schema, then against each earlier revision's schema, newest first. The first exact match is recorded as the database's revision, so a following upgrade() applies whatever revisions come after it. If two revisions describe the same schema, the newest is recorded. An older revision's schema is derived from the migrations' own downgrade functions, never maintained separately.

Returns True only when this call records a revision. An empty database or one that already has a JASIL revision is left unchanged and returns False.

Raises:

Type	Description
SchemaCompatibilityError	If no revision matches exactly; the message names the closest revision and lists its differences.

Source code in jasil/migrations/__init__.py

def adopt_existing_schema(engine: "Engine") -> bool:
    """Validate and adopt complete, unversioned JASIL tables.

    The tables are matched exactly against the head schema, then against each
    earlier revision's schema, newest first. The first exact match is recorded
    as the database's revision, so a following ``upgrade()`` applies whatever
    revisions come after it. If two revisions describe the same schema, the
    newest is recorded. An older revision's schema is derived from the
    migrations' own ``downgrade`` functions, never maintained separately.

    Returns ``True`` only when this call records a revision. An empty database
    or one that already has a JASIL revision is left unchanged and returns
    ``False``.

    Raises:
        SchemaCompatibilityError: If no revision matches exactly; the message
            names the closest revision and lists its differences.
    """
    _require_alembic()
    from alembic import command
    from alembic.runtime.migration import MigrationContext
    from sqlalchemy import inspect

    from jasil.orm import jasil_table_names

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

        inspector = inspect(connection)
        if not set(inspector.get_table_names()) & jasil_table_names():
            return False

        closest: tuple[str, list[str], int] | None = None
        for revision, metadata in _revision_schemas():
            differences, weight = _schema_differences(connection, inspector, metadata)
            if not differences:
                command.stamp(_config(connection), revision)
                connection.commit()
                return True
            if closest is None or weight < closest[2]:
                closest = (revision, differences, weight)

        if closest is None:
            raise RuntimeError("JASIL has no packaged migration head")
        raise SchemaCompatibilityError(
            f"Cannot adopt an incompatible JASIL schema (closest revision {closest[0]}): " + "; ".join(closest[1])
        )

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

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.adopt_existing_schema(engine) for legacy unversioned tables, "
            "then jasil.migrations.upgrade(engine) at deploy time."
        )