Splatty (Python)

Python client for Splatty. Captures exceptions and logs and ships them over the envelope protocol. Mirrors splatty-ruby and splatty-js.

Standard library only — no runtime dependencies. Every integration is duck-typed against what it adapts, so installing Splatty never pulls in Celery or a web framework; you only wire up the ones you already use.

• Installation
• Quick start
• Configuration
• Capturing events
• WSGI
• Uncaught exceptions
• Background jobs
• Logs
• Shutting down
• API reference
• Wire protocol

Installation

pip install splatty

Requires Python 3.10 or newer.

Quick start

import os
import splatty

splatty.init(
    url=os.environ.get("SPLATTY_URL", "https://splatty.app"),
    dsn=os.environ.get("SPLATTY_DSN"),
    environment=os.environ.get("SPLATTY_ENVIRONMENT", "development"),
    release=os.environ.get("SPLATTY_RELEASE"),
)

try:
    do_something()
except Exception as e:
    splatty.capture_exception(e)

init() validates the config, creates the client and installs whichever integrations the config asks for. Call it once, as early in boot as you can. init() never raises on bad configuration: a missing DSN or invalid URL logs one warning and turns every capture into a no-op, so misconfiguring Splatty can't stop the host app from booting.

Configuration

Every option is a keyword argument to init(). Unset options fall back to the environment variables noted below.

option	default	notes
url	SPLATTY_URL or https://splatty.app	the Splatty server
dsn	SPLATTY_DSN	the project's DSN key — a bare key, not a URL
environment	SPLATTY_ENVIRONMENT or development
release	SPLATTY_RELEASE	announced on boot when set
enabled	True	False turns every capture into a no-op
logs	True	install the batching log handler on the root logger
capture_unhandled	False	wrap sys.excepthook / threading.excepthook
send_default_pii	False	send request headers verbatim
context_lines	5	source lines either side of a stack frame; 0 disables
server_name	hostname
open_timeout	5.0	connect timeout, seconds
read_timeout	10.0
logger	logging.getLogger("splatty")	where SDK warnings go
before_send	None	edit or drop events; return None to drop
log_options	None	dict of LogHandler kwargs (batch_size, flush_interval, queue_limit, host, level)

By default (send_default_pii=False) sensitive request headers — Cookie, Authorization, CSRF tokens, API keys and similar — are replaced with [Filtered] before an event leaves the process. Set send_default_pii=True only if you understand that cookies and auth tokens will then be transmitted and stored.

When release is set, init() announces it in a background thread so the deploy shows up in Splatty without waiting for the app to log or raise something first. Every process announces; the server keeps one deployment per release and environment.

Capturing events

splatty.capture_exception(e)
splatty.capture_message("payment reconciliation drifted", level="warning")

Both accept scope keyword arguments: tags, extra, contexts, transaction, request and level. Both return the event id, or None when the event was dropped or the SDK is disabled.

An exception object is only reported once — capturing the same exception a second time (say, once in an except block and again by the WSGI middleware as it re-raises) produces a single event. Chained exceptions (raise ... from e, and implicit chains) are unwrapped and sent as one event, oldest cause first.

Each frame carries context_lines lines of source either side of the line that raised, read from disk at capture time and capped by an in-process cache.

WSGI

from splatty.wsgi import CaptureExceptions

app = CaptureExceptions(app)

Exceptions escaping the wrapped app are captured with the request context (URL, method, headers — scrubbed unless send_default_pii) attached and then re-raised, so the server's own error handling still runs. An X-Request-Id header becomes a request_id tag.

Works with anything that speaks WSGI: Flask (app.wsgi_app = CaptureExceptions(app.wsgi_app)), Django (wrap application in wsgi.py), gunicorn, uWSGI.

Uncaught exceptions

splatty.init(dsn=..., capture_unhandled=True)

Wraps sys.excepthook (reported as fatal) and threading.excepthook (reported as error), ships queued logs, and then calls the previous hook — crash output and exit semantics are unchanged.

Background jobs

When Celery is already imported by the host app, init() connects to the task_failure signal, so failed tasks are reported with job_backend, job_class and job_queue tags, the task id, retry count and truncated arguments as extra data, and the task name as the transaction. Splatty never imports Celery on its own — mirroring how the Ruby SDK only hooks the job backends it finds loaded.

For any other job runner, capture in its failure hook yourself:

splatty.capture_exception(e, tags={"job_backend": "rq", "job_class": job.func_name},
                          transaction=job.func_name)

Capture-once applies here too: a job failure that surfaces through two paths produces a single event.

Logs

With logs=True (the default), init() installs a logging.Handler on the root logger, so anything that logs through the stdlib logging module is buffered and shipped to Splatty in batches — flushed every 15 seconds, when 100 entries accumulate, and on close(). The queue is bounded (5,000 entries, oldest dropped first) and shipping happens on a daemon thread, so logging never blocks and never holds the process open.

Entries keep their structured extra fields; request_id, method, path, status, duration_ms, controller and action are promoted to first-class columns. Records logged by Splatty itself and request logs for Splatty's own intake endpoints are dropped to avoid feedback loops when dogfooding.

Note the stdlib's usual rule still applies: the root logger only forwards records that pass its level, so configure logging.basicConfig(level=...) (or your framework's logging setup) as you normally would. Disable shipping entirely with logs=False, or tune with log_options={"batch_size": ..., "flush_interval": ...}.

Shutting down

splatty.flush()  # ship queued logs, keep running
splatty.close()  # wait for the release announcement, flush, uninstall, drop the client

close() drains the final log batch over the wire, so call it (or flush()) before a worker process exits.

API reference

Lifecycle — init(**options) -> Client, flush(), close().

Accessors — client(), configuration(), enabled(), log_handler().

Capture — capture_exception(e, **scope), capture_message(message, level="info", **scope).

Integrations — splatty.wsgi.CaptureExceptions, splatty.excepthook (install/uninstall), splatty.celery (install/uninstall), splatty.LogHandler.

Building blocks, if you're assembling your own pipeline — Client, Configuration, Transport, Scrubber, LineCache, splatty.event.build_exception_event, splatty.event.build_message_event, encode_args, map_level.

Constants — VERSION, SDK_NAME, DEFAULT_URL, FILTERED, SENSITIVE_HEADER_PATTERN, INTAKE_PATH_PATTERN, MAX_ARGS_LENGTH, DEFAULT_BATCH_SIZE, DEFAULT_FLUSH_INTERVAL, DEFAULT_QUEUE_LIMIT.

Wire protocol

Everything is POSTed gzipped to <url>/api/envelope over a keep-alive connection, with Content-Type: application/x-splatty-envelope and Authorization: Bearer <dsn>. The body is three newline-separated lines: an envelope header, an item header, and the JSON payload.

{"event_id":"…","sent_at":"…","dsn":"…","sdk":{"name":"splatty.python","version":"0.1.0"}}
{"type":"event","content_type":"application/json","length":1234}
{"event_id":"…","timestamp":"…","platform":"python","level":"error","exception":{…}}

Log batches use the same shape. Their envelope header carries no event_id, and the item header is {"type":"log","item_count":N,"content_type": "application/vnd.splatty.items.log+json","length":…} over a {"host":…,"items":[…]} payload.

Transport failures never raise and never bubble into your code — they're warned about through config.logger and the send returns None.

Development

pip install -e . pytest
pytest

License

MIT.