xOgYwlT-loki-logger 1.0.3

A python loggin library for Grafana Loki

loki_logger

A Python logging library for Grafana Loki with a clean class hierarchy, multiple handlers, structured context, and HTTP/async push support.

---

Installation

# No external dependencies required for synchronous usage.
pip install loki_logger          # from PyPI (when published)

# For async support:
pip install aiohttp

---

Quick Start

from loki_logger import configure, get_logger, LokiHandler, LogLevel

configure(
    level=LogLevel.INFO,
    handlers=[
        LokiHandler(
            url="http://localhost:3100",
            labels={"app": "my-service", "env": "production"},
        )
    ],
    default_labels={"region": "us-east-1"},
)

logger = get_logger("myapp")
logger.info("Server started", port=8080)
logger.error("Database timeout", exc_info=True, db_host="pg01")

---

Architecture

loki_logger/
├── levels.py        LogLevel enum  (DEBUG / INFO / WARNING / ERROR / CRITICAL)
├── record.py        LogRecord dataclass
├── formatters.py    BaseFormatter, JSONFormatter, TextFormatter, LabelFormatter
├── handlers.py      BaseHandler, ConsoleHandler, LokiHandler,
│                    AsyncLokiHandler, BufferedLokiHandler
├── logger.py        Logger class + global registry (get_logger / configure)
├── context.py       Thread-local context binding (LogContext, bind_context, …)
└── middleware.py    WSGI / ASGI HTTP logging middleware

---

Handlers

ConsoleHandler

Writes formatted lines to stdout (DEBUG/INFO) or stderr (WARNING+).

from loki_logger import ConsoleHandler, TextFormatter

handler = ConsoleHandler(formatter=TextFormatter())

LokiHandler

Synchronous HTTP POST to /loki/api/v1/push per log record.

from loki_logger import LokiHandler

handler = LokiHandler(
    url="http://localhost:3100",
    labels={"app": "api"},
    timeout=5.0,
    username="admin",   # optional — for Grafana Cloud
    password="secret",
)

BufferedLokiHandler

Batches records and flushes them periodically or when the buffer is full.

from loki_logger import BufferedLokiHandler

handler = BufferedLokiHandler(
    url="http://localhost:3100",
    labels={"app": "api"},
    max_batch_size=100,    # flush when 100 records accumulate
    flush_interval=5.0,    # also flush every 5 seconds
)

# Always close cleanly on shutdown:
handler.close()

AsyncLokiHandler

For asyncio applications. Schedules pushes as tasks on the running event loop.

import asyncio
from loki_logger import AsyncLokiHandler, Logger

handler = AsyncLokiHandler(url="http://localhost:3100")
logger = Logger("myapp", handlers=[handler])

async def main():
    logger.info("Async event", user_id=42)
    await handler.flush()   # wait for all pending pushes

asyncio.run(main())

---

Formatters

Class	Output style
JSONFormatter	Single-line JSON object
TextFormatter	Configurable human-readable text
LabelFormatter	key=value pairs (great for LogQL)

from loki_logger import JSONFormatter, TextFormatter, LabelFormatter

# JSON (default)
JSONFormatter(indent=None)

# Text with custom template
TextFormatter(template="[{timestamp}] {level}: {message} {extra}")

# Key=value pairs
LabelFormatter()

---

Structured Context

Attach fields automatically to all records created on the current thread:

from loki_logger import bind_context, clear_context, LogContext, get_logger

logger = get_logger()

# Manual bind / clear
bind_context(request_id="req-123", user_id=42)
logger.info("Processing")   # ← includes request_id & user_id
clear_context()

# Context manager (recommended)
with LogContext(trace_id="abc", tenant="acme"):
    logger.info("Inside block")   # ← includes trace_id & tenant
# context is automatically restored here

---

Child Loggers

from loki_logger import get_logger

root = get_logger("myapp")
db   = root.get_child("db")        # name = "myapp.db"
http = root.get_child("http")      # name = "myapp.http"

db.info("Query executed", query_ms=12)

---

HTTP Middleware

WSGI (Flask / Django)

from loki_logger import LoggingMiddleware, get_logger

app.wsgi_app = LoggingMiddleware(
    app.wsgi_app,
    logger=get_logger("http"),
    exclude_paths={"/healthz", "/metrics"},
)

ASGI (FastAPI / Starlette)

from loki_logger import LoggingMiddleware, get_logger

app.add_middleware(
    LoggingMiddleware,
    logger=get_logger("http"),
    exclude_paths={"/healthz"},
)

Each request emits a record with: method, path, status_code, duration_ms, remote_addr.

---

Loki Payload Format

Records are grouped by label set and pushed as Loki streams:

{
  "streams": [
    {
      "stream": { "app": "api", "env": "prod", "level": "info" },
      "values": [
        ["1700000000000000000", "{\"level\":\"INFO\",\"message\":\"Hello\"}"]
      ]
    }
  ]
}

---

API Reference (Summary)

# Global logger registry
logger = get_logger("myapp")            # create or retrieve
configure(level, handlers, ...)        # configure root logger

# Logger instance
logger.debug / info / warning / error / critical / exception(msg, **extra)
logger.add_handler(handler)
logger.remove_handler(handler)
logger.get_child("suffix")

# LogLevel
LogLevel.DEBUG / INFO / WARNING / ERROR / CRITICAL
LogLevel.from_string("info")

# LogRecord
record.timestamp_ns          # nanoseconds (required by Loki)
record.as_dict()             # dict representation
record.merged_labels(defaults)

# Context
bind_context(**fields)
clear_context()
get_context() -> dict
LogContext(**fields)         # context manager

---

License

MIT