python-telegram-logging 0.0.2

A Python logging handler that sends logs to Telegram

Python Telegram Logging

A Python logging handler that sends logs to Telegram with support for both synchronous and asynchronous operations.

Table of Contents

• Who is Target User?
• Features
• Installation
• Quick Start
  • Synchronous Usage
  • Asynchronous Usage
  • Queued Usage (for synchronous handlers)
• Advanced Usage
  • Custom Formatting
  • Error Handling
• Handler Comparison
• Technical Details
• Requirements
• License
• TODO

Who is Target User?

This package is ideal for developers with simple Python applications who want to receive error notifications directly via Telegram. Suppose you lack the resources to maintain a full-fledged logging/alerting setup using tools like Kibana, Elasticsearch, or Grafana, or do not have time to configure complex alert rules. In that case, this package provides a lightweight and hassle-free solution

Features

• 🚀 Multiple Handler Types:

  • SyncTelegramHandler: Synchronous handler using requests
  • AsyncTelegramHandler: Asynchronous handler with built-in queue using aiohttp
  • QueuedTelegramHandler: Thread-safe queued wrapper for synchronous handlers only (currently)

• 🔒 Thread Safety: All handlers are thread-safe and can be used in multi-threaded applications

• 🎨 Formatting Support:

  • HTML formatting
  • Markdown formatting
  • Custom formatters support

• 🛡️ Error Handling:

  • Rate limiting with configurable strategies (TODO)
  • Automatic message splitting for long logs
  • Custom error callbacks (in case if message is not sent, raised exception)

Installation

pip install python-telegram-logging

Quick Start

Use one of the exanple below or check examples folder.

Synchronous Usage

import logging
from python_telegram_logging import SyncTelegramHandler, ParseMode

# Create and configure the handler
handler = SyncTelegramHandler(
    token="YOUR_BOT_TOKEN",
    chat_id="YOUR_CHAT_ID",
    parse_mode=ParseMode.HTML
)

# Add it to your logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.addHandler(handler)

# Use it!
logger.info("Hello from Python! 🐍")

Asynchronous Usage

import logging
from python_telegram_logging import AsyncTelegramHandler, ParseMode

# Create and configure the handler
handler = AsyncTelegramHandler(
    token="YOUR_BOT_TOKEN",
    chat_id="YOUR_CHAT_ID",
    parse_mode=ParseMode.HTML
)

# Add it to your logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.addHandler(handler)

# Use it in your async code!
logger.info("Hello from async Python! 🐍")

Queued Usage (for synchronous handlers)

⚠️ Important: QueuedTelegramHandler is designed to work with synchronous handlers only. For asynchronous applications, use AsyncTelegramHandler directly as it already includes queue functionality.

import logging
from python_telegram_logging import SyncTelegramHandler, QueuedTelegramHandler, ParseMode

# Create the base synchronous handler
base_handler = SyncTelegramHandler(
    token="YOUR_BOT_TOKEN",
    chat_id="YOUR_CHAT_ID",
    parse_mode=ParseMode.HTML
)

# Wrap it in a queued handler for non-blocking operation
handler = QueuedTelegramHandler(base_handler, queue_size=1000)

# Add it to your logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
logger.addHandler(handler)

# Use it!
logger.info("Hello from queued logger! 🐍")

Advanced Usage

Custom Formatting

import logging
from python_telegram_logging import SyncTelegramHandler, ParseMode

# Create a custom formatter
class HTMLFormatter(logging.Formatter):
    def format(self, record):
        return f"""
<b>{record.levelname}</b>: {record.getMessage()}
<code>
File: {record.filename}
Line: {record.lineno}
</code>
"""

# Use the custom formatter
handler = SyncTelegramHandler(
    token="YOUR_BOT_TOKEN",
    chat_id="YOUR_CHAT_ID",
    parse_mode=ParseMode.HTML
)
handler.setFormatter(HTMLFormatter())

Error Handling

from python_telegram_logging import SyncTelegramHandler

def on_error(error: Exception):
    print(f"Failed to send log to Telegram: {error}")

handler = SyncTelegramHandler(
    token="YOUR_BOT_TOKEN",
    chat_id="YOUR_CHAT_ID",
    error_callback=on_error
)

Handler Comparison

Feature	SyncTelegramHandler	AsyncTelegramHandler	QueuedTelegramHandler
Blocking	Yes	No	No
Thread-Safe	Yes	Yes	Yes
Dependencies	requests	aiohttp	-
Use Case	Simple scripts	Async applications	High-performance sync apps
Message Order	Guaranteed	Best-effort	Best-effort
Queue Support	No	Built-in	Yes (sync handlers only)
Handler Type	Sync	Async	Sync wrapper

Technical Details

• Rate limiting: Implements a token bucket algorithm to respect Telegram's rate limits
• Message splitting: Automatically splits messages longer than 4096 characters
• Thread safety: Uses appropriate synchronization primitives for each context
• Resource management: Proper cleanup of resources on handler close
• Error handling: Configurable error callbacks and retry strategies

Requirements

• Python 3.8+
• requests >= 2.31.0
• aiohttp >= 3.8.6

License

MIT License

Contributing

Feel free to open issues or PR.

Please install and run pre-commit rules .pre-commit-config.yaml:

pre-commit install && ` pre-commit run --all-files

TODO

• what if queue is full?

Notes: Configurable blocking behavior: Added block_on_full option to either block when the queue is full or continue with alternative handling. Selective message dropping: Added discard_level_on_full to allow dropping less important messages (e.g., DEBUG) while ensuring critical messages (ERROR, CRITICAL) are handled. Timeout control: Added timeout parameter to prevent indefinite blocking if block_on_full is True. Default behavior: By default, it will:

• Not block (block_on_full=False)
• Silently drop DEBUG messages when full
• Call handleError() for more important messages