Peeng Python SDK — Batched Log Delivery for Python 3.8+

The official Peeng Python SDK (pip install peeng) is a thread-safe client that delivers logs to Peeng in the background via a daemon thread. Every log call enqueues the entry and returns immediately — the SDK handles batching, retries, and idempotency automatically — so logging never blocks your application or crashes it when Peeng is unavailable.

MIT licensed. Requires Python 3.8+. The only runtime dependency is requests.

Install

pip install peeng

pip3 install peeng

Quick start

Construct a client, call any logging method, and call close() when you’re done:

from peeng import PeengClient

client = PeengClient(api_key="pk_test_xxx", environment="production")

client.info("user signed up", status_code=200, service="auth-api")
client.error(
    "payment failed",
    status_code=500,
    service="billing-api",
    metadata={"order_id": "ord_123"},
)

client.close()  # flush remaining logs and stop the background thread

If you never call close() explicitly, it still runs automatically at process exit via atexit — but calling it yourself gives you a deterministic shutdown point.

Why use the SDK

Non-blocking

Log calls enqueue onto an in-memory buffer and return immediately. A background daemon thread flushes it — every flush_interval seconds, or as soon as the buffer hits max_batch_size.

Automatic retries

Network errors, 429s, and 5xxs are retried with exponential backoff and jitter, reusing one Idempotency-Key per batch so retries can’t double-insert.

Never crashes your app

Delivery failures are reported through on_error (or written to stderr) instead of raising out of a logging call.

Thread-safe

Safe to call from multiple threads simultaneously — the internal buffer is protected by a lock.

Constructor parameters

Pass keyword arguments to PeengClient(...) to configure the client.

Parameter	Type	Default	Description
`api_key`	`str`	required	Your Peeng project API key.
`base_url`	`str`	`http://localhost:3000`	Peeng API base URL — set to `https://api.peeng.dev` for real traffic.
`environment`	`str`	`None`	Default `environment` for calls that don’t override it.
`service`	`str`	`None`	Default `service` for calls that don’t override it.
`max_batch_size`	`int`	`100`	Flush as soon as the buffer hits this size. Hard-capped at `500`.
`flush_interval`	`float`	`2.0`	Seconds between background flushes.
`max_retries`	`int`	`3`	Retry attempts before giving up on a batch.
`on_error`	`Callable[[Exception], None]`	writes to stderr	Called with delivery failures instead of raising.
`request_timeout`	`float`	`10.0`	Per-request timeout in seconds.

Logging methods

The SDK exposes one convenience method per log level — debug, info, warn, error, and fatal — plus a generic log() form for dynamic level selection:

from peeng import PeengClient, LogLevel

client = PeengClient(api_key="pk_test_xxx", environment="production")

client.debug("cache miss", status_code=200)
client.info("request completed", status_code=200, request_id="req_abc")
client.warn("slow query", status_code=200, metadata={"duration_ms": 1400})
client.error("unhandled exception", status_code=500, stack_trace="Traceback ...")
client.fatal("out of memory", status_code=500, hostname="worker-3")

# equivalent generic form
client.log(LogLevel.INFO, "request completed", status_code=200)

status_code is required on every call (100–599). The following keyword arguments are optional: service, metadata (any JSON-serializable dict), stack_trace, hostname, request_id, user_id, and timestamp (a datetime object or an ISO-8601 string). See the field reference for what each one means.

Context manager

Use PeengClient as a context manager to have close() called automatically when the block exits:

from peeng import PeengClient

with PeengClient(api_key="pk_test_xxx", environment="production") as client:
    client.info("service started", status_code=200)
    # ... application code ...
# client.close() called automatically on exit

Graceful shutdown

Call flush() to block until currently buffered logs are sent, and close() to stop the background thread and flush any remaining logs:

client.flush()             # block until currently buffered logs are sent
client.flush(timeout=2.0)  # give it up to 2 seconds, then return

client.close()             # stop background thread and flush remaining logs

Call close() (or use the context manager) before your process exits for a deterministic delivery guarantee — the atexit hook is a safety net, not a replacement for explicit shutdown in long-running services.

Custom batching configuration

Override the default batching and retry behaviour at construction time:

client = PeengClient(
    api_key="pk_test_xxx",
    base_url="https://api.peeng.dev",
    environment="production",
    service="checkout-api",
    max_batch_size=250,
    flush_interval=5.0,
    max_retries=5,
)

Error handling

Supply an on_error callback to receive delivery failures without letting them surface as exceptions in your application code:

def handle_delivery_failure(exc: Exception) -> None:
    print(f"failed to ship logs to Peeng: {exc}")

client = PeengClient(
    api_key="pk_test_xxx",
    environment="production",
    on_error=handle_delivery_failure,
)

Your on_error handler receives one of two exception types:

peeng.PeengApiError — the server responded with an HTTP error envelope. Carries status_code, message, and error (from the server envelope). Raised for non-retryable errors (e.g. 400) or after retries are exhausted (e.g. repeated 5xx).
requests exception — a network-level failure with no HTTP response at all (DNS failure, connection refused), surfaced after retries are exhausted.

Retry behaviour: 429, 5xx, and network/connection errors are retried up to max_retries times with exponential backoff and jitter, reusing the same Idempotency-Key on every retry of a given batch. Any other 4xx (e.g. a 400 validation error) is non-retryable and reported to on_error immediately. See Errors for the full retryable/non-retryable rules.

Get Started

Essentials

SDKs

Peeng Python SDK — Batched Log Delivery for Python 3.8+

Install

Quick start

Why use the SDK

Non-blocking

Automatic retries

Never crashes your app

Thread-safe

Constructor parameters

Logging methods

Context manager

Graceful shutdown

Custom batching configuration

Error handling

​Install

​Quick start

​Why use the SDK

Non-blocking

Automatic retries

Never crashes your app

Thread-safe

​Constructor parameters

​Logging methods

​Context manager

​Graceful shutdown

​Custom batching configuration

​Error handling

Install

Quick start

Why use the SDK

Constructor parameters

Logging methods

Context manager

Graceful shutdown

Custom batching configuration

Error handling