Last active
March 15, 2023 23:46
-
-
Save byt3bl33d3r/5723748bc40748777111b03354a3c2b5 to your computer and use it in GitHub Desktop.
Structured logging and event capture
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 | |
) | |
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#! /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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment