Demo of using formatting python logger for json output

demo.md

Default output:

❯ python logger.py 
2024-09-12 09:24:53 INFO [email protected] Hello world

Output in a deployed env:

❯ ENVIRONMENT=dev python logger.py 
{"level": "INFO", "date": "2024-09-12 09:25:04,645", "user_email": "[email protected]", "module": "logger", "lineno": 62, "message": "Hello world"}

logger.py

import os
import logging
import json

LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
ENV = os.getenv("ENVIRONMENT", "local")
DATE_FORMAT = "%Y-%m-%d %H:%M:%S"


def get_user_email() -> str:
    # In cloud run, this is present in the header (not environment variable)
    raw_str = os.getenv(
        "x-goog-authenticated-user-email", "accounts.google.com:[email protected]"
    )
    return raw_str.split(":")[-1]


# This filter adds the user email to all log records
class UserEmailFilter(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        record.user_email = get_user_email()
        return True


class JsonFormatter(logging.Formatter):
    """Formatter to dump error message into JSON"""

    def format(self, record: logging.LogRecord) -> str:
        record_dict = {
            "level": record.levelname,
            "date": self.formatTime(record),
            "user_email": record.user_email,  # type: ignore[attr-defined]
            "module": record.module,
            "lineno": record.lineno,
            "message": record.getMessage(),
        }
        return json.dumps(record_dict)


logger = logging.getLogger(__name__)
logger.setLevel(LOG_LEVEL)

# Remove existing handlers
for handler in logger.handlers[:]:
    logger.removeHandler(handler)

stream_h = logging.StreamHandler()
if ENV.lower() != "local":
    # Format logs in json in deployments
    stream_h.setFormatter(JsonFormatter(datefmt=DATE_FORMAT))
else:
    # Otherwise use a standard format for stdout
    stream_h.setFormatter(
        logging.Formatter(
            fmt="%(asctime)s %(levelname)s %(user_email)-8s %(message)s",
            datefmt=DATE_FORMAT,
        )
    )
logger.addHandler(stream_h)
logger.addFilter(UserEmailFilter())


# Sample usage
logger.info("Hello world")