taolib.testing.logging_config

taolib.testing.logging_config#

日志配置模块。

提供统一的日志配置功能，支持控制台、文件输出和远程日志平台。

Attributes#

`_LOG_FORMAT`
`_DATE_FORMAT`

Classes#

`JSONFormatter`	JSON 行格式日志格式化器。
`SensitiveDataFilter`	日志敏感数据脱敏过滤器。
`RemoteLogHandler`	远程日志平台 Handler。

Functions#

`configure_logging`(→ SensitiveDataFilter \| None)	配置日志系统。
`get_logger`(→ logging.Logger)	获取指定名称的日志记录器。
`configure_remote_logging`(→ RemoteLogHandler)	配置同时输出到本地和远程日志平台。

Module Contents#

taolib.testing.logging_config._LOG_FORMAT = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'#

taolib.testing.logging_config._DATE_FORMAT = '%Y-%m-%d %H:%M:%S'#

class taolib.testing.logging_config.JSONFormatter(service: str = '', datefmt: str | None = None)#

Bases: logging.Formatter

JSON 行格式日志格式化器。

输出每行一个 JSON 对象，便于 ELK/Loki 等日志聚合系统解析。

参数:

service -- 服务名称，写入每条日志的 service 字段。
datefmt -- 时间格式字符串，默认为 ISO 8601。

service = ''#

format(record: logging.LogRecord) → str#

Format the specified record as text.

The record's attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

class taolib.testing.logging_config.SensitiveDataFilter(enable_password: bool = True, enable_jwt: bool = True, enable_api_key: bool = True, enable_email: bool = True, enable_phone: bool = True, enable_ip: bool = True, custom_rules: list[tuple[str, str]] | None = None)#

Bases: logging.Filter

日志敏感数据脱敏过滤器。

对日志消息中的敏感数据进行脱敏处理，支持密码、JWT密钥、API密钥、邮箱、手机号和IP地址等多种敏感数据类型。

参数:

enable_password -- 是否启用密码脱敏，默认为 True。
enable_jwt -- 是否启用 JWT 密钥脱敏，默认为 True。
enable_api_key -- 是否启用 API 密钥脱敏，默认为 True。
enable_email -- 是否启用邮箱脱敏，默认为 True。
enable_phone -- 是否启用手机号脱敏，默认为 True。
enable_ip -- 是否启用 IP 地址脱敏，默认为 True。
custom_rules -- 自定义脱敏规则列表，每条规则为 (pattern, replacement) 元组。

示例

>>> filter = SensitiveDataFilter()
>>> logger = logging.getLogger("test")
>>> logger.addFilter(filter)
>>> logger.info("password=secret123")  # 输出: password=***

enable_password = True#

enable_jwt = True#

enable_api_key = True#

enable_email = True#

enable_phone = True#

enable_ip = True#

custom_rules = []#

_patterns: list[tuple[re.Pattern, collections.abc.Callable[[re.Match], str]]] = []#

_compile_patterns() → None#: 编译所有启用的脱敏模式。

_mask_api_key(match: re.Match) → str#

对 API 密钥进行脱敏处理。

参数:: match -- 正则匹配对象。
返回:: 脱敏后的字符串，保留前4位和后4位。

_mask_email(match: re.Match) → str#

对邮箱地址进行脱敏处理。

参数:: match -- 正则匹配对象。
返回:: 脱敏后的字符串，保留首字符和域名。

filter(record: logging.LogRecord) → bool#

过滤日志记录，对敏感数据进行脱敏。

参数:: record -- 日志记录对象。
返回:: 始终返回 True，允许所有日志记录通过。

_sanitize(message: str) → str#

对消息进行脱敏处理。

参数:: message -- 原始消息字符串。
返回:: 脱敏后的消息字符串。

_sanitize_args(args: tuple | dict) → tuple | dict#

对日志参数进行脱敏处理。

参数:: args -- 日志参数，可以是元组或字典。
返回:: 脱敏后的参数。

_sanitize_value(value) → str#

对单个值进行脱敏处理。

参数:: value -- 原始值。
返回:: 脱敏后的字符串表示。

taolib.testing.logging_config.configure_logging(level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] = 'INFO', log_file: pathlib.Path | str | None = None, format_string: str | None = None, date_format: str | None = None, enable_sanitize: bool = True, sanitize_config: dict | None = None, format_mode: Literal['text', 'json'] | None = None, service: str = '') → SensitiveDataFilter | None#

