Default output:
❯ python logger.py
2024-09-12 09:24:53 INFO user@test.local Hello worldOutput in a deployed env:
❯ ENVIRONMENT=dev python logger.py
{"level": "INFO", "date": "2024-09-12 09:25:04,645", "user_email": "user@test.local", "module": "logger", "lineno": 62, "message": "Hello world"}| 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:user@test.local" | |
| ) | |
| 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") |