Python SDK · BufferedRecorder
The default recorder buffers events in memory and flushes in batches on a background thread. Because record() only enqueues, it's safe to call from async def handlers without blocking the event loop:
rec = es.new_recorder(buffer_size=2000, flush_interval=2.0)
# or
rec = recorder.new(project_id, api_key, buffer_size=2000)
Options
| Option | Default | Purpose |
|---|---|---|
buffer_size |
1000 |
Capacity of the in-memory event buffer. |
flush_size |
100 |
Event count that triggers an immediate flush. |
flush_interval |
5.0 |
Max seconds between flushes when the size threshold isn't reached. |
flush_timeout |
30.0 |
Per-flush timeout (seconds) for batch HTTP calls. |
overflow |
"drop-newest" |
Behavior when the buffer is full: "drop-newest", "block", or "error". |
drain_timeout |
30.0 |
Max seconds close() waits for in-flight events. |
request_timeout |
10.0 |
Per-request HTTP timeout (seconds). |
logger |
module logger | A logging.Logger for diagnostics (overflow warnings, flush errors). |
base_url |
production | Override the ingestion endpoint (tests, staging). |
transport |
urllib |
Custom transport callable (primarily for tests). |
auto_idempotency_key |
off | Copy event.id into event.idempotency_key when the latter is empty. |
Times are in seconds (Python convention), where the Node SDK uses milliseconds. overflow accepts either the OverflowPolicy enum or the equivalent string.
Overflow policies
"drop-newest" # drop the incoming event, log a warning (default)
"block" # wait for space (until the buffer drains or the recorder closes)
"error" # raise BufferFullError
Close & flush
close()flushes pending events and waits up todrain_timeoutfor the final HTTP call. RaisesDrainTimeoutErrorif the deadline is hit. Call it exactly once, on the path that actually runs at shutdown.flush(timeout=None)drains buffered events at call time. Useful in tests and for graceful shutdown that needs to observe drain success.stats()returns counters (dropped,flushed,flush_errs,pending,buffer_size) for observability.
Where close goes
The recorder is constructed once at import time and lives as long as the process, so nothing in request handling ever closes it. Hook the framework's shutdown. For FastAPI or Starlette that is the lifespan context manager:
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
yield # the app serves for the duration of the yield
rec.close() # runs, and drains
app = FastAPI(lifespan=lifespan)
close() is synchronous: it joins the background flush thread and blocks for up to drain_timeout. That is fine at shutdown, when nothing else is being served. Use await asyncio.to_thread(rec.close) if you would rather keep the event loop free while it drains.
Flask, Django, and any other WSGI app use their process manager's equivalent hook (Gunicorn's worker_exit, for instance). What matters is that the hook runs when the worker is sent SIGTERM, not only on a clean interpreter exit.