配置日志系统。

参数:

level -- 日志级别，默认为 "INFO"。
log_file -- 日志文件路径，如果为 None 则只输出到控制台。
format_string -- 日志格式字符串，如果为 None 则使用默认格式。仅 text 模式生效。
date_format -- 日期格式字符串，如果为 None 则使用默认格式。
enable_sanitize -- 是否启用敏感数据脱敏，默认为 True。
sanitize_config -- 脱敏配置字典，支持以下键： - enable_password: 是否启用密码脱敏 - enable_jwt: 是否启用 JWT 密钥脱敏 - enable_api_key: 是否启用 API 密钥脱敏 - enable_email: 是否启用邮箱脱敏 - enable_phone: 是否启用手机号脱敏 - enable_ip: 是否启用 IP 地址脱敏 - custom_rules: 自定义脱敏规则列表
format_mode -- 日志输出格式，"text"``（默认）或 ``"json"。设为 None 时读取环境变量 LOG_FORMAT，未设置则默认 "text"。
service -- 服务名称，仅 json 模式下写入日志。

返回:

如果启用脱敏，返回 SensitiveDataFilter 实例；否则返回 None。

示例

>>> configure_logging(level="DEBUG", enable_sanitize=True)
>>> configure_logging(format_mode="json", service="config-center")

taolib.testing.logging_config.get_logger(name: str) → logging.Logger#

获取指定名称的日志记录器。

参数:: name -- 日志记录器名称，通常使用 __name__。
返回:: 配置好的 Logger 实例。

class taolib.testing.logging_config.RemoteLogHandler(endpoint: str, service: str, api_key: str | None = None, batch_size: int = 50, flush_interval: float = 5.0, level: int = logging.INFO)#

Bases: logging.Handler

远程日志平台 Handler。

将日志记录通过 HTTP 发送到远程日志平台，支持批量发送和优雅降级。

参数:

endpoint -- 远程日志平台的 HTTP 端点 URL。
service -- 当前服务的名称。
api_key -- API 认证密钥（可选）。
batch_size -- 批量发送的日志条数，默认为 50。
flush_interval -- 刷新间隔（秒），默认为 5.0。
level -- 日志级别，默认为 INFO。

示例

>>> from taolib.testing.logging_config import RemoteLogHandler
>>> handler = RemoteLogHandler(
...     endpoint="http://localhost:8100/api/v1/logs/ingest",
...     service="my-app",
... )
>>> logging.getLogger().addHandler(handler)

endpoint#

service#

api_key = None#

batch_size = 50#

flush_interval = 5.0#

_buffer: list[dict] = []#

_lock#

_running = True#

_formatter#

_flush_thread#

emit(record: logging.LogRecord) → None#

发送日志记录到远程平台。

参数:: record -- 日志记录对象。

_flush_loop() → None#: 定期刷新缓冲区。

_flush() → None#: 将缓冲区日志发送到远程平台。

_send_logs(logs: list[dict]) → None#

发送日志到远程平台。

参数:: logs -- 日志条目列表。
抛出:: Exception -- 如果发送失败。

close() → None#: 关闭 Handler 并刷新缓冲区。

taolib.testing.logging_config.configure_remote_logging(endpoint: str, service: str, api_key: str | None = None, batch_size: int = 50, flush_interval: float = 5.0, level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] = 'INFO', log_file: pathlib.Path | str | None = None) → RemoteLogHandler#

配置同时输出到本地和远程日志平台。

该函数会先调用 configure_logging() 配置本地日志输出，然后添加 RemoteLogHandler 到 root logger。

参数:

endpoint -- 远程日志平台的 HTTP 端点 URL。
service -- 当前服务的名称。
api_key -- API 认证密钥（可选）。
batch_size -- 批量发送的日志条数，默认为 50。
flush_interval -- 刷新间隔（秒），默认为 5.0。
level -- 日志级别，默认为 "INFO"。
log_file -- 本地日志文件路径，如果为 None 则只输出到控制台。

返回:

配置好的 RemoteLogHandler 实例。

示例

>>> from taolib.testing.logging_config import configure_remote_logging
>>> handler = configure_remote_logging(
...     endpoint="http://localhost:8100/api/v1/logs/ingest",
...     service="my-app",
... )

taolib.testing.logging_config

目录

taolib.testing.logging_config#

Attributes#

Classes#

Functions#

Module Contents#