mihari-logger 0.1.0

Lightweight log collection and transport library for Python

Mihari Logger

Lightweight log collection and transport library for Python.

Mihari Logger collects structured log entries and ships them to an HTTP API with automatic batching, gzip compression, retry with exponential backoff, and graceful shutdown. It works as a standalone client or as a drop-in logging.Handler for Python's standard library.

Features

• Structured JSON log entries with ISO 8601 timestamps
• Automatic batching (configurable batch size and flush interval)
• Gzip compression (enabled by default)
• Retry with exponential backoff on transient failures
• Thread-safe background flushing
• Graceful shutdown via atexit
• Auto-captured metadata: hostname, PID, Python version, platform
• Python logging.Handler integration
• Built on httpx for modern HTTP support

Installation

pip install mihari-logger

Quick Start

Standalone Client

from mihari import Mihari

client = Mihari(
    token="your-api-token",
    endpoint="https://logs.example.com",
)

client.info("Application started", version="1.2.0", port=8080)
client.warn("Cache miss rate high", rate=0.42)
client.error("Connection failed", host="db.local", retries=3)
client.debug("Query executed", sql="SELECT ...", duration_ms=12)
client.fatal("Out of memory", heap_mb=1024)

# Logs are flushed automatically on exit, or call:
client.close()

With Python's logging Module

import logging
from mihari import MihariHandler

handler = MihariHandler(
    token="your-api-token",
    endpoint="https://logs.example.com",
)

logging.basicConfig(handlers=[handler], level=logging.INFO)
logger = logging.getLogger("myapp")

logger.info("Server started")
logger.warning("Disk usage at 85%%")
logger.error("Request failed", extra={"path": "/api/users", "status": 500})

With structlog

import structlog
import logging
from mihari import MihariHandler

handler = MihariHandler(
    token="your-api-token",
    endpoint="https://logs.example.com",
)
logging.basicConfig(handlers=[handler], level=logging.DEBUG, format="%(message)s")

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.StackInfoRenderer(),
        structlog.dev.set_exc_info,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
)

log = structlog.get_logger()
log.info("payment.processed", amount=99.99, currency="USD")

Configuration

Client Options

from mihari import Mihari

client = Mihari(
    token="your-api-token",
    endpoint="https://logs.example.com",
    transport_config={
        "batch_size": 10,        # Entries per batch (default: 10)
        "flush_interval": 5.0,   # Seconds between flushes (default: 5.0)
        "max_retries": 3,        # Retry attempts (default: 3)
        "base_delay": 1.0,       # Initial retry delay in seconds (default: 1.0)
        "max_delay": 30.0,       # Maximum retry delay in seconds (default: 30.0)
        "gzip": True,            # Enable gzip compression (default: True)
        "timeout": 10.0,         # HTTP timeout in seconds (default: 10.0)
    },
    default_meta={               # Attached to every log entry
        "service": "payment-api",
        "env": "production",
    },
)

Handler Options

from mihari import MihariHandler
import logging

handler = MihariHandler(
    token="your-api-token",
    endpoint="https://logs.example.com",
    level=logging.WARNING,       # Only forward WARNING+ to Mihari
    transport_config={...},      # Same options as above
    default_meta={"service": "web"},
)

Log Entry Format

Each log entry is sent as JSON:

{
  "dt": "2024-01-15T10:30:00.123456Z",
  "level": "info",
  "message": "Request handled",
  "hostname": "web-01",
  "pid": "12345",
  "python_version": "3.12.1",
  "platform": "linux",
  "method": "GET",
  "path": "/api/users"
}

API Specification

• Auth: Bearer token in Authorization header
• Endpoint: POST {base_url}/logs
• Content-Type: application/json
• Content-Encoding: gzip (when enabled)
• Success (202): {"status": "accepted", "count": N}
• Auth Error (401): {"error": "Invalid or missing authentication token"}

Development

# Clone and install in development mode
git clone https://github.com/mihari/mihari-logger-python.git
cd mihari-logger-python
pip install -e ".[dev]"

# Run tests
pytest

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

# Lint
ruff check src/ tests/

# Type check
mypy src/

License

MIT