日志别再print了！深入对比Python三大日志方案¶

import logging

# 创建 logger
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# 创建文件 handler
handler = logging.FileHandler("app.log")
formatter = logging.Formatter(
    "%(asctime)s | %(levelname)s | %(name)s | %(message)s"
)
handler.setFormatter(formatter)
logger.addHandler(handler)

# 使用
def process_order(order_id):
    logger.info("Processing order %s", order_id)
    try:
        do_work(order_id)
    except Exception:
        logger.exception("Order failed")

from loguru import logger
logger.info("Service started")

logger.add(
    "logs/app.log",
    rotation="10 MB",     # 超过10MB自动轮转
    retention="7 days",   # 保留7天
    level="INFO",
    format="{time} | {level} | {message}"
)

try:
    risky_call()
except Exception:
    logger.exception("Risky call failed")  # 自动捕获完整堆栈

# 绑定 request_id
request_logger = logger.bind(request_id="req_123")
request_logger.info("Incoming request")  # 自动包含 request_id

from fastapi import FastAPI, Request
from loguru import logger
import uuid
app = FastAPI()
@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request.state.logger = logger.bind(request_id=request_id)
    response = await call_next(request)
    return response
@app.get("/health")
async def health(request: Request):
    request.state.logger.info("Health check")  # 自动带 request_id
    return {"status": "ok"}

import asyncio
from loguru import logger
async def worker(name):
    logger.info("Worker started {}", name)
    await asyncio.sleep(1)
    logger.info("Worker finished {}", name)
async def main():
    await asyncio.gather(worker("A"), worker("B"))
asyncio.run(main())

logger.add(
    "logs/app.log",
    enqueue=True,     # 日志入队，后台线程写入
    rotation="50 MB",
    level="INFO"
)

logger.add(
    "logs/app.json",
    serialize=True,   # 输出 JSON 格式
    level="INFO"
)
logger.info("User login", user_id=42, source="web")

# 输出: {"time": "2026-03-21T10:00:00", "level": "INFO", "message": "User login", "user_id": 42, "source": "web"}

from loguru import logger
import sys
def setup_logging():

    # 移除默认的 stderr 输出
    logger.remove()

    # 控制台输出（开发环境）
    logger.add(
        sys.stdout,
        level="INFO",
        format="{time} | {level} | {message}"
    )

    # 文件输出（生产环境）
    logger.add(
        "logs/app.log",
        level="INFO",
        rotation="100 MB",
        retention="10 days",
        enqueue=True,
        compression="zip"    # 压缩旧日志
    )

from loguru import logger
logger.add("sync.log", rotation="30 days")
def main():
    logger.info("Sync started")
    try:
        do_sync()
    except Exception as e:
        logger.exception("Sync failed")
        raise
    logger.info("Sync completed")
if __name__ == "__main__":
    main()

logger.add(
    "api.log",
    enqueue=True,
    serialize=True,
    rotation="500 MB",
    retention="7 days"
)

# 中间件绑定 request_id
@app.middleware("http")
async def log_request(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    with logger.contextualize(request_id=request_id):
        response = await call_next(request)
        return response

# 之前
import logging
logger = logging.getLogger(__name__)
logger.info("Processing %s", order_id)

# 之后
from loguru import logger
logger.info("Processing {}", order_id)  # 注意格式变化

import logging
from loguru import logger
class InterceptHandler(logging.Handler):
    def emit(self, record):
        level = logger.level(record.levelname).name
        logger.log(level, record.getMessage())

# 全局拦截
logging.basicConfig(handlers=[InterceptHandler()], level=0)

logger.remove()  # 必须先调用，否则会输出两份

def mask_secrets(record):
    if "password" in record["message"]:
        record["message"] = record["message"].replace(record["extra"].get("password", ""), "***")
    return True
logger.add("app.log", filter=mask_secrets)