Ghi nhật ký không cần phải lộn xộn. Thay vì rải rác các lệnh print() hoặc logging ở khắp mọi nơi, bạn có thể đưa vào một LoggingToolkitMixin để cung cấp cho bất kỳ lớp Python nào các nhật ký có cấu trúc, sẵn sàng cho JSON mà không cần nhiều nỗ lực.

Mixin Hoàn Chỉnh

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





# Ví dụ sử dụng

class PaymentService(LoggingToolkitMixin):

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

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

        # ... logic nghiệp vụ ...

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

Cách Sử Dụng

1. Cấu hình ghi nhật ký JSON một lần khi khởi động

configure_logging_json("DEBUG")

correlation_id.set("Glinteco-2025")

2. Mở rộng bất kỳ lớp nào bằng mixin

service = PaymentService()

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

3. Nhận các nhật ký JSON có cấu trúc như sau

{"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"}

Tại Sao Nó Hoạt Động

Bộ định dạng JSON → nhật ký có thể đọc được bằng máy cho các công cụ tìm kiếm & giám sát
ID Tương quan → theo dõi các yêu cầu trên các tác vụ không đồng bộ
Trình trang trí thời gian → đo thời gian thực hiện phương thức
Mixin có thể tái sử dụng → thả nó vào bất kỳ lớp nào mà không cần tạo khuôn mẫu