Example: observe an execution

When a call succeeds after a retry, the caller only sees the value. This example shows the two records RelPrim keeps of the same execution: the execution report returned with the result, and the structured events delivered while it runs.

The problem#

A ticket classifier sometimes needs a second attempt. Everything looks fine to the user, but if a growing share of calls needs retries, somebody should know before it turns into an outage. Reconstructing that from scattered log lines is slow and unreliable.

Implementation#

This version needs no extra packages: InMemoryEventSink collects events in memory. The demo classifier fails once.

import asyncio

from relprim import EventEmitter, InMemoryEventSink, RetryPolicy, async_operation


class TemporaryProviderError(Exception):
    pass


class DemoClassifier:
    def __init__(self) -> None:
        self.calls = 0

    async def classify(self, ticket: str) -> str:
        self.calls += 1
        if self.calls == 1:
            raise TemporaryProviderError("connection reset by peer")
        return "billing"


async def no_sleep(delay: float) -> None:
    """Skip the backoff delay so the example runs instantly."""


async def main() -> None:
    sink = InMemoryEventSink()
    classifier = DemoClassifier()

    result = await (
        async_operation("classify_ticket", classifier.classify)
        .with_events(EventEmitter(sinks=(sink,)))
        .with_retry(
            RetryPolicy(
                max_attempts=3,
                retry_on=(TemporaryProviderError,),
                async_sleeper=no_sleep,
            )
        )
        .run("I was charged twice")
    )

    # The report: what happened, available after the call returns.
    report = result.report
    print(f"value={result.value} status={report.status}")
    print(f"attempt_count={report.attempt_count} retry_count={report.retry_count}")
    if report.last_error is not None:
        print(f"last error: {report.last_error.type}: {report.last_error.message}")

    # The events: the same execution, delivered to sinks as it happens.
    for event in await sink.events():
        payload = event.payload
        if event.event_type.value == "attempt.failed":
            print(f"{event.event_type}  error_type={payload['error_type']}")
        elif event.event_type.value == "retry.scheduled":
            print(f"{event.event_type}  delay_source={payload['delay_source']}")
        else:
            print(event.event_type)


asyncio.run(main())

It prints:

value=billing status=succeeded
attempt_count=2 retry_count=1
last error: TemporaryProviderError: connection reset by peer
operation.started
attempt.started
attempt.failed  error_type=TemporaryProviderError
retry.scheduled  delay_source=backoff
attempt.started
attempt.succeeded
operation.succeeded

What happens#

The report answers “what happened to this call?” after it returns: status, attempt_count, retry_count, each attempt with its status, duration and error, and last_error. It is available on every OperationResult, and on OperationExecutionError.report when the operation fails.

The events describe the same execution as it happens, in order: operation.started, then attempt.started and attempt.failed, retry.scheduled (with delay_source backoff), the second attempt, and operation.succeeded. Each event has an event_type, the operation_name, a timestamp and a payload of JSON scalars. Event types, payload keys and report metadata keys are a stable contract in 1.x.

Send the same events elsewhere#

The emitter accepts any sinks, so changing where events go does not change the operation. With the relprim[otel] extra, OpenTelemetryEventSink adds each event to the active span:

from relprim import EventEmitter
from relprim.opentelemetry import OpenTelemetryEventSink

events = EventEmitter(sinks=(OpenTelemetryEventSink(),))

Your application still configures OpenTelemetry (tracer provider, exporter, spans); RelPrim only adds events to the span that is current. See OpenTelemetry.

SQLiteEventStore keeps a durable local history using only the standard library:

from relprim import EventEmitter, SQLiteEventStore

store = SQLiteEventStore("relprim-events.db")
await store.initialize()
events = EventEmitter(sinks=(store,))

# Later: the stored history of one operation, oldest first.
history = await store.history(operation_name="classify_ticket", limit=100)

It is meant for local and single-host persistence, not as a shared event database. See SQLite event store.

Why this matters#

The report gives each caller a precise answer without parsing logs. The events give operators a continuous stream with the same vocabulary, which can be counted (“how many calls needed a retry?”), traced or stored. Both come from the execution itself, so they agree.

Production note#

Sinks run inline with the operation, so keep emit fast. Payloads carry exception types rather than exception messages, and idempotency keys never appear in them. A validation.failed event does include the validator’s reason, which can describe the rejected value, so review what your validators and operation names reveal before exporting events.

Related documentation#

• Structured events: every event type and payload key.
• Core concepts: operations, attempts and reports.
• OpenTelemetry and SQLite event store.
• Runnable repository examples: structured_events.py, opentelemetry_integration.py and sqlite_event_store.py.