ログ記録は面倒なものではありません。print() またはログ記録呼び出しをあちこちに散布する代わりに、LoggingToolkitMixinをドロップして、ほぼ手間をかけずに、構造化され、JSON対応のログを任意のPythonクラスに提供できます。

完全なMixin

import logging, json, datetime, contextvars, time, functools



correlation_id = contextvars.ContextVar("correlation_id", default="-")





class JsonFormatter(logging.Formatter):

    def format(self, record):

        payload = {

            "ts": datetime.datetime.utcfromtimestamp(record.created).isoformat() + "Z",

            "level": record.levelname,

            "logger": record.name,

            "msg": record.getMessage(),

        }

        fields = getattr(record, "fields", None)

        if isinstance(fields, dict):

            payload.update(fields)

        return json.dumps(payload, ensure_ascii=False)





def configure_logging_json(level="INFO"):

    handler = logging.StreamHandler()

    handler.setFormatter(JsonFormatter())

    root = logging.getLogger()

    root.handlers[:] = [handler]

    root.setLevel(getattr(logging, level.upper(), logging.INFO))





class LoggingMixin:

    @property

    def logger(self):

        return logging.getLogger(f"{self.__class__.__module__}.{self.__class__.__name__}")





class LoggingToolkitMixin(LoggingMixin):

    def log(self, level: int, msg: str, **fields):

        fields.setdefault("correlation_id", correlation_id.get())

        self.logger.log(level, msg, extra={"fields": fields})



    def debug(self, msg: str, **f): self.log(logging.DEBUG, msg, **f)

    def info(self, msg: str,  **f): self.log(logging.INFO,  msg, **f)

    def warn(self, msg: str,  **f): self.log(logging.WARNING, msg, **f)

    def error(self, msg: str, **f): self.log(logging.ERROR, msg, **f)



    def is_debug(self) -> bool:

        return self.logger.isEnabledFor(logging.DEBUG)



    def timed(self, name=None):

        def deco(func):

            label = name or func.__name__

            @functools.wraps(func)

            def wrapper(*args, **kwargs):

                t0 = time.perf_counter()

                try:

                    return func(*args, **kwargs)

                finally:

                    self.info("timed", op=label, ms=round((time.perf_counter()-t0)*1000, 2))

            return wrapper

        return deco





# 使用例

class PaymentService(LoggingToolkitMixin):

    def charge(self, order_id: str, amount: float):

        self.info("charge.start", order_id=order_id, amount=amount)

        # ... ビジネスロジック ...

        self.info("charge.success", order_id=order_id)

使用方法

1. 起動時に一度JSONログ記録を設定します

configure_logging_json("DEBUG")

correlation_id.set("Glinteco-2025")

2. 任意のクラスをmixinで拡張します

service = PaymentService()

service.charge("order-42", 19.99)

3. このような構造化されたJSONログを取得します

{"ts": "2025-09-07T14:30:01Z", "level": "INFO", "logger": "__main__.PaymentService", "msg": "charge.start", "order_id": "order-42", "amount": 19.99, "correlation_id": "glinteco-2025"}

{"ts": "2025-09-07T14:30:01Z", "level": "INFO", "logger": "__main__.PaymentService", "msg": "charge.success", "order_id": "order-42", "correlation_id": "glinteco-2025"}