ditliti-sdk 0.1.2

Official Python SDK for Ditliti (error tracking, tracing, replays, and profiling ingestion).

ditliti-sdk

Official Python SDK for Ditliti - error tracking, tracing, session replay, and profiling ingestion.

This SDK talks to a self-hosted Ditliti instance (ingestion-api). It does not connect to any managed/hosted service - point it at your own deployment's ingestion URL.

Install

pip install https://github.com/kangartha/inspectora/releases/download/sdk-v0.1.2/ditliti_sdk-0.1.2-py3-none-any.whl

Use pip install ditliti-sdk==0.1.2 only after PyPI confirms publication.

Quick start

from ditliti import DitlitiClient, install_global_exception_hook

client = DitlitiClient(
    endpoint="http://localhost:8305",
    api_key="mp_xxx",
    project_id="PROJECT_UUID",
    release="1.2.3",
    environment="production",
)

install_global_exception_hook(client)

try:
    risky_operation()
except Exception as exc:
    client.capture_exception(exc, context={"extra": "context"})

API

DitlitiClient(endpoint, api_key, project_id, release=None, environment=None, before_send=None, max_payload_bytes=200_000, max_retries=2, retry_base_seconds=0.15, queue_on_failure=False, max_queue_size=100, queue_file=None)

• client.set_user(user) / client.set_session(session_id)
• client.add_breadcrumb(...)
• client.capture_exception(exc, context=..., tags=...)
• client.capture_message(message, level=..., context=..., tags=...)
• client.capture_replay_segment(...)
• client.capture_profile(...)
• client.send_envelope(envelope) - low-level batch submit.
• client.flush_queue() - retry events queued after a failed send (when queue_on_failure=True).

Framework/runtime integrations

• install_global_exception_hook(client) - installs a global sys.excepthook.
• DitlitiLogHandler(client, level=logging.ERROR) - a logging.Handler that forwards log records as events.
• DitlitiASGIMiddleware(app, client) / install_fastapi(app, client) - FastAPI/Starlette/ASGI exception capture.
• DitlitiDjangoMiddleware(get_response, client) - Django exception capture middleware.

Distributed tracing

Generates real W3C traceparent (https://www.w3.org/TR/trace-context/) trace context, so a trace can be propagated across service/microservice boundaries instead of staying siloed per project:

# Service A: start a trace and call Service B, forwarding the header.
client.start_trace()
requests.get("https://service-b.internal/work", headers={"traceparent": client.get_traceparent()})

# Service B: continue the same trace from the inbound header.
# DitlitiASGIMiddleware / DitlitiDjangoMiddleware do this automatically, or call manually:
client.start_trace(request.headers.get("traceparent"))

with client.start_span("db.query", tags={"db": "postgres"}) as span:
    ...  # do the work; span.finish() is called automatically on exit

# capture_exception/capture_message automatically tag `trace_id` when a trace is active,
# so errors show up correlated to the trace at GET /api/v1/traces/:trace_id (org-wide).

• client.start_trace(incoming_traceparent=None) - starts a new trace, or continues one from an inbound traceparent header.
• client.get_traceparent() - the header value to forward on outgoing requests.
• client.start_span(operation_name, tags=...) → a Span usable as a context manager or via .finish(extra_tags=...).

Database query instrumentation (Database Insights)

ditliti.db wraps a query call with a db.query span carrying the OpenTelemetry-shaped attributes the backend uses for Database Insights (slow query / N+1 / error-rate / regression detection). The server parameterizes and hashes db.query.text itself - send the real SQL text, never a redacted one:

from ditliti.db import trace_query, instrument_dbapi_cursor, django_execute_wrapper

# Generic wrapper - works for any driver/ORM:
rows = trace_query(
    client,
    system="postgresql",
    text="SELECT * FROM orders WHERE id = %s",
    execute=lambda: cursor.execute("SELECT * FROM orders WHERE id = %s", (order_id,)) or cursor.fetchall(),
)

# Or wrap any PEP 249 (DB-API 2.0) cursor - sqlite3, psycopg2, pymysql, mysql-connector - transparently:
cursor = instrument_dbapi_cursor(client, connection.cursor(), system="postgresql")
cursor.execute("SELECT * FROM orders WHERE id = %s", (order_id,))
rows = cursor.fetchall()

# Django: register via the official execute_wrapper hook.
from django.db import connection
connection.execute_wrappers.append(django_execute_wrapper(client, system="postgresql"))

# SQLAlchemy: hooks the Engine's before/after/error cursor events (requires sqlalchemy installed).
from ditliti.db import instrument_sqlalchemy_engine
instrument_sqlalchemy_engine(client, engine, system="postgresql")

None of these ever set db.query.summary or ditliti.query_hash (server-computed). On failure they record error.type and re-raise the original exception unchanged; on success they record db.response.returned_rows when the driver reports a row/affected count.

Transport behavior

• Retries with exponential backoff on network failures and 5xx responses; no retry on permanent 4xx.
• before_send callback to redact/mutate or drop (return None) a payload before it's sent.
• Optional JSONL queue (queue_file=...) persists failed sends across process restarts.

More examples

See examples/basic.py in this package's source repository.

License

MIT - see LICENSE.