File size: 1,450 Bytes
1e23d14 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | import logging
import sys
from typing import Optional
def setup(
name: str = "gcp-tool",
level: str = "INFO",
log_file: Optional[str] = None,
fmt: str = "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
) -> logging.Logger:
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper(), logging.INFO))
logger.handlers.clear()
formatter = logging.Formatter(fmt, datefmt="%Y-%m-%dT%H:%M:%S")
stream_handler = logging.StreamHandler(sys.stdout)
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)
if log_file:
file_handler = logging.FileHandler(log_file)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.propagate = False
return logger
class StructuredLogger:
def __init__(self, name: str) -> None:
self._log = logging.getLogger(name)
def _fmt(self, msg: str, **kw) -> str:
if not kw:
return msg
pairs = " ".join(f"{k}={v!r}" for k, v in kw.items())
return f"{msg} {pairs}"
def info(self, msg: str, **kw) -> None:
self._log.info(self._fmt(msg, **kw))
def warning(self, msg: str, **kw) -> None:
self._log.warning(self._fmt(msg, **kw))
def error(self, msg: str, **kw) -> None:
self._log.error(self._fmt(msg, **kw))
def debug(self, msg: str, **kw) -> None:
self._log.debug(self._fmt(msg, **kw))
|