obs-app-sdk

Python client for the easy-obs.com ingestion API - send logs, metrics and traces from your application.

Install

pip install ct_obs_app

(the PyPI project is named ct_obs_app; the importable module is obs_app_sdk - see Quickstart below)

Get an API key

Log in at the obs_app dashboard and copy your API key from the profile page (rotate it there too, if needed). The SDK authenticates every request with it.

Quickstart

from datetime import datetime, timezone

from obs_app_sdk import Client, CaptureTrace

client = Client(api_key="your-api-key")

# Logging
client.log("Something happened", level=4)  # 1=Critical .. 5=Debug, default 4=Info
client.log(
    "Something happened",
    datetime=datetime.now(timezone.utc),  # defaults to time of receipt if omitted
    app="my-app", domain="example.com", container_id="ctr-1", logger_name="myapp.worker",
    extra={"user_id": 42},
)
client.log_batch([{"message": "a"}, {"message": "b"}])

# Metrics
client.add_metric(value=1.0, metric="request_duration_ms", auto_create=True)
client.add_metrics([
    {"metric": "request_duration_ms", "value": 12.3},
    {"metric": "queue_depth", "value": 4},
], auto_create=True)

# Traces - nested context managers build one tree, sent as a single request when the
# outermost one exits
with CaptureTrace(client, "handle_request", app="my-app"):  # app is optional; the inner spans take it from here
    with CaptureTrace(client, "query_db"):
        ...
    with CaptureTrace(client, "render_response"):
        ...

# Or submit logs/metrics/traces together in one request (validated all-or-nothing)
client.ingest(
    logs={"logs": [{"message": "started"}]},
    metrics={"metrics": [{"metric": "request_duration_ms", "value": 12.3}]},
    traces=[{"start": "...", "end": "...", "text": "handled"}],
)

Stdlib logging integration

import logging
from obs_app_sdk import Client, ObsAppLogHandler

client = Client(api_key="your-api-key")
logging.getLogger().addHandler(ObsAppLogHandler(client, app="my-app", domain="example.com"))

Each stdlib log record becomes one POST /api/log/ call. logger_name and datetime are filled in automatically from the record; app/domain/container_id are fixed tags you set once on the handler. For high-volume logging, call client.log_batch(...) directly instead of relying on the handler.

Reporting view errors

from obs_app_sdk import Client, capture_errors

client = Client(api_key="your-api-key")

@capture_errors(client, app="my-app", domain="example.com")
def my_view(request, user_id):
    ...

Wrap a view (or any function) with capture_errors. If it raises, the decorator sends an ERROR-level log (level=2) to the API with the exception message, the full traceback, and repr() of the call's args/kwargs under extra, then re-raises - your framework's normal error handling (e.g. a 500 response) still runs unchanged; this only observes. A failure while reporting (network error, bad API key, etc.) is swallowed so it can never replace or mask the original exception.

repr() of every argument is sent as-is (truncated past 2000 characters) - be mindful of passing secrets/PII as view arguments if that's a concern for your application.

Background (fire-and-forget) mode

client = Client(api_key="your-api-key", background=True, on_error=lambda exc: print("send failed:", exc))
client.log("Something happened")  # returns None immediately - queued, not yet sent

Pass background=True and every call is queued and sent from a single background thread instead of blocking the caller. Because the call has already returned by the time the request actually goes out, errors (auth, validation, rate limit, network) can't be raised back to the caller - they're only passed to on_error, if you provide one, otherwise dropped. Client.log returns None in this mode instead of the log id, since it isn't known yet. Only use this for data you can afford to lose occasionally (logs, metrics) - not for anything you need to confirm succeeded.

The queue is unbounded and the worker thread is a daemon, so anything queued right before the process exits can be lost. flush_on_exit=True (the default) registers an atexit hook that gives the queue a few seconds to drain; call client.flush(timeout=...) directly, or use Client as a context manager, for more control.

Error handling

All methods raise on non-2xx responses:

• ObsAppApiKeyError - missing/invalid API key (HTTP 401)
• ObsAppValidationError - invalid request data (HTTP 400); .errors holds the server's per-field error dict when there is one, else None
• ObsAppRateLimitedError - HTTP 429. The API blocks the offending IP for 1 hour once its rolling rate-limit window trips - do not retry in a loop on this error, it won't help
• ObsAppError - base class; also raised for network failures and unexpected status codes

Notes

• This client is synchronous and makes one HTTP request per call - there is no background buffering/auto-flush thread. Use the batch methods (log_batch, add_metrics, ingest) to reduce request count for high-volume callers.
• add_metrics (batch) only supports metric names, not metric_id - use add_metric (singular) for id-based lookups.