Python Logging Handler for Grafana Loki

logger.py

import logging

from logi_handler import LokiHandler

LOKI_URL: str = "http://loki:3100/loki/api/v1/push" # Standard loki url

logger = logging.getLogger(__name__)

loki_handler = LokiHandler(url=LOKI_URL)

date_format = "%Y-%m-%d %H:%M:%S"
log_format = "%(asctime)s %(levelname)-8s %(name)s: %(message)s"

formatter = logging.Formatter(log_format, date_format)

loki_handler.setFormatter(formatter)

logger.addHandler(loki_handler)

logger.setLevel(logging.DEBUG)
logger.propagate = False #  Do not send logs from this logger to the parent logger

logger.info("This is info") # This is going to be visible on the Loki side :)

loki_handler.py

import logging
from logging.handlers import HTTPHandler
import urllib3

# Choosing urllib3 for its performance benefits:
# urllib3 is noted for its speed and efficiency in handling HTTP requests.
# I also tested the differences between urllib3 and requests during load test using k6
# and all tests were passing in my FastAPI app with urllib3
# using requests - all tests were failing.
# REFERENCE for urllib3's performance: https://www.zenrows.com/blog/urllib-vs-urllib3-vs-requests

# Define a timeout for HTTP requests to prevent hanging connections.
TIMEOUT: int = 60

class LokiHandler(HTTPHandler):
    def __init__(self, url: str):
        logging.Handler.__init__(self)
        
        self.url = url

    def mapLogRecord(self, record):
        # Maps the log record into the format expected by Loki.
        # The record.created attribute is a float timestamp in seconds since the epoch.
        # Multiplying by 1e9 (1,000,000,000) to convert seconds to nanoseconds, as required by Loki.
        # Check: https://grafana.com/docs/loki/latest/reference/api/#ingest-logs
        return {
            "streams": [
                {
                    "stream": {"app": "<Your App>"},  # Define the app label for the log stream.
                    "values": [[str(int(record.created * 1e9)), self.format(record)]],
                }
            ]
        }

    def emit(self, record):
        # Send the log record to Loki.
        mapped_log_record = self.mapLogRecord(record)

        # Use urllib3 to post the log record to Loki.
        urllib3.request(
            "POST",
            self.url,
            body=json.dumps(mapped_log_record),  # Convert the record to JSON.
            headers={"Content-Type": "application/json"},
            timeout=TIMEOUT,  # Use the predefined timeout.
        )

        # To check Loki's response you can do `r = urllib3.request(...); print(r.status)`
        # `204` means log was successfuly send to Loki.