taolib.testing.audit.logger

审计日志记录器。

提供审计日志的记录功能，支持多种存储后端。

Attributes

logger

Classes

AuditStorageProtocol	审计日志存储后端协议。
InMemoryAuditStorage	内存审计日志存储。
FileAuditStorage	文件审计日志存储。
MongoDBAuditStorage	MongoDB 审计日志存储。
AuditLogger	审计日志记录器。

Module Contents

taolib.testing.audit.logger.logger

class taolib.testing.audit.logger.AuditStorageProtocol

Bases: Protocol

审计日志存储后端协议。

定义审计日志存储后端需要实现的所有操作。

async save(log: taolib.testing.audit.models.AuditLog) → None

保存审计日志。

参数:

log -- 审计日志实例

async save_batch(logs: collections.abc.Sequence[taolib.testing.audit.models.AuditLog]) → None

批量保存审计日志。

参数:

logs -- 审计日志列表

async query(filter_params: taolib.testing.audit.models.AuditLogFilter) → list[taolib.testing.audit.models.AuditLogResponse]

查询审计日志。

参数:

filter_params -- 查询过滤器

返回:

匹配的审计日志列表

async count(filter_params: taolib.testing.audit.models.AuditLogFilter) → int

统计审计日志数量。

参数:

filter_params -- 查询过滤器

返回:

匹配的日志数量

async delete_old_logs(before: datetime.datetime) → int

删除旧日志。

参数:

before -- 删除此时间之前的日志

返回:

删除的日志数量

class taolib.testing.audit.logger.InMemoryAuditStorage(max_size: int = 10000)

内存审计日志存储。

适用于测试和开发环境。

_logs

日志存储列表

_max_size

最大存储数量

_logs: list[taolib.testing.audit.models.AuditLog] = []

_max_size = 10000

async save(log: taolib.testing.audit.models.AuditLog) → None

保存审计日志。

参数:

log -- 审计日志实例

async save_batch(logs: collections.abc.Sequence[taolib.testing.audit.models.AuditLog]) → None

批量保存审计日志。

参数:

logs -- 审计日志列表

async query(filter_params: taolib.testing.audit.models.AuditLogFilter) → list[taolib.testing.audit.models.AuditLogResponse]

查询审计日志。

参数:

filter_params -- 查询过滤器

返回:

匹配的审计日志列表

async count(filter_params: taolib.testing.audit.models.AuditLogFilter) → int

统计审计日志数量。

参数:

filter_params -- 查询过滤器

返回:

匹配的日志数量

async delete_old_logs(before: datetime.datetime) → int

删除旧日志。

参数:

before -- 删除此时间之前的日志

返回:

删除的日志数量

_match_filter(log: taolib.testing.audit.models.AuditLog, filter_params: taolib.testing.audit.models.AuditLogFilter) → bool

检查日志是否匹配过滤器。

参数:

• log -- 审计日志

• filter_params -- 过滤器

返回:

是否匹配

class taolib.testing.audit.logger.FileAuditStorage(file_path: str | pathlib.Path, max_size: int = 100000)

文件审计日志存储。

将审计日志保存到 JSON 文件中。

_file_path

日志文件路径

_max_size

最大存储数量

_file_path

_max_size = 100000

_ensure_file_exists() → None

确保日志文件存在。

_read_logs() → list[taolib.testing.audit.models.AuditLog]

读取所有日志。

返回:

日志列表

_write_logs(logs: list[taolib.testing.audit.models.AuditLog]) → None

写入所有日志。

参数:

logs -- 日志列表

async save(log: taolib.testing.audit.models.AuditLog) → None

保存审计日志。

参数:

log -- 审计日志实例

async save_batch(logs: collections.abc.Sequence[taolib.testing.audit.models.AuditLog]) → None

批量保存审计日志。

参数:

logs -- 审计日志列表

async query(filter_params: taolib.testing.audit.models.AuditLogFilter) → list[taolib.testing.audit.models.AuditLogResponse]

查询审计日志。

参数:

filter_params -- 查询过滤器

返回:

匹配的审计日志列表

async count(filter_params: taolib.testing.audit.models.AuditLogFilter) → int

统计审计日志数量。

参数:

filter_params -- 查询过滤器

返回:

匹配的日志数量

async delete_old_logs(before: datetime.datetime) → int

删除旧日志。

参数:

before -- 删除此时间之前的日志

返回:

删除的日志数量

_match_filter(log: taolib.testing.audit.models.AuditLog, filter_params: taolib.testing.audit.models.AuditLogFilter) → bool

检查日志是否匹配过滤器。

class taolib.testing.audit.logger.MongoDBAuditStorage(client: motor.motor_asyncio.AsyncIOMotorClient, database_name: str = 'audit', collection_name: str = 'logs')

MongoDB 审计日志存储。

使用 MongoDB 作为审计日志存储后端。

_client

MongoDB 客户端

_database_name

