Extension guide

Formatters

Use Formatter protocol to implement your formatter type. Note that you don’t need to inherit from it. You need to call add_formatter_type() afterward to register your custom formatter type.

from uvlog import add_formatter_type, LogRecord

class IdioticFormatter:

    def format_record(self, record: LogRecord, /) -> bytes:
        return b'Bang!'

add_formatter_type(IdioticFormatter)

Handlers

Use Handler protocol to implement your handler type. Note that you don’t need to inherit from it. You need to call add_handler_type() afterward to register your custom handler type.

You can create your own queue handler by inheriting from QueueHandler. You must implement open_stream() and close_stream() methods and also py:meth:~uvlog.handlers.QueueHandler.write_records method which should write a batch of formatted log records (byte strings) to your destination. You also would need to create your own set_destination() method.

See QueueStreamHandler as an example.

Example: HTTP logger

Here’s an example of a simple HTTP queue handler using requests library.

In this example a batch of logs is sent to some url in NDJSON format (newline delimited JSON). You may of course customize the write_records method to suit your own needs.

import requests
from typing import List
from uvlog import QueueHandler, JSONFormatter, add_handler_type

class QueueHTTPHandler(uvlog.QueueHandler):

    terminator = b'\n'

    headers: dict

    def __init__(self):
        QueueHandler.__init__(self)
        self.headers = {}
        self._formatter = JSONFormatter()
        self._url = None
        self._stream = None

    def set_destination(self, url: str, /):
        self._url = urlparse(url).geturl()

    def open_stream(self):
        self._stream = _session = requests.Session()
        _session.headers.update(self.headers)

    def close_stream(self):
        if self._stream is not None:
            self._stream.close()
            self._stream = None

    def write_records(self, records: List[bytes], /):
        self._stream.post(self._url, data=self.terminator.join(records) + b'\n', timeout=1)

add_handler_type(QueueHTTPHandler)

Configuration example.

import os

uvlog.configure({
    'loggers': {
        '': {
            'handlers': ['https://mylogs.com:8888']
        }
    },
    'handlers': {
        'https://mylogs.com:8888': {
            'class': 'QueueHTTPHandler',
            'headers': {
                'Authorization': f'Basic {os.getenv("LOGS_CLIENT_KEY")}'
            }
        }
    }
})