Loguru實作指南：打造穩健、結構化、並行且可上線的Python日誌管道

In this tutorial, we implement a practical use case with Loguru, a powerful, flexible, and production-ready logging library for Python. We start by building a clean, idempotent logging setup that can be safely rerun without duplicating handlers or producing messy output. From there, we move step by step through structured logging, contextual logging, custom log levels, global patching, callable formatters, and in-memory sinks. We also handle real-world logging needs such as rich exception traces, JSON log files, custom rotation, compression, retention, async logging, threaded execution, multiprocessing-safe logging, and standard logging module interception. By keeping everything in a Colab-ready workflow, we make it easy to test, inspect, and understand how Loguru can support debugging, monitoring, and observability in serious Python applications. Copy CodeCopiedUse a different Browser!pip install -q loguru nest_asyncio import os, sys, time, json, glob, gzip, shutil, asyncio, logging, itertools, multiprocessing from collections import deque from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor from loguru import logger try: import nest_asyncio; nest_asyncio.apply() except Exception as e: print("nest_asyncio not applied:", e) WORKDIR = "/content/loguru_demo" if os.path.isdir("/content") else "/tmp/loguru_demo" os.makedirs(WORKDIR, exist_ok=True); os.chdir(WORKDIR) for f in glob.glob("*"): try: os.remove(f) except OSError: pass print(f"Working directory: {WORKDIR}\n") RESULTS = [] def check(name, condition, detail=""): ok = bool(condition); RESULTS.append((name, ok)) print(f" [{'PASS' if ok else 'FAIL'}] {name}" + (f" — {detail}" if detail else "")) def banner(t): print(f"\n{'='*64}\n {t}\n{'='*64}") _seq = itertools.count(1) def global_patcher(record): record["extra"].setdefault("env", "colab") record["extra"]["seq"] = next(_seq) _NOISE = {"env", "seq", "app"} def console_formatter(record): fmt = ("<green>{time:HH:mm:ss.SSS}</green> | <level>{level: <8}</level> | " "<cyan>{name}:{function}:{line}</cyan> - <level>{message}</level>") if any(k not in _NOISE for k in record["extra"]): fmt += " | <yellow>{extra}</yellow>" return fmt + "\n{exception}" We install Loguru and supporting dependencies, import all required libraries, and prepare a clean working directory for the tutorial. We also create a small verification helper to test each feature as the tutorial runs. We then define a global patcher and console formatter so that every log record carries useful metadata and appears in a readable format. Copy CodeCopiedUse a different Browserclass MemorySink: def __init__(self, capacity=2000): self.buffer = deque(maxlen=capacity) def write(self, message): self.buffer.append(message.record) def flush(self): pass def has_level(self, name): return any(r["level"].name == name for r in self.buffer) def find(self, pred): return [r for r in self.buffer if pred(r)] MAX_BYTES = 1500 def size_rotation(message, file): return file.tell() + len(message) > MAX_BYTES def gzip_compression(filepath): with open(filepath, "rb") as fi, gzip.open(filepath + ".gz", "wb") as fo: shutil.copyfileobj(fi, fo) os.remove(filepath) def keep_latest_retention(files): for old in sorted(files, key=os.path.getmtime, reverse=True)[3:]: try: os.remove(old) except OSError: pass class InterceptHandler(logging.Handler): def emit(self, record): try: level = logger.level(record.levelname).name except ValueError: level = record.levelno frame, depth = logging.currentframe(), 2 while frame and frame.f_code.co_filename == logging.__file__: frame, depth = frame.f_back, depth + 1 (logger.opt(depth=depth, exception=record.exc_info) .bind(stdlib_logger=record.name) .log(level, record.getMessage())) def mp_worker(n): logger.bind(child=os.getpid()).info("hello from child item {}", n) return os.getpid() We create reusable logging components that make the tutorial more practical and production-like. We define an in-memory sink, custom file rotation, compression, and retention functions to control how logs are stored. We also built a standard logging interceptor and a multiprocessing worker to connect Loguru to external libraries and child processes. Copy CodeCopiedUse a different Browserbanner("1) logger.configure(): handlers + custom level + extra + patcher") mem = MemorySink() logger.configure( handlers=[ {"sink": sys.stderr, "format": console_formatter, "level": "DEBUG", "colorize": True, "backtrace": True, "diagnose": True}, {"sink": mem, "level": "DEBUG", "format": "{message}"}, {"sink": "structured.jsonl", "serialize": True, "level": "DEBUG", "enqueue": True}, {"sink": "errors.log", "level": "ERROR", "enqueue": True, "backtrace": True, "diagnose": False, "format": "{time:YYYY-MM-DD HH:mm:ss} | {level} | " "{name}:{function}:{line} | {message}"}, ], levels=[{"name": "NOTICE", "no": 22, "color": "<blue><bold>", "icon": ""}], extra={"app": "loguru-advanced"}, patcher=global_patcher, ) logger.debug("debug"); logger.info("info"); logger.success("SUCCESS level ships built-in") logger.warning("warning"); logger.log("NOTICE", "custom level between INFO and SUCCESS") banner("2) bind() / contextualize() / patch()") logger.bind(user_id=42, request_id="abc-123").info("bound context") with logger.contextualize(task="batch-job", run=7): logger.info("inside contextualized block") logger.patch(lambda r: r["extra"].update(epoch=round(time.time()))).info("per-call patched record") banner("3) @logger.catch + context-manager form") def inner(d): return d["a"] / d["b"] def outer(d): return inner(d) @logger.catch(reraise=False) def compute(d): return outer(d) compute({"a": 1, "b": 0}) with logger.catch(message="handled inside a with-block"): raise ValueError("boom in block") banner("4) opt(lazy=True), inline colors, record access") logger.opt(lazy=True).debug("lazy sum = {}", lambda: sum(i*i for i in range(1_000_000))) logger.opt(colors=True).info("inline <red>colors</red> <green>work</green>") logger.opt(record=True).info("emitted from source line {record[line]}") We configure Loguru with multiple handlers, including console output, memory capture, JSON logging, and error logging. We then demonstrate structured logging with bound context, contextual blocks, patched records, and a custom log level. We also explore exception handling and useful opt() features such as lazy evaluation, inline colors, and record access. Copy CodeCopiedUse a different Browserbanner("5) custom rotation/compression/retention (forces real rotation)") ev_id = logger.add("events_{time:HHmmss_SSS}.log", rotation=size_rotation, compression=gzip_compression, retention=keep_latest_retention, enqueue=True, level="DEBUG", format="{time:HH:mm:ss.SSS} | {level: <8} | {message}") for i in range(80): logger.bind(idx=i).debug("rotating event line number {}", i) logger.complete(); logger.remove(ev_id) print(f" archives created: {sorted(glob.glob('events_*.gz'))}") banner("6a) ThreadPoolExecutor with per-thread contextualize()") thread_caps = [] tid = logger.add(thread_caps.append, level="DEBUG", format="{message}", filter=lambda r: "worker_id" in r["extra"]) def worker(n): with logger.contextualize(worker_id=n): logger.info("thread work item {}", n) return n * n with ThreadPoolExecutor(max_workers=8) as ex: sq = list(ex.map(worker, range(8))) logger.complete(); logger.remove(tid) worker_ids = {m.record["extra"]["worker_id"] for m in thread_caps} banner("6b) async coroutine sink + await logger.complete()") async def run_async_demo(): sunk = [] async def async_sink(message): await asyncio.sleep(0); sunk.append(message.record["message"]) sid = logger.add(async_sink, level="DEBUG", catch=True) async def task(n): with logger.contextualize(coro=n): logger.info("async task {} start", n) await asyncio.sleep(0.01) logger.success("async task {} done", n) await asyncio.gather(*(task(i) for i in range(5))) await logger.complete() logger.remove(sid) return sunk try: async_msgs = asyncio.run(run_async_demo()) except RuntimeError: async_msgs = asyncio.get_event_l