数据库名称

_collection_name

集合名称

_client

_database_name = 'audit'

_collection_name = 'logs'

_collection

async save(log: taolib.testing.audit.models.AuditLog) → None

保存审计日志。

参数:

log -- 审计日志实例

async save_batch(logs: collections.abc.Sequence[taolib.testing.audit.models.AuditLog]) → None

批量保存审计日志。

参数:

logs -- 审计日志列表

async query(filter_params: taolib.testing.audit.models.AuditLogFilter) → list[taolib.testing.audit.models.AuditLogResponse]

查询审计日志。

参数:

filter_params -- 查询过滤器

返回:

匹配的审计日志列表

async count(filter_params: taolib.testing.audit.models.AuditLogFilter) → int

统计审计日志数量。

参数:

filter_params -- 查询过滤器

返回:

匹配的日志数量

async delete_old_logs(before: datetime.datetime) → int

删除旧日志。

参数:

before -- 删除此时间之前的日志

返回:

删除的日志数量

async create_indexes() → None

创建索引。

_build_query(filter_params: taolib.testing.audit.models.AuditLogFilter) → dict[str, Any]

构建 MongoDB 查询条件。

参数:

filter_params -- 过滤器

返回:

MongoDB 查询字典

class taolib.testing.audit.logger.AuditLogger(storage: AuditStorageProtocol, default_user_id: str | None = None, default_ip_address: str | None = None)

审计日志记录器。

提供审计日志的记录功能，支持同步和异步操作。

_storage

存储后端

_default_user_id

默认用户 ID

_default_ip_address

默认 IP 地址

_storage

_default_user_id = None

_default_ip_address = None

async log(action: str, resource_type: str, resource_id: str | None = None, user_id: str | None = None, details: dict[str, Any] | None = None, ip_address: str | None = None, user_agent: str | None = None, status: str = 'success', error_message: str | None = None) → taolib.testing.audit.models.AuditLog

记录审计日志。

参数:

• action -- 操作类型

• resource_type -- 资源类型

• resource_id -- 资源 ID

• user_id -- 用户 ID

• details -- 操作详情

• ip_address -- IP 地址

• user_agent -- User-Agent

• status -- 操作状态

• error_message -- 错误信息

返回:

创建的审计日志实例

async log_create(resource_type: str, resource_id: str, user_id: str | None = None, details: dict[str, Any] | None = None, ip_address: str | None = None, user_agent: str | None = None) → taolib.testing.audit.models.AuditLog

记录创建操作。

参数:

• resource_type -- 资源类型

• resource_id -- 资源 ID

• user_id -- 用户 ID

• details -- 操作详情

• ip_address -- IP 地址

• user_agent -- User-Agent

返回:

创建的审计日志实例

async log_update(resource_type: str, resource_id: str, user_id: str | None = None, details: dict[str, Any] | None = None, ip_address: str | None = None, user_agent: str | None = None) → taolib.testing.audit.models.AuditLog

记录更新操作。

参数:

• resource_type -- 资源类型

• resource_id -- 资源 ID

• user_id -- 用户 ID

• details -- 操作详情

• ip_address -- IP 地址

• user_agent -- User-Agent

返回:

创建的审计日志实例

async log_delete(resource_type: str, resource_id: str, user_id: str | None = None, details: dict[str, Any] | None = None, ip_address: str | None = None, user_agent: str | None = None) → taolib.testing.audit.models.AuditLog

记录删除操作。

参数:

• resource_type -- 资源类型

• resource_id -- 资源 ID

• user_id -- 用户 ID

• details -- 操作详情

• ip_address -- IP 地址

• user_agent -- User-Agent

返回:

创建的审计日志实例

async log_login(user_id: str, ip_address: str | None = None, user_agent: str | None = None, success: bool = True, error_message: str | None = None) → taolib.testing.audit.models.AuditLog

记录登录操作。

参数:

• user_id -- 用户 ID

• ip_address -- IP 地址

• user_agent -- User-Agent

• success -- 是否成功

• error_message -- 错误信息

返回:

创建的审计日志实例

async log_logout(user_id: str, ip_address: str | None = None, user_agent: str | None = None) → taolib.testing.audit.models.AuditLog

记录登出操作。

参数:

• user_id -- 用户 ID

• ip_address -- IP 地址

• user_agent -- User-Agent

返回:

创建的审计日志实例

async query(filter_params: taolib.testing.audit.models.AuditLogFilter) → list[taolib.testing.audit.models.AuditLogResponse]

查询审计日志。

参数:

filter_params -- 查询过滤器

返回:

匹配的审计日志列表

async count(filter_params: taolib.testing.audit.models.AuditLogFilter) → int

统计审计日志数量。

参数:

filter_params -- 查询过滤器

返回:

匹配的日志数量

async delete_old_logs(days: int = 90) → int

删除旧日志。

参数:

days -- 保留天数

返回:

删除的日志数量