Logthread for Python

Send your Python logs to Logthread. Standard library logging in, searchable, live-tailable logs out. No runtime dependencies (the optional logthread[otel] extra adds traces and metrics), Python 3.9 or newer.

Install#

pip install https://downloads.logthread.com/sdk/python/0.2.1/logthread-0.2.1-py3-none-any.whl

The package is hosted by Logthread, not on PyPI yet. Install it from its exact URL (in requirements.txt: logthread @ https://downloads.logthread.com/sdk/python/0.2.1/logthread-0.2.1-py3-none-any.whl), never with --extra-index-url: pip would then also ask PyPI for logthread and take whichever version is highest. The current version, size and SHA-256 are on the download page.

Quick start#

export LOGTHREAD_ENDPOINT=https://ingest.example.com   # shown in Logthread onboarding
export LOGTHREAD_INGEST_KEY=<your ingest key>

import logging
import logthread

logthread.install()                                   # one line at startup

logging.getLogger(__name__).info("order created", extra={"order_id": 42})

Service name. Records are grouped by service.name. Set it with logthread.init(service="api"), or with the LOGTHREAD_SERVICE environment variable, or with the standard OTEL_SERVICE_NAME (checked in that order). Without any of them the script name (or the app's top-level package under gunicorn/uvicorn) is used, so name your services explicitly.

That is all. Every logging call at info or above is forwarded; extra={...} fields become attributes; logger.exception(...) / exc_info=True adds exception.type, exception.message and exception.stacktrace. A runnable example is in examples/basic.py.

Other ways to use it:

client = logthread.init(service="api", environment="prod")   # explicit options
client.info("hello", {"user_id": 7})                          # direct API: trace/debug/info/warn/error/fatal
logthread.get_logger().error("boom", exc_info=True)

logging.getLogger().addHandler(logthread.LogthreadHandler(client))   # manual wiring

Libraries that bridge to stdlib logging work without extra code; see "structlog and loguru" below.

install(level=None, *, set_root_level=True, **options) attaches the handler with threshold level (default: min_level, i.e. info). Python's root logger defaults to WARNING, so to let info records through install() lowers the root logger level to the handler threshold (it never raises it). This also lets other root handlers (such as a console handler from logging.basicConfig) see those records. Pass set_root_level=False to leave the root level alone.

Configuration#

Precedence: explicit options (keyword arguments to init() / Client() / install()) > environment > defaults. Invalid values fall back to the default with one diagnostic; nothing throws.

With no key or no endpoint the SDK disables itself, prints one line to stderr (logthread: LOGTHREAD_INGEST_KEY is not set; logs are not being sent) and your app carries on.

What it does for your app#

• A log call never blocks, raises, or does I/O. It checks the level, builds the record and appends it to a bounded in-memory queue (about 10 to 20 microseconds per call on a laptop).
• Bounded memory. The queue never exceeds max_queue_records and max_queue_bytes. When full, the *new* record is dropped and counted (dropped_queue_full). At most one batch is in flight and at most one batch awaits a retry.
• Background delivery. One daemon thread does all network I/O with blocking http.client. It never uses your event loop or thread pools and never keeps the process alive.
• No feedback loops. The SDK's own output is never forwarded: loggers named logthread or logthread.*, records raised from the worker thread, and records logged from inside on_error are ignored.
• Failures are swallowed. Network and gateway errors are counted in stats() and reported through on_error / debug only.
• Fork safe. In a forked child (gunicorn, multiprocessing) the inherited queue is discarded and a fresh worker starts on demand.
• Bounded shutdown. logthread.shutdown(timeout=5) and flush(timeout=5) never take longer than their timeout and never raise. A single atexit hook flushes on interpreter exit. Logging after shutdown is a no-op.

Shutdown and signals#

logthread.shutdown(timeout=5) flushes and stops the client; an atexit hook calls it on a normal exit. A process that is *killed* never runs atexit, so on SIGTERM (Kubernetes, systemd, docker stop) the last seconds of logs are lost unless something flushes. Two ways:

• handle_signals=True (env LOGTHREAD_HANDLE_SIGNALS=true): on SIGTERM or SIGINT the SDK flushes for up to shutdown_timeout_ms, then runs the handler that was installed before it (so Gunicorn, Celery or your own handler keep control) or, when there was none, restores the default and raises the signal again so the process still ends the way it would have (SIGINT still raises KeyboardInterrupt). Python only allows this from the main thread, so create the client there; otherwise one line is printed and nothing is installed.
• A shutdown hook in your server, which works from any thread and is the right choice for servers that manage their own signals:

# gunicorn.conf.py
import logthread
def worker_exit(server, worker):          # also fine in post_worker_init for logthread.init(...)
    logthread.shutdown()

# FastAPI / Starlette lifespan (uvicorn)
@contextlib.asynccontextmanager
async def lifespan(app):
    yield
    logthread.shutdown()

# Celery
from celery.signals import worker_shutdown
worker_shutdown.connect(lambda **kw: logthread.shutdown())

Masking values (before_send)#

redact_keys masks by attribute name. For values hiding in free text, pass before_send. It receives a dict {"level": "info", "message": str, "attributes": {...}} after attributes are flattened and redact_keys has run, in the logging thread (keep it quick). Return the event, edited in place or a new dict, to send it; return None or False to drop it. The result goes through the same rules as any record: logthread.* names are dropped, redact_keys applies to keys you add, and the attribute and size limits hold. If the hook raises, the record is dropped (never sent unmasked) and on_error is told.

def scrub(event):
    event["message"] = re.sub(r"\b\d{13,16}\b", "[card]", event["message"])
    if event["attributes"].get("path") == "/healthz":
        return None                       # drop noise
    return event

logthread.init(before_send=scrub)

Proxy and TLS#

Log delivery honours HTTPS_PROXY / HTTP_PROXY (either case) and NO_PROXY (host, .suffix, host:port, *). Loopback addresses are never proxied from the environment. proxy="http://user:pass@proxy:3128" sets one in code, proxy=False ignores the environment. HTTPS goes through a CONNECT tunnel, so the proxy never sees the key or the logs. Only http:// proxy URLs are supported; anything else prints one line and is ignored. create_annotation follows the same rules.

To trust a private CA, or the CA of a TLS-inspecting proxy, set ca_file (or LOGTHREAD_CA_FILE) to a PEM file; the built-in roots stay trusted. Redirects are never followed.

structlog and loguru#

structlog bridges to the standard library, so the Logthread handler sees its records. End the processor chain with render_to_log_kwargs so bound values become attributes:

import logging, structlog, logthread

logthread.install()
structlog.configure(
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.stdlib.render_to_log_kwargs,        # event_dict -> logging's msg + extra
    ],
)
structlog.get_logger("shop").info("order created", order_id=42)   # attribute order_id

loguru has its own sink API; use loguru_sink:

from loguru import logger
import logthread

logger.remove()                                          # optional: drop loguru's stderr sink
logger.add(logthread.loguru_sink(), level="INFO")        # any loguru level, mapped by number
logger.bind(order_id=42).info("order created")           # bind() / contextualize() values become attributes
logger.exception("failed")                               # exception.type / message / stacktrace

The sink skips records from logthread.* and never raises. loguru is not a dependency; the sink only reads the record loguru hands it.

structlog, without the stdlib bridge. structlog_processor() sends the event dict directly and returns it unchanged, so the renderer after it still runs:

import structlog, logthread

structlog.configure(processors=[
    structlog.contextvars.merge_contextvars,
    structlog.processors.add_log_level,                  # level
    structlog.processors.format_exc_info,                # optional: exc_info -> "exception" string
    logthread.structlog_processor(),                     # event -> body, logger -> logger.name, rest -> attributes
    structlog.processors.JSONRenderer(),
])
structlog.get_logger("shop").info("order created", order_id=42)

level / log_level (else the method name) is the level, exc_info or exception becomes exception.*, a timestamp becomes the event time, the request correlation id is added, and the processor never raises.

Delivery and retries#

Batches are sent in order, one at a time, as OTLP/HTTP JSON to POST {endpoint}/v1/logs (gzip-compressed from 1 KiB). The gateway answers 200 only after a durable queue write.

After max_retries retries (so max_retries + 1 attempts) the batch is dropped (dropped_retries_exhausted) and delivery moves on. Delivery is at-least-once: a retry after an unknown outcome may duplicate, so every record carries a stable event_id attribute (a random UUID generated at log time).

Record model#

• Attribute values: strings, bools, ints (intValue), floats (doubleValue; NaN/Inf become strings), dates (RFC 3339 strings); None is skipped. Nested mappings flatten to dotted keys up to three segments (a.b.c); anything deeper, and lists/tuples/sets, becomes a JSON string. Unserializable values become "[unserializable]" without losing the record.
• Limits: body 32 KiB, attribute key 128 chars, string values 8 KiB (all truncated with ...[truncated]), at most 124 attributes per record (the gateway allows 128 and stamps 4 itself; extras are dropped and counted in sdk.attributes_dropped). The logthread.* namespace is reserved and dropped.
• Severity: trace 1, debug 5, info 9, warn 13, error 17, fatal 21. stdlib levels map by range (CRITICAL is fatal).
• logging records add logger.name, code.filepath (the file name), code.lineno, code.function.

Correlating with traces#

If opentelemetry is installed and a span is active when you log, the record carries its trace_id and span_id. Pass your own with extra={"trace_id": "<32 hex>", "span_id": "<16 hex>"} to override. Any other attribute (for example request_id) is sent as is. OpenTelemetry is never required for logging; logthread[otel] adds the traces and metrics side.

Traces and metrics (logthread[otel])#

logthread.otel.start_telemetry() points OpenTelemetry at Logthread in one call: traces and metrics go to the same endpoint, with the same ingest key, as your logs (OTLP/HTTP protobuf, gzip, Authorization: Bearer). The logging SDK keeps no dependencies; the OpenTelemetry packages come only with the extra, and are imported only when start_telemetry() runs.

pip install 'logthread[otel] @ https://downloads.logthread.com/sdk/python/0.2.1/logthread-0.2.1-py3-none-any.whl'
pip install opentelemetry-distro && opentelemetry-bootstrap -a install   # optional: instrument Django, Flask, FastAPI, requests, httpx, psycopg, SQLAlchemy, redis, celery, ...

# first lines of the entry file, before the code you want traced is imported
from logthread.otel import start_telemetry
telemetry = start_telemetry(service_name="api", environment="production")
# ... on shutdown (also runs at interpreter exit):
telemetry.shutdown()                 # flushes queued spans and metrics

Or with no code change, from the environment only (LOGTHREAD_ENDPOINT, LOGTHREAD_INGEST_KEY, LOGTHREAD_SERVICE, ...):

python -m logthread.otel app.py              # a script
python -m logthread.otel -m uvicorn app:app  # a module

A log written inside a span carries its trace_id and span_id (see above), and the span carries the same ids, so no extra step links them.

Cardinality guard. Every metric point loses per-request attributes (url.full, url.path, url.query, http.url, http.target, user_agent.original, user.id, enduser.id, session.id, request.id, client.address, peer addresses and ports, trace_id, span_id) and has long values cut, points that become identical are merged, and each metric is bounded to cardinality_limit series per export (the rest are summed into one overflow series, so totals stay exact). A user id or a raw path can therefore never turn into a time series; use route templates (http.route) for endpoints. The Python metrics SDK cannot drop attributes by name or cap series in a View, so the guard wraps the OTLP metric exporter instead (logthread.otel.guard_metrics_data is the function it applies). The .NET distro uses an allow-list on the meters it wires; here the guard covers every instrument, including your own.

Nothing is sent, and one line goes to stderr, when the key or endpoint is missing; if an OpenTelemetry package is missing, start_telemetry() raises an ImportError that names pip install 'logthread[otel]'. Call it once per process: it sets the global OpenTelemetry providers, which OpenTelemetry allows only once. Tested with opentelemetry-sdk 1.27, 1.35 and 1.45. Python does not run atexit hooks on SIGTERM: under gunicorn, uvicorn or Celery call telemetry.shutdown() in the same hooks as logthread.shutdown() (see above). Logs keep going through logthread.install().

Correlation ids#

Every log written while handling one HTTP request carries the same correlation_id attribute, so Logthread can group them across services. Add the middleware once; no other code changes.

How it works: the first configured request header that is present (default x-correlation-id, then x-request-id; option correlation_headers, env LOGTHREAD_CORRELATION_HEADERS, comma-separated) is used if, trimmed, it is 1 to 128 characters of [A-Za-z0-9._:-]. Anything else, or no header, falls back to the trace id of a valid W3C traceparent header (lowercase), and failing that a freshly generated UUID v4. The id lives in a contextvars context for the whole request (awaits, asyncio.to_thread, anyio/Starlette threadpools) and is echoed on the response as X-Correlation-Id (the first configured name) unless the app already set that header or echo_correlation_header=False (env LOGTHREAD_ECHO_CORRELATION_HEADER=false). An explicit correlation_id attribute in a log call wins; logs outside a request carry none.

Put the middleware as early as possible.

FastAPI / Starlette / Quart / Django ASGI

import logthread

app.add_middleware(logthread.CorrelationMiddleware)       # Starlette, FastAPI
# or wrap any ASGI app:  app = logthread.CorrelationMiddleware(app)

Flask / any WSGI app (including Django WSGI)

app.wsgi_app = logthread.WSGICorrelationMiddleware(app.wsgi_app)

Django (sync or async; Django is not needed to import the module)

MIDDLEWARE = [
    "logthread.django.CorrelationMiddleware",   # first
    # ...
]

Background jobs and queue consumers

with logthread.with_correlation_id(message.headers.get("x-correlation-id")):
    handle(message)                # invalid or missing id: a UUID is generated

@logthread.with_correlation_id     # a fresh id per call; works for async functions too
def nightly_job(): ...

logthread.get_correlation_id()     # the current id, or None

Outgoing calls (opt-in; the header is added only inside a context and never overwrites one the request already has):

logthread.instrument_requests()    # all `requests` calls (no-op if requests is absent)
logthread.instrument_httpx()       # httpx clients created after this call (no-op if absent)
headers = logthread.correlation_headers()   # {"X-Correlation-Id": id} for manual use

Threads and executors. asyncio.to_thread and asyncio.create_task carry the context. A plain threading.Thread or loop.run_in_executor starts with an empty one, so wrap the target where the request is handled with logthread.copy_context, which copies every contextvars value (the correlation id included):

threading.Thread(target=logthread.copy_context(work), args=(order,)).start()
await loop.run_in_executor(None, logthread.copy_context(work), order)
pool.submit(logthread.copy_context(work), order)

Options are read from the process-wide client set by logthread.init() / install(), from middleware arguments (CorrelationMiddleware(app, headers=[...], echo=False)), or the environment.

Mark a deploy or event (annotations)#

Draw a marker on your charts when something happens, such as a deploy. This needs an API key with the "Create annotations" permission (Keys → API keys), not an ingest key. Set it as LOGTHREAD_API_KEY; LOGTHREAD_API_URL defaults to https://logthread.com. time defaults to now and must be within the last 30 days; times are sent as UTC ISO 8601 with a Z. Errors carry the standard code and message (forbidden for a key without the permission, invalid_request for a bad body).

import logthread

item = logthread.create_annotation(
    "deploy 1.4.2",
    text="rolled out to production",   # optional
    service="api",                      # optional
    tags=["deploy"],                    # optional
    # time= / end=: datetime or ISO string. api_key= / endpoint= override the environment.
)
# failures raise logthread.LogthreadApiError with .code, .message and .status

The Python SDK ships no command line tool, so CI uses the one-liner or curl below. It uses only the standard library.

In CI, a GitHub Actions step after your deploy:

- name: Mark deploy in Logthread
  env:
    LOGTHREAD_API_KEY: ${{ secrets.LOGTHREAD_API_KEY }}
  run: |
    python -c "import os, logthread; logthread.create_annotation('deploy ' + os.environ['GITHUB_SHA'][:7], service='api', tags=['deploy'])"

The same call without the SDK:

curl -fsS -X POST https://logthread.com/api/annotations \
  -H "Authorization: Bearer $LOGTHREAD_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"title":"deploy '"$GITHUB_SHA"'","service":"api","tags":["deploy"]}'

Troubleshooting#

• Set LOGTHREAD_DEBUG=true (or debug=True) to see diagnostics on stderr, or pass on_error=print.
• logthread.stats() returns queued, sent_records, sent_batches, dropped_queue_full, dropped_rejected, dropped_retries_exhausted, dropped_shutdown, retries, last_error and disabled_until (epoch seconds while the 401/403 cool-off is active, else None).
• Nothing arrives and no error: check the startup stderr line (missing key/endpoint) and that your logger level lets info through (see install() above).
• Short-lived scripts: call logthread.flush() (the atexit hook also flushes, for up to 5 s).

Limits#

• install() / init() send logs only. Traces and metrics go through logthread.otel.start_telemetry() (above), or through any standard OpenTelemetry setup pointed at the same Logthread endpoint and ingest key: OTEL_EXPORTER_OTLP_ENDPOINT=$LOGTHREAD_ENDPOINT, OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer%20$LOGTHREAD_INGEST_KEY. Onboarding has the exact snippet per language. Either way this SDK attaches the active trace and span ids to log records so logs and traces link up.
• Delivery is best effort under pressure: when the queue is full or the gateway is down for longer than the retry budget, records are dropped (and counted) rather than blocking your app.
• Retry-After in HTTP-date form is not parsed (falls back to backoff). request_timeout_ms is a per-socket-operation timeout, not a total deadline. Only http:// proxies are supported.

Report a change#

Report a deploy, config change, flag flip or migration so an incident can list what changed before it. It uses the same API key as annotations; type is deploy, config, flag, migration, infra or other, and a retry with the same id updates the change. See What changed before an incident.

logthread.create_change("deploy", "checkout 2.4", service="checkout", version="2.4", id=os.environ["GITHUB_RUN_ID"])