Python SDK · Recorder

With the recorder constructed and the middleware wired, your handlers can record events. The examples below are lifted from the secrets-vault examples repo.

Simple write

The most common case. One event per request. Set fields on current_event() and the middleware records on response finish. Handlers that don't set action are skipped entirely (no audit noise):

from everscribe.asgi import current_event
from everscribe.event import Target

@app.post("/api/secrets/{secret_id}/reveal")
async def reveal(secret_id: str, request: Request):
    me = actor(request)
    if me is None:
        return PlainTextResponse("unknown actor", status_code=401)

    e = current_event()
    e.tenant_id = me.tenant_id
    e.target = Target(type="secret", id=secret_id)

    s = vault.get(secret_id)
    if s is None or s.tenant_id != me.tenant_id:
        return PlainTextResponse("not found", status_code=404)

    e.action = "secret.reveal"
    e.with_field("name", s.name)

    return JSONResponse({"value": s.value})

The middleware auto-fills result from the response status, so error paths and success paths both record correctly without you setting result by hand.

Mutation with before/after diff

For events that change a resource, populate the change field so the audit trail shows what changed, not just that something did:

@app.post("/api/secrets/{secret_id}/rotate")
async def rotate(secret_id: str, request: Request):
    me = actor(request)
    e = current_event()
    e.tenant_id = me.tenant_id
    e.target = Target(type="secret", id=secret_id)

    body = await request.json()
    s = vault.get(secret_id)
    if s is None:
        return PlainTextResponse("not found", status_code=404)

    before = snapshot(s)          # pre-mutation state
    s.value = body["value"]
    s.rotated_at = now_iso()
    after = snapshot(s)           # post-mutation state

    e.action = "secret.rotate"
    e.diff(before, after)         # populates e.change with the diff

    return Response(status_code=204)

snapshot is your function. Typically a dataclass that mirrors the resource. The diff helper computes the JSON patch between the two snapshots and stores it in event.change.

Redacted fields

e.diff accepts a with_redacted_fields(...) option that replaces sensitive paths with "[REDACTED]" before the diff is stored. Paths are JSON pointers:

from everscribe.event import with_redacted_fields

e.diff(before, after,
    with_redacted_fields("/password_hash", "/billing/credit_card"))

Two approaches, pick whichever fits the resource shape:

• Redact paths at diff time with with_redacted_fields. Shorter when you have a small known set of paths to strip from an otherwise-safe object.
• Pre-redact in your snapshot function. Safer for resources where the sensitive field is large, structured, or easy to forget. The vault examples take this approach: snapshot() always returns value="[REDACTED]" so the plaintext never even reaches the SDK.

Batch recording

The buffered recorder batches events under the hood. record() enqueues; a background loop sends batches every flush_interval or when flush_size is reached, whichever comes first. You don't need to call any batch API.

For high-throughput services, tune the batch behavior via constructor options:

rec = es.new_recorder(
    buffer_size=5000,
    flush_size=500,
    flush_interval=2.0,   # seconds
)

For graceful shutdown, drain the buffer so pending events make it to the API before the process exits:

rec.close()  # flushes pending events, then disposes

In FastAPI, call it from a lifespan shutdown hook.

Multiple events per request

Some handlers naturally produce more than one event. Sharing a secret with three recipients, fan-out notifications, bulk imports. Call new_from_context() once per event you want to record. Each call returns an independent clone of the per-request template (actor, origin pre-filled); record each clone explicitly. The recorder batches them together on the next flush.

from everscribe.event import new_from_context, Target

@app.post("/api/secrets/{secret_id}/share")
async def share(secret_id: str, request: Request):
    me = actor(request)
    body = await request.json()
    user_ids = body.get("user_ids", [])

    vault.share(me.tenant_id, secret_id, user_ids)

    # One audit event per recipient. Each new_from_context() call yields a
    # fresh clone with actor + origin already filled in.
    for uid in user_ids:
        e = new_from_context()
        e.tenant_id = me.tenant_id
        e.action = "secret.share"
        e.target = Target(type="secret", id=secret_id)
        e.with_fields("shared_with", uid)
        rec.record(e)

    return Response(status_code=204)

Direct recording

Not every event comes from an HTTP request. Cron jobs, queue workers, CLI scripts, startup hooks. Anywhere the ASGI middleware isn't in the path, you build each event yourself: set actor to a service identity and set result by hand since there's no response status to derive it from.

from everscribe.event import Event, Actor, Target, Result

# rotate_expiring_secrets is a scheduled job that rotates any secret due
# for rotation. Runs out of band — no middleware, no request, just the
# recorder you constructed at boot.
def rotate_expiring_secrets(rec, vault):
    for s in vault.due_for_rotation():
        e = Event("secret.rotate")
        e.actor = Actor(type="service", id="rotation-worker")
        e.target = Target(type="secret", id=s.id)
        e.result = Result(status="ok")
        try:
            vault.rotate(s.id, generate())
        except Exception as err:
            e.result = Result(status="error", message=err)  # exception -> str(err)
        rec.record(e)

The same recorder instance backs both your request handlers and your background jobs. Events from both paths land in the same project and batch together. Construct one recorder at boot, share it across both surfaces, and close() it once on shutdown.