AWS Simple Logger (for when you know logs will be sent to CloudWatch Logs)

aws_logger.py

"""
Logging utility for logging to CloudWatch Logs, to use in e.g. AWS Lambda.

Some small features included:
- log structured JSON as that works really well with CloudWatch Logs filters and CloudWatch Logs Insights
- support turning any Python object into JSON
- replace line endings for nice folding of entries in CloudWatch Logs
- do not buffer stdout 
"""


import json


class SimpleLogger:

    """SimpleLogger for logging to CloudWatch Logs.
    Does not filter by LogLevel, but instead logs everything to CloudWatch.
    Logs can be filtered by level easily when querying in CloudWatch Logs
    
    >>> logger = SimpleLogger("MyLoggerName")
    >>> logger.info("my message", event={"test": 123}, another_field="foo")
    {
        "logger": "MyLoggerName",
        "level": "INFO",
        "message": "my message",
        "event": {
            "test": 123
        },
        "another_field": "foo"
    }
    """

    def __init__(self, name: str):
        self.name = name

    def _log(self, level, message: str, **kwargs):
        # No need to include a timestamp as CloudWatch Logs assigns one already
        self.fields = {
            "logger": self.name,
            "level": level,
            "message": message,
        }
        self.fields.update(kwargs)

        # Dump to JSON (convert non-JSON-able objects to str)
        # Replace line endings
        # Flush to stdout immediately (no need to set PYTHONUNBUFFERED)
        print(
            json.dumps(self.fields, indent=2, default=str).replace("\n", "\r"),
            flush=True,
        )

    def debug(self, message: str, **kwargs):
        self._log("DEBUG", message, **kwargs)

    def info(self, message: str, **kwargs):
        self._log("INFO", message, **kwargs)

    def warn(self, message: str, **kwargs):
        self._log("WARN", message, **kwargs)

    def error(self, message: str, **kwargs):
        self._log("ERROR", message, **kwargs)