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 to drain_timeout for the final HTTP call. Raises DrainTimeoutError if 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.