tg-logging-handler

Production-grade Telegram destination for Python's standard logging module.

Attach one handler and your logs go to a Telegram chat — asynchronously, so a slow or down Bot API never blocks your application. Batching, retries with backoff, 429 handling, oversized-message splitting, bounded-queue back-pressure, and parse_mode escaping are all handled for you.

Install

pip install tg-logging-handler

Requires Python 3.10+. The only runtime dependency is httpx.

Quickstart

import logging
from tg_logging_handler import TelegramLoggingHandler

# Reads TG_TOKEN / TG_CHAT_ID from the environment.
logging.getLogger().addHandler(TelegramLoggingHandler())

logging.error("Something broke")  # delivered to your Telegram chat

TGLoggingHandler is a shorter alias for the same class.

Provide credentials explicitly instead of via the environment:

handler = TelegramLoggingHandler(token="123456:ABC-DEF", chat_id=-100123456789)
logging.getLogger().addHandler(handler)

Why this handler

• Never blocks your app. emit() only snapshots the record and enqueues it; a single daemon thread does all network I/O. A dead Bot API can never stall a request handler.
• Never crashes your app. Send failures are retried, then reported to stderr and counted — never raised into your code. The only exception you can catch is TelegramConfigError, and only at construction time.
• No logging recursion. Internal diagnostics go to stderr, never back through logging.

Configuration

All constructor arguments after token / chat_id are keyword-only.

Environment variables

Batching

Batch records into fewer messages to stay well under Telegram's rate limits:

handler = TelegramLoggingHandler(
    level=logging.ERROR,
    batch_size=10,  # up to 10 records per message
    flush_interval=10.0,  # ...or flush a partial batch after 10s
)

A batch is sent when it fills (batch_size) or when flush_interval elapses, whichever comes first.

parse_mode / rich formatting

Set parse_mode and the handler escapes each formatted record so log content (tracebacks, arbitrary user data) can never break Telegram's parser:

handler = TelegramLoggingHandler(parse_mode="MarkdownV2")

Because the whole formatter output is escaped, a message like v1.2_final is delivered literally rather than 400-ing the request or rendering as accidental formatting.

dictConfig

LOGGING = {
    "version": 1,
    "handlers": {
        "telegram": {
            "()": "tg_logging_handler.TelegramLoggingHandler",
            "level": "ERROR",
            "batch_size": 5,
        },
    },
    "root": {"handlers": ["telegram"], "level": "WARNING"},
}

import logging.config

logging.config.dictConfig(LOGGING)

Inspecting delivery

handler.stats returns an immutable snapshot of cumulative counters:

s = handler.stats
print(s.queued, s.sent, s.batches_sent, s.retries, s.failed, s.dropped)

Shutdown

close() drains the queue (bounded by shutdown_timeout), stops the worker, and releases the HTTP client. It is idempotent and registered via atexit, so it runs automatically at interpreter exit. For deterministic tests, call it explicitly or use logging.shutdown().

Development

uv sync                          # create .venv + install package + dev deps
uv run pytest                    # run the test suite
uv run ruff check . && uv run mypy   # lint + type-check