Structured logging and event capture

api.py

from logger import capturer
from typing import Optional
from fastapi import FastAPI

app = FastAPI()

@app.get("/logs")
async def get_logs(event_name: Optional[str] = None):

    if not event_name:
        return capturer.entries

    if event_name:
        filter_expression = lambda l: l['event'] == event_name

    return list(
        filter(
            filter_expression,
            capturer.entries
        )
    )

app.py

#! /usr/bin/env python3

import logging
import logging.handlers
from time import sleep

from logger import (
    log,
    plain_formatter,
    json_capture_formatter,
    capturer
)

from pprint import pprint

# Log to a file as well as console
fh = logging.handlers.WatchedFileHandler("myapp.log")
fh.setFormatter(plain_formatter)
log.addHandler(fh)

# This will effectively duplicate all logs and it will save a copy of them in capturer.entries in JSON format
jh = logging.StreamHandler()
jh.setFormatter(json_capture_formatter)
log.addHandler(jh)

log.info('this is an event', event_data='some good stuff', interface='127.0.0.1', hacked='everything')

pprint(capturer.entries)

while True:
    sleep(5)

logger.py

import logging
import logging.handlers
import structlog

capturer = structlog.testing.LogCapture()
timestamper = structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S")

plain_formatter = structlog.stdlib.ProcessorFormatter(
    processors = [
        structlog.stdlib.ProcessorFormatter.remove_processors_meta,
        #structlog.stdlib.add_log_level,
        structlog.dev.ConsoleRenderer(colors=False),
    ]
)

colored_formatter = structlog.stdlib.ProcessorFormatter(
    processors = [
        structlog.stdlib.ProcessorFormatter.remove_processors_meta,
        #structlog.stdlib.add_log_level,
        structlog.dev.ConsoleRenderer(colors=True, exception_formatter=structlog.dev.rich_traceback),
    ]
)

debug_formatter = structlog.stdlib.ProcessorFormatter(
    processors = [
        structlog.stdlib.ProcessorFormatter.remove_processors_meta,
        structlog.stdlib.add_log_level,
        #structlog.stdlib.add_logger_name,
        structlog.dev.ConsoleRenderer(colors=True, exception_formatter=structlog.dev.rich_traceback),
    ]
)

json_capture_formatter = structlog.stdlib.ProcessorFormatter(
    processors = [
        structlog.stdlib.ProcessorFormatter.remove_processors_meta,
        capturer,
        structlog.processors.JSONRenderer(),
    ]
)

sh = logging.StreamHandler()
sh.setFormatter(colored_formatter)

myapp_logger = logging.getLogger("myapp")
myapp_logger.setLevel(logging.INFO)
myapp_logger.addHandler(sh)

structlog.configure(
    processors=[
        #structlog.stdlib.filter_by_level,
        #structlog.stdlib.add_logger_name,
        #structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        timestamper,
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        #structlog.processors.UnicodeDecoder(),
        structlog.stdlib.ProcessorFormatter.wrap_for_formatter
    ],
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

log = structlog.get_logger("myapp")