Python SDK · Middleware and interceptors

An adapter installs a per-request Event (reachable with current_event()), auto-populates actor (via your resolver) and origin (from headers), and records the event after the handler completes. Handlers just set action / target on it and the adapter does the rest. The ASGI adapter covers FastAPI, Starlette, and any ASGI app; Flask, Django, and gRPC have their own.

All adapters share the same record lifecycle in everscribe.event; each supplies only its own framework-specific transport bindings.

Actor Resolver

The middleware needs a resolve_actor: a function that derives the Actor for a request. It receives the Starlette Request, so identity can come straight off it. A session cookie, a JWT, a header, an OAuth token. Whatever your auth stack uses.

from starlette.requests import Request

from everscribe.event import Actor

# resolve_actor reads identity off the request. Real apps read a session
# cookie or JWT here; this demo reads a header.
def resolve_actor(request: Request) -> Actor:
    user = users.get(request.headers.get("x-user-id", ""))
    if user is None:
        return Actor(type="anonymous")
    return Actor(
        type="user",
        id=user.id,
        display_name=user.name,
        email=user.email,
    )

ActorResolver type:

ActorResolver = Callable[[Request], Actor]

Wiring it up

Add the middleware to your app. It installs current_event() and auto-records on response finish:

from everscribe.asgi import EverscribeMiddleware

app.add_middleware(EverscribeMiddleware, recorder=rec, resolve_actor=resolve_actor)

Ordering matters. If your resolver reads a session, that session middleware must run before the audit middleware (i.e. be the outer layer). Starlette runs the last-added middleware first, so add your session middleware after EverscribeMiddleware. Get it backwards and the resolver sees no session and every event records as anonymous.

Other frameworks

Flask and Django get their own dedicated adapters rather than a generic WSGI wrapper, because each has a hook point a WSGI-level middleware can't reach.

Flask

from everscribe.flask import EverscribeFlask

# Direct construction:
ext = EverscribeFlask(app, recorder=rec, resolve_actor=resolve_actor)

# Or app-factory style:
ext = EverscribeFlask()
ext.init_app(app, recorder=rec, resolve_actor=resolve_actor)

EverscribeFlask hooks before_request and teardown_request, not app.wsgi_app. before_request runs inside the Flask application context, so a resolver here can read flask.g, flask.session, or flask_login.current_user directly - none of which a WSGI-level wrapper outside the app context could see. teardown_request always runs, even when a view raises and the exception propagates unhandled (the default under app.testing/app.debug), so a raising view still gets recorded. This is why Flask gets a dedicated adapter instead of reusing the ASGI one or a bare WSGI shim.

Requires Flask: pip install "everscribe[flask]".

Django

MIDDLEWARE = [
    ...,
    "django.contrib.auth.middleware.AuthenticationMiddleware",
    "everscribe.django.EverscribeMiddleware",  # after auth
]

EverscribeMiddleware is a standard Django middleware (Middleware(get_response)), mounted through MIDDLEWARE like any other. Django itself has already parsed request.user/request.session and converted view exceptions into an HttpResponse by the time this middleware's own get_response(request) call returns, so resolve_actor typically reads identity straight off request.user. Since Django instantiates middleware with no seam for extra constructor arguments, recorder/resolve_actor/logger can also be set via EVERSCRIBE_RECORDER/EVERSCRIBE_RESOLVE_ACTOR/EVERSCRIBE_LOGGER in Django settings instead of (or alongside) constructor keywords.

Requires Django: pip install "everscribe[django]".

Reaching the event

Reach the event inside a handler with current_event() (from everscribe.event). Some adapters also expose it on the request:

@app.post("/login")
async def login():
    current_event().action = "user.login"  # or request.state.everscribe_event
    return {"ok": True}

@app.post("/login")
def login():
    current_event().action = "user.login"
    return "ok"

def login(request):
    current_event().action = "user.login"  # or event_from_request(request)
    return HttpResponse("ok")

class Service(pb_grpc.ServiceServicer):
    def Login(self, request, context):  # sync and async
        current_event().action = "user.login"
        return pb.LoginReply()

gRPC

Two interceptors, because grpcio's sync and async servers are not interchangeable: EverscribeServerInterceptor extends grpc.ServerInterceptor for a grpc.server(); EverscribeAsyncServerInterceptor extends grpc.aio.ServerInterceptor for a grpc.aio.server(). Both live in everscribe.grpc.

import grpc
from everscribe.grpc import EverscribeServerInterceptor

server = grpc.server(
    futures.ThreadPoolExecutor(),
    interceptors=[EverscribeServerInterceptor(recorder=rec, resolve_actor=resolve_actor)],
)

from everscribe.grpc import EverscribeAsyncServerInterceptor

server = grpc.aio.server(
    interceptors=[EverscribeAsyncServerInterceptor(recorder=rec, resolve_actor=resolve_actor)],
)

All four RPC shapes are supported: unary-unary, unary-stream (server streaming), stream-unary (client streaming), and stream-stream (bidi streaming). This is broader than the sdk-go/sdk-rust/sdk-node gRPC adapters (unary, plus server-streaming for Go and Rust), because grpcio's intercept_service replaces the handler's callable outright rather than hooking a single fixed lifecycle point, so every arity gets wrapped the same way.

Every RPC is recorded by default: action defaults to the full method name (for example /billing.v1.Billing/RefundInvoice), so an RPC records unless the handler clears it. The HTTP adapters record nothing until a handler names the event.

Result.code carries the HTTP equivalent of the gRPC status, not the native gRPC code - OK maps to 200, NOT_FOUND to 404, and so on, matching the mapping sdk-go's and sdk-node's gRPC adapters use.

Requires grpcio: pip install "everscribe[grpc]".