一套好的日志方案不应该把输出位置写死。下面这套 Python 标准库方案遵循一个简单模型：
- 默认只输出到控制台，开箱可用，适合脚本、开发环境和容器；
- 通过环境变量或应用设置，可同时开启控制台与文件；
- 后端可扩展到 Elasticsearch、Loki、CloudWatch、Kafka 等集中日志系统；
- 所有输出保持统一的时间、等级、模块名、Trace ID 和消息格式。
它适用于命令行工具、后台任务、Web 服务和 Agent 项目。示例使用 Python 3.10+ 类型标注，日志核心仅依赖标准库 logging。
不要记录 API Key、Cookie、密码、私钥、完整请求头或未经脱敏的隐私数据。运行日志必须加入
.gitignore，不要提交到版本库。
0. 最快接入：一个通用模块
创建 logging_setup.py。不要命名为 logging.py，以免遮蔽 Python 标准库：
your-project/
├── logging_setup.py # 复制下文代码
├── main.py
├── logging.yaml # 可选：部署级配置
└── logs/ # 运行时生成，加入 .gitignore
若将模块放进包内，例如 my_app/logging_setup.py，后文 YAML 中的类路径要改为 my_app.logging_setup.TraceIdFilter。
1. 通用日志模块：默认控制台，可按设置开启文件
把以下代码复制到 logging_setup.py。默认参数为 enable_console=True、enable_file=False，因此没有任何设置时只向控制台输出。将两个开关同时设为 True，即可同时写控制台和文件。
"""可复用的日志初始化、输出目标和 Trace ID 上下文。"""
import json
import logging
import logging.config
from collections.abc import Mapping
from contextvars import ContextVar, Token
from datetime import datetime
from pathlib import Path
from typing import Any
from uuid import uuid4
from zoneinfo import ZoneInfo
trace_id: ContextVar[str] = ContextVar("trace_id", default="-")
class TraceIdFilter(logging.Filter):
"""把当前 Trace ID 注入每条 LogRecord，供 formatter 使用。"""
def filter(self, record: logging.LogRecord) -> bool:
record.trace_id = trace_id.get()
return True
class TimezoneFormatter(logging.Formatter):
"""以指定 IANA 时区输出人类可读日志。"""
def __init__(
self,
format: str | None = None,
datefmt: str | None = None,
style: str = "%",
timezone_name: str = "UTC",
) -> None:
super().__init__(fmt=format, datefmt=datefmt, style=style)
self._timezone = ZoneInfo(timezone_name)
def formatTime(self, record: logging.LogRecord, datefmt: str | None = None) -> str:
value = datetime.fromtimestamp(record.created, tz=self._timezone)
return value.strftime(datefmt) if datefmt else value.isoformat(timespec="milliseconds")
class JsonFormatter(logging.Formatter):
"""输出一行一个 JSON 对象（NDJSON），适合日志采集器。"""
def __init__(self, timezone_name: str = "UTC") -> None:
super().__init__()
self._timezone = ZoneInfo(timezone_name)
def format(self, record: logging.LogRecord) -> str:
payload: dict[str, Any] = {
"timestamp": datetime.fromtimestamp(
record.created, tz=self._timezone
).isoformat(timespec="milliseconds"),
"level": record.levelname,
"logger": record.name,
"trace_id": getattr(record, "trace_id", "-"),
"message": record.getMessage(),
}
if record.exc_info:
exc_type, exc_value, _ = record.exc_info
payload["exception"] = {
"type": exc_type.__name__ if exc_type else "Exception",
"message": str(exc_value),
"stacktrace": self.formatException(record.exc_info),
}
return json.dumps(payload, ensure_ascii=False, default=str)
def new_trace_id() -> str:
return uuid4().hex
def set_trace_id(value: str) -> Token[str]:
return trace_id.set(value)
def reset_trace_id(token: Token[str]) -> None:
trace_id.reset(token)
def build_logging_config(
*,
level: str = "INFO",
log_format: str = "text",
timezone_name: str = "UTC",
enable_console: bool = True,
enable_file: bool = False,
file_path: str = "logs/app.log",
max_bytes: int = 10 * 1024 * 1024,
backup_count: int = 5,
) -> dict[str, Any]:
"""构建标准 dictConfig；至少保留控制台，避免无日志输出。"""
formatter_name = "json" if log_format.lower() == "json" else "standard"
handlers: dict[str, dict[str, Any]] = {}
if enable_console:
handlers["console"] = {
"class": "logging.StreamHandler",
"level": level,
"formatter": formatter_name,
"filters": ["trace_id"],
"stream": "ext://sys.stdout",
}
if enable_file:
handlers["file"] = {
"class": "logging.handlers.RotatingFileHandler",
"level": level,
"formatter": formatter_name,
"filters": ["trace_id"],
"filename": file_path,
"maxBytes": max_bytes,
"backupCount": backup_count,
"encoding": "utf-8",
}
if not handlers:
handlers["console"] = {
"class": "logging.StreamHandler",
"level": level,
"formatter": formatter_name,
"filters": ["trace_id"],
"stream": "ext://sys.stdout",
}
return {
"version": 1,
"disable_existing_loggers": False,
"filters": {"trace_id": {"()": TraceIdFilter}},
"formatters": {
"standard": {
"()": TimezoneFormatter,
"format": (
"%(asctime)s | %(levelname)-8s | %(name)s | "
"trace_id=%(trace_id)s | %(message)s"
),
"datefmt": "%Y-%m-%d %H:%M:%S",
"timezone_name": timezone_name,
},
"json": {"()": JsonFormatter, "timezone_name": timezone_name},
},
"handlers": handlers,
"root": {"level": level, "handlers": list(handlers)},
}
DEFAULT_LOGGING_CONFIG = build_logging_config()
def configure_logging(config: Mapping[str, Any] | None = None) -> None:
"""应用外部 dictConfig；不传配置时使用默认控制台输出。"""
logging_config = dict(config) if config else DEFAULT_LOGGING_CONFIG
_create_log_directories(logging_config)
logging.config.dictConfig(logging_config)
def _create_log_directories(config: Mapping[str, Any]) -> None:
"""为含 filename 的文件 handler 自动创建父目录。"""
handlers = config.get("handlers", {})
if not isinstance(handlers, Mapping):
return
for handler in handlers.values():
if not isinstance(handler, Mapping):
continue
filename = handler.get("filename")
if isinstance(filename, str) and filename:
Path(filename).expanduser().parent.mkdir(parents=True, exist_ok=True)
日志默认格式：
2026-07-20 16:00:00 | INFO | my_app.service | trace_id=4a8c... | 任务已完成
| 字段 | 格式占位符 | 用途 |
|---|---|---|
| 时间 | %(asctime)s | 事件发生时间；datefmt 控制为 年-月-日 时:分:秒。 |
| 等级 | %(levelname)-8s | DEBUG、INFO、WARNING、ERROR、CRITICAL；-8s 用于对齐。 |
| 模块 | %(name)s | logging.getLogger(__name__) 自动给出的模块路径。 |
| Trace ID | %(trace_id)s | 同一次请求、消息或后台任务的关联标识。 |
| 消息 | %(message)s | 业务代码写入的日志消息。 |
timezone_name 默认是 UTC，文本和 JSON formatter 都使用同一个时区。通过 LOG_TIMEZONE 或 build_logging_config(timezone_name="Asia/Shanghai") 传入 IANA 时区名即可调整，例如 Asia/Shanghai、America/Los_Angeles。需要确保部署系统提供 IANA 时区数据库；部分精简镜像或 Windows 环境可能需要额外提供 tzdata。
JSON 格式：NDJSON 与 UTC/可配置时区
设置 LOG_FORMAT=json 后，每条日志输出为一行 JSON（NDJSON），控制台、文件、Fluent Bit、Filebeat 和 OpenTelemetry Collector 都能逐行解析：
{"timestamp":"2026-07-20T16:00:00.123+00:00","level":"INFO","logger":"my_app.service","trace_id":"4a8c...","message":"任务已完成"}
固定字段为 timestamp、level、logger、trace_id 和 message；在 logger.exception(...) 或带有异常信息的日志中，还会增加 exception.type、exception.message、exception.stacktrace。JSON 的 timestamp 使用 timezone_name：默认 UTC 为 +00:00，设置 LOG_TIMEZONE=Asia/Shanghai 后会输出 +08:00。json.dumps() 会正确转义换行、引号和非 ASCII 字符，因此一条日志始终保持一行。
2. 环境变量或应用设置：选择输出目标
入口只在初始化阶段配置日志。下面示例以环境变量作为最通用的设置来源；也可以替换成 Django、FastAPI、Pydantic Settings、配置中心或任意项目自己的设置对象。
# main.py
import logging
import os
from logging_setup import (
build_logging_config,
new_trace_id,
reset_trace_id,
set_trace_id,
configure_logging,
)
logger = logging.getLogger(__name__)
def env_flag(name: str, default: bool) -> bool:
value = os.getenv(name)
if value is None:
return default
return value.strip().lower() in {"1", "true", "yes", "on"}
def configure_application_logging() -> None:
environment = os.getenv("APP_ENV", "development")
default_level = "DEBUG" if environment == "development" else "INFO"
config = build_logging_config(
level=os.getenv("LOG_LEVEL", default_level),
log_format=os.getenv("LOG_FORMAT", "text"),
timezone_name=os.getenv("LOG_TIMEZONE", "UTC"),
enable_console=env_flag("LOG_CONSOLE", True),
enable_file=env_flag("LOG_FILE", False),
file_path=os.getenv("LOG_FILE_PATH", "logs/app.log"),
)
configure_logging(config)
def main() -> None:
token = set_trace_id(new_trace_id())
try:
configure_application_logging()
logger.info("应用启动")
# 执行业务逻辑
except Exception:
logger.exception("应用运行失败")
raise
finally:
reset_trace_id(token)
if __name__ == "__main__":
main()
建议的环境设置如下：
| 场景 | APP_ENV | LOG_CONSOLE | LOG_FILE | LOG_FORMAT | LOG_TIMEZONE | 输出结果 |
|---|---|---|---|---|---|---|
| 未设置任何变量 | development | true | false | text | UTC | 默认仅控制台。 |
| 本地开发 | development | true | false | text | Asia/Shanghai | 控制台 + DEBUG 级别。 |
| 传统服务器 | production | true | true | text 或 json | UTC | 同时控制台和滚动文件。 |
| Docker/Kubernetes | production | true | false | json | UTC | NDJSON 写 stdout，由平台采集。 |
如果 LOG_CONSOLE=false 且 LOG_FILE=false，通用模块会回退到控制台，防止应用完全失去日志。要实现“只写文件”，请设置 LOG_CONSOLE=false 且 LOG_FILE=true；但生产环境通常不建议隐藏控制台日志。
3. YAML 配置：显式组合 console、file 和自定义后端
使用 YAML 的项目可以直接把 logging 字典交给 configure_logging()。root.handlers 就是输出目标的总开关：
logging:
version: 1
disable_existing_loggers: false
filters:
trace_id:
(): logging_setup.TraceIdFilter
formatters:
standard:
(): logging_setup.TimezoneFormatter
format: "%(asctime)s | %(levelname)-8s | %(name)s | trace_id=%(trace_id)s | %(message)s"
datefmt: "%Y-%m-%d %H:%M:%S"
timezone_name: Asia/Shanghai
json:
(): logging_setup.JsonFormatter
timezone_name: UTC
handlers:
console:
class: logging.StreamHandler
level: INFO
formatter: standard
filters: [trace_id]
stream: ext://sys.stdout
file:
class: logging.handlers.RotatingFileHandler
level: INFO
formatter: standard
filters: [trace_id]
filename: logs/app.log
maxBytes: 10485760
backupCount: 5
encoding: utf-8
root:
level: INFO
handlers: [console, file]
要让某个目标输出 JSON，只需把该 handler 的 formatter 改为 json；例如容器环境通常将 console.formatter 改为 json，传统服务器可让 console 保持 standard、file 使用 json，以同时兼顾本地阅读和采集。timezone_name 可用任意有效 IANA 时区名，建议集中设置为 UTC，不要让不同 handler 混用时区。
选择不同的 root.handlers，就得到不同组合：
# 开发、容器或默认模式：只输出控制台
handlers: [console]
# 传统服务器：同时输出控制台与文件
handlers: [console, file]
# 使用自定义集中日志 handler 后：控制台与集中日志
handlers: [console, elasticsearch]
YAML 的 handlers 只是声明能力，只有出现在 root.handlers 或某个具名 logger 的 handlers 列表中才会真正生效。每个使用 %(trace_id)s 的 handler 都必须包含 filters: [trace_id]。
若 YAML 加载可能失败，可先调用 configure_logging() 应用默认控制台配置，再加载 YAML 并二次调用 configure_logging(config["logging"])。这样解析失败也有可读的诊断日志；业务模块仍不应自行配置日志。
将运行日志目录加入 Git 忽略：
logs/
4. 文件日志切分：大小、时间或平台托管
同一个文件 handler 只选择一种切分策略。
策略 A：按文件大小切分
RotatingFileHandler 适合日志量有波动、重点是限制磁盘占用的传统服务器：
file:
class: logging.handlers.RotatingFileHandler
filename: logs/app.log
maxBytes: 10485760 # 10 MiB
backupCount: 5
encoding: utf-8
当前文件达到 10 MiB 后会变为 app.log.1，更早文件依次递增；超过 5 个备份时自动删除最旧文件。磁盘占用大致不超过 60 MiB（当前文件加 5 个备份，实际略有波动）。
策略 B：按时间切分
TimedRotatingFileHandler 适合按天归档、按日期查询的服务。用下面的 file 配置替换策略 A：
file:
class: logging.handlers.TimedRotatingFileHandler
filename: logs/app.log
when: midnight
interval: 1
backupCount: 14
utc: true
encoding: utf-8
它会在每天 UTC 零点后的下一次日志写入时切分，通常生成 app.log.2026-07-20，并保留 14 天。when 也可用 H（小时）和 D（天）。
策略 C：平台托管
Docker、Kubernetes、ELK、Loki 和云日志平台通常只保留 console，由平台完成采集、轮转、索引和保留：
root:
level: INFO
handlers: [console]
多进程服务不要让多个进程无协调地写同一个本地文件；此时优先选择平台托管。
5. 扩展到 Elasticsearch 等集中日志后端
首选：应用 stdout → 采集器 → Elasticsearch
生产环境最稳妥的架构通常是：
Python 应用 → stdout → Fluent Bit / Filebeat / OpenTelemetry Collector → Elasticsearch → Kibana
应用只保留 console handler，采集器负责批量发送、重试、缓存、断网续传、索引生命周期和权限管理。这种方式不会因为 Elasticsearch 短暂不可用而阻塞业务线程，也是容器环境的首选。
日志平台要求 JSON 时，设置 LOG_FORMAT=json 或将 YAML handler 的 formatter 设为 json；采集器即可逐行解析。字段已包含时间、级别、logger 名称、Trace ID、消息和异常堆栈，适合建立 Elasticsearch 索引和 Kibana 查询。
备选：应用内自定义 Elasticsearch handler
标准库没有 Elasticsearch handler，但 dictConfig 支持加载自定义或第三方 handler。实现后可如下声明：
handlers:
elasticsearch:
class: my_app.logging_handlers.ElasticsearchHandler
level: INFO
formatter: standard
filters: [trace_id]
hosts: ["https://es.example.internal:9200"]
index: app-logs
root:
level: INFO
handlers: [console, elasticsearch]
ElasticsearchHandler 应由项目自己实现或选用经过维护的库，并至少满足以下条件：
- 业务线程只写入有界队列，由后台线程/进程批量写 ES；不能同步等待网络请求；
- 设置连接和请求超时、指数退避与最大重试次数；
- 队列满、ES 不可用或认证失败时，有明确策略：丢弃低级别日志、写本地降级文件或发送监控告警；
- 使用批量写入，并为索引设置生命周期策略，避免无限增长；
- 不在日志 handler 内再次记录同一个 logger 的错误，避免递归日志风暴；
- URL、用户名和令牌通过环境变量或密钥管理服务提供，绝不写入 YAML 或日志。
相同的扩展模式也适用于 Kafka、CloudWatch、Sentry、Loki 等后端：将目标实现为独立 handler，并保留 console 作为可观测性和故障降级出口。
6. Trace ID：请求、异步任务与下游调用
ContextVar 会随 asyncio 任务上下文传播，因此适合异步 Web 服务和 Agent 工作流。请求入口优先复用上游的 Trace ID；没有时生成新的，并在结束时复位：
async def handle_request(request):
incoming_trace_id = request.headers.get("X-Trace-ID")
current_trace_id = incoming_trace_id or new_trace_id()
token = set_trace_id(current_trace_id)
try:
logger.info("收到请求")
response = await dispatch(request)
response.headers["X-Trace-ID"] = current_trace_id
return response
except Exception:
logger.exception("请求处理失败")
raise
finally:
reset_trace_id(token)
不可信公网请求应限制 Trace ID 的长度和字符集，或始终由网关生成。向下游服务发请求时，把当前 Trace ID 放在约定请求头中，即可关联跨服务调用。
7. 业务模块固定写法与上线检查
所有业务模块只需要：
import logging
logger = logging.getLogger(__name__)
记录变量时使用参数化日志，异常块中使用 logger.exception()：
logger.info("订单处理完成: order_id=%s", order_id)
try:
process_order(order_id)
except OSError:
logger.exception("订单处理失败: order_id=%s", order_id)
raise
不要在业务模块调用 basicConfig()、dictConfig() 或自行添加 handler；这会造成重复输出、格式不一致，甚至影响其他库的日志。
上线前检查：
- 默认配置能在没有环境变量时只输出控制台；
- 传统服务器明确启用
LOG_FILE=true或[console, file]； - 每个使用
%(trace_id)s的 handler 都配置了TraceIdFilter； - 日志目录有写权限，且
logs/已加入.gitignore； - 文件轮转只选择大小或时间其中一种；
- 容器环境优先 stdout + 采集器，不直接同步写 Elasticsearch；
- 日志中不包含凭据、令牌、密码、Cookie、私钥和未经脱敏的用户数据。
复制 logging_setup.py 后，默认行为就是控制台输出；通过环境变量、YAML 的 handler 组合或自定义 handler，可以逐步扩展为控制台 + 文件、平台采集或 Elasticsearch 等集中日志后端。