pylogkit 0.1.0

Structured logging with Slack integration, rate limiting, and deduplication

pylogkit

Structured logging with Slack integration, rate limiting, and deduplication.

Built on top of Python's standard logging module — no custom APIs to learn.

Installation

# Core (JSON + colored formatters)
pip install pylogkit

# With Slack support
pip install pylogkit[slack]

# With colored output
pip install pylogkit[color]

# Everything
pip install pylogkit[all]

Or as a git dependency:

# pyproject.toml
dependencies = [
    "pylogkit[all] @ git+ssh://[e-mail]/analytics-team-global/pylogkit.git",
]

Quick Start

import os
from pylogkit import setup_logging, get_logger

setup_logging(
    level="INFO",
    service_name="my-api",
    slack_token=os.environ.get("SLACK_BOT_TOKEN"),
    slack_channel="#errors",
)

logger = get_logger(__name__)

logger.info("Server started", extra={"port": 8000})
logger.error("Request failed", extra={"endpoint": "/users", "status": 500})

Features

Auto-detect output format

• TTY (development): colored human-readable output with extras
• Non-TTY (production/Docker): structured JSON

Force a specific format:

setup_logging(json_format=True)   # always JSON
setup_logging(json_format=False)  # always colored

Structured logging with extra fields

Extra fields are included in both JSON and colored output:

logger.info("Deal processed", extra={
    "deal_id": "123",
    "pipeline": "laba_czech",
    "duration": 1.23,
})

JSON output:

{"timestamp": "2026-03-18T12:00:00+00:00", "level": "INFO", "logger": "myapp.deals", "message": "Deal processed", "deal_id": "123", "pipeline": "laba_czech", "duration": 1.23}

Colored output:

2026-03-18 12:00:00 INFO     myapp.deals — Deal processed  [deal_id=123 pipeline=laba_czech duration=1.23]

Logger with static context

Bind context fields to a logger — they appear in every message:

logger = get_logger(__name__, pipeline="laba_czech", env="production")

logger.info("Course synced")
# All messages from this logger will include pipeline= and env=

Calling get_logger() again with the same name updates the context:

logger = get_logger(__name__, pipeline="v2")
# pipeline is now "v2", env remains from the previous call

Request-scoped context with contextvars

Inject request-scoped fields (request_id, user_id, etc.) without passing extra={} to every log call:

import contextvars
from pylogkit import setup_logging

log_context: contextvars.ContextVar[dict] = contextvars.ContextVar(
    "log_context", default={}
)

setup_logging(
    service_name="my-api",
    ctx_var=log_context,
)

# In your request handler / middleware:
log_context.set({"request_id": "abc-123", "user_id": "42"})
logger.info("Processing request")
# Output includes request_id and user_id automatically

Works with async frameworks (FastAPI, aiohttp) — each coroutine gets its own context.

Slack notifications

ERROR and CRITICAL logs go to Slack with:

• Color-coded attachments (yellow/red/dark red)
• Module and line number
• Extra fields as "Context" block
• Formatted traceback

setup_logging(
    slack_token="xoxb-...",
    slack_channel="#alerts",
    slack_level="ERROR",          # minimum level (default: ERROR)
)

Or use incoming webhook (simpler, scoped to one channel):

setup_logging(
    slack_webhook_url="https://hooks.slack.com/services/T.../B.../xxx",
)

How it works internally:

• Messages are queued and sent from a background thread (non-blocking)
• Respects Slack's 1 msg/sec rate limit
• Exponential backoff on consecutive failures (up to 5 min)
• If queue is full (100 messages) — drops silently, never blocks the app
• If Slack is unreachable — prints to stderr, never crashes
• Flushes remaining messages on shutdown

Rate limiting

Prevents Slack spam. Each unique message can fire at most N times per period:

setup_logging(
    slack_token="xoxb-...",
    slack_channel="#alerts",
    slack_rate_limit=1,       # max 1 message per period (default)
    slack_rate_period=60.0,   # per 60 seconds (default)
)

If the same error fires 500 times in a minute — only the first one goes to Slack.

Deduplication

Suppresses identical errors within a time window. When the window expires, the next message includes the suppressed count:

setup_logging(
    slack_token="xoxb-...",
    slack_channel="#alerts",
    slack_dedupe_window=300.0,  # 5 minutes (default)
)

Example: "Connection refused (suppressed 47 duplicates)"

Set to 0 to disable deduplication.

Per-logger level overrides

Silence noisy third-party libraries:

setup_logging(
    level="INFO",
    loggers={
        "httpx": "WARNING",
        "sqlalchemy.engine": "WARNING",
        "celery": "INFO",
    },
)

Static context fields

Add fields to every log record in the application:

setup_logging(
    service_name="my-api",
    extra_context={
        "env": "production",
        "region": "eu-west-1",
    },
)

Advanced: using components directly

All components can be used independently without setup_logging():

import logging
from pylogkit import JsonFormatter, ColoredFormatter, RateLimitFilter, DeduplicationFilter
from pylogkit import SlackHandler  # lazy import, requires slack-sdk

# Custom handler with JSON
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())

# Slack with custom filters
slack = SlackHandler(token="xoxb-...", channel="#alerts")
slack.addFilter(RateLimitFilter(rate=3, period=120))
slack.addFilter(DeduplicationFilter(window=600))

logging.getLogger().addHandler(handler)
logging.getLogger().addHandler(slack)

Celery integration

Prevent Celery from hijacking your logging setup:

from celery.signals import setup_logging as celery_setup_logging
from pylogkit import setup_logging

@celery_setup_logging.connect
def configure_celery_logging(**kwargs):
    setup_logging(
        service_name="worker",
        slack_token=os.environ.get("SLACK_BOT_TOKEN"),
        slack_channel="#worker-errors",
    )

setup_logging() parameters

Parameter	Type	Default	Description
level	str	"INFO"	Root log level
service_name	str	None	Added to every record as service field
json_format	bool	auto	True = JSON, False = colored, None = auto-detect
slack_token	str	None	Slack Bot token (xoxb-...)
slack_channel	str	None	Slack channel (required with token)
slack_webhook_url	str	None	Incoming webhook URL (alternative to token)
slack_level	str	"ERROR"	Minimum level for Slack
slack_rate_limit	int	1	Max messages per period
slack_rate_period	float	60.0	Rate limit window (seconds)
slack_dedupe_window	float	300.0	Deduplication window (seconds), 0 to disable
extra_context	dict	None	Static fields for every record
loggers	dict	None	Per-logger level overrides
ctx_var	ContextVar	None	Request-scoped context variable

Development

# Install with dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Run tests with coverage
pytest --cov=pylogkit --cov-report=term-missing

# Lint
ruff check src/ tests/
ruff format src/ tests/

# Type check
mypy src/pylogkit/

License

MIT