yitool 2.0.0

@yitech/yitool 工具包

yitool

yitool 是面向 Python 业务后台的基建原子能力包。

它已经不只是一个“工具函数合集”。现在的核心价值是把后台系统里反复出现的 审批、待办、附件、下载、通知、导入导出、审计、时间线、业务上下文、幂等、锁、 缓存韧性、任务编排、FastAPI/Celery 集成等能力沉淀为一组框架无关、可测试、 可替换存储的 yi_xxx 模块。

项目定位

yitool 适合这些项目：

• FastAPI 后台、运营管理后台、BFF、内部平台。
• Celery worker、定时任务、异步导入导出、报表任务。
• 需要审计、权限、审批、附件、通知、下载票据的业务系统。
• 希望把业务原子能力从 Web 框架、数据库模型和任务队列中解耦出来的团队。
• 需要在单测、本地脚本、服务端应用中复用同一套基础能力的项目。

核心设计原则：

• yi_xxx 优先：业务后台原子能力是第一入口，传统 utils 是补充。
• 框架无关：核心模型和服务不绑定 FastAPI、Celery、SQLAlchemy 或 MinIO。
• 显式边界：通过 create_xxx_service()、protocol、adapter 和 lifecycle API 接入宿主项目。
• 可替换存储：默认内存 store 适合测试和单进程工具，生产环境注入 DB、Redis 或对象存储实现。
• 可审计：高风险能力优先接入 yi_audit，并可适配 yi_timeline / yi_progress。
• 少隐式全局：默认对象通过显式 get/set/reset 生命周期 API 管理。

安装

pip install .
pip install -e .

本地开发也可以用 uv：

uv pip install -e .

按需安装可选依赖：

pip install "yitool[fastapi]"
pip install "yitool[celery]"
pip install "yitool[db]"
pip install "yitool[cache]"
pip install "yitool[security]"
pip install "yitool[monitor]"
pip install "yitool[excel]"
pip install "yitool[all]"

能力地图

业务后台原子能力

模块	能力	典型场景
yi_approval	顺序审批、通过、驳回、转交、撤回、取消、过期清理、权限钩子、审计、待办/时间线适配	费用报销、权限申请、内容发布、异常单复核
yi_assignment	责任分派、认领、转派、释放、完成、取消、过期清理、审计	工单、待办、人工复核、客服派单
yi_attachment	附件元数据、绑定业务资源、标签、下载票据适配	订单附件、合同附件、报表附件
yi_bulk_action	批量后台操作、逐项结果、部分成功、dry-run、fail-fast、进度/时间线适配	批量启用、禁用、删除、打标、转派
yi_download	一次性下载 token、过期、复用防护、权限校验、文件名规范、MinIO/存储 URL、审计	导出结果、附件下载、报表下载
yi_feature_flag	后端功能开关、定向规则、百分比灰度、加权 variants、store 协议	灰度发布、租户白名单、实验分流
yi_file_intake	文件大小、扩展名、MIME、hash、重复检测、扫描 hook	上传前预检、导入前校验、附件接入
yi_idempotency	幂等 key、请求 fingerprint、reserve/complete/fail 生命周期、store 协议、缓存适配	支付提交、表单重复提交、Webhook、任务触发
yi_job	CSV/Excel 导入、错误行、幂等提交、后台导出、结果文件、过期清理	大批量导入导出、异步报表
yi_notification	邮件、短信、站内信、Push、Webhook 通知编排、偏好过滤、幂等去重	消息中心、审批提醒、任务通知
yi_quota	租户/用户/资源配额、周期窗口、检查、消费、释放、store 协议	导出次数、上传容量、API 调用、任务额度
yi_resource_lock	短 TTL 业务资源锁、续租、释放、强制释放、权限钩子、审计	防重复编辑、审批占用、工单处理锁
yi_sequence	业务流水号、scope、重置周期、格式化模板、号段预留	订单号、发票号、工单号、批次号
yi_scheduler	cron/interval/date 调度、due scan、lease claim、task dispatch	定时报表、延迟任务、重试补跑、清理任务
yi_state_machine	有限状态机、状态定义、transition、guard、hook、审计	订单状态、工单状态、审批草稿状态
yi_template	邮件、短信、站内信、Webhook payload 的变量渲染和模板校验	通知模板、Webhook payload、验证码文案

可信操作与可观测

模块	能力	典型场景
yi_audit	actor/resource/action/result、字段变更、请求元数据、敏感字段 redaction	后台操作审计、权限变更、敏感操作留痕
yi_change_history	资源 revision、字段级 before/after、ChangeSet 多资源聚合	详情页变更历史、批量修改追踪
yi_timeline	资源时间线、审计/变更/附件/任务/审批等 entry 归一	后台详情页动态、操作轨迹
yi_progress	进度状态、事件、snapshot、导入导出适配	异步任务进度、批量操作进度
yi_context	actor、tenant、workspace、request、trace、locale、source、tags、metadata 的 contextvars 上下文	HTTP 请求、Celery 任务、CLI、审计/通知/权限上下文传递
yi_outbox	可靠消息 outbox、状态、dispatcher contract	事务后事件投递、消息最终一致
yi_health	框架无关健康检查聚合器	服务健康检查、依赖探测
yi_monitor	Prometheus 指标与 FastAPI middleware	指标暴露、请求监控
yi_perf	性能监控与轻量 instrumentation	本地性能分析、执行耗时观察

运行时与集成边界

模块	能力	典型场景
yi_fast	FastAPI app factory、响应模型、限流、幂等 dependency、middleware、security lifecycle	FastAPI API 服务
yi_celery	Celery app factory、显式生命周期 API、任务锁	Worker、异步任务、定时任务
yi_task	任务 envelope、handler registry、inline/celery dispatcher、生命周期状态	让业务任务脱离具体队列实现
yi_event_bus	事件发布订阅、监听器、显式默认实例生命周期	进程内事件、领域事件
yi_db	SQLAlchemy/SQLModel 辅助、连接、会话、查询、UoW	数据库应用、Repository/UoW
yi_cache	内存/Redis 缓存、TTL store、cache-aside、singleflight、SWR、逻辑过期、分布式锁、幂等兼容入口	缓存、验证码 TTL、短锁、防击穿
yi_storage	本地/S3 存储抽象、对象存储服务、显式默认存储生命周期	文件上传、下载、附件、导入导出结果
yi_config	配置加载、settings lifecycle	多环境配置、测试隔离
yi_policy	policy gateway、tenant/workspace/ownership/admin guard	权限前置判断、资源访问策略

通用能力与兼容层

模块	能力	典型场景
yi_mail	SMTP/SES/SendGrid/Console mailer、邮箱验证码、验证码安全策略	邮件发送、登录/绑定邮箱验证码
yi_masking	手机、邮箱、身份证、银行卡、token、密钥、嵌套结构脱敏	后台展示、日志、审计、导出
yi_resilience	retry/backoff、sync/async retry、Circuit Breaker	第三方 API、DB、缓存、任务调用保护
yi_serializer	对象序列化、装饰器、JSON/msgpack 相关能力	API payload、缓存值、事件消息
yi_webhook	HMAC 签名、重放保护	Webhook 入站/出站安全
log	rich 日志、结构化日志、上下文	应用日志
utils	字符串、日期、文件、JSON/YAML、路径、随机数、校验等工具函数	低层工具补充
shared / misc / cli	共享组件、历史兼容工具、命令行入口	兼容旧项目和本地工具

快速示例

审批 + 待办 + 审计

from yitool.yi_approval import ApprovalActor, ApprovalResource, ApprovalStep, create_approval_service
from yitool.yi_audit import create_audit_logger

approval = create_approval_service(audit_logger=create_audit_logger())

requester = ApprovalActor(actor_id="requester-1", tenant_id="tenant-a")
manager = ApprovalActor(actor_id="manager-1", tenant_id="tenant-a")

submitted = approval.submit(
    ApprovalResource(resource_type="expense", resource_id="expense-1", tenant_id="tenant-a"),
    actor=requester,
    steps=(ApprovalStep(step_id="manager", name="Manager review", assignee=manager),),
    title="Expense approval",
)

assert submitted.accepted
assert submitted.request is not None

approved = approval.approve(submitted.request.approval_id, actor=manager, reason="ok")
assert approved.accepted

需要把当前审批步骤落成待办时，使用 yi_approval 的 assignment adapter；需要在详情页展示审批动态时，使用 timeline adapter。核心审批服务本身不强制依赖具体待办表或时间线表。

安全下载票据

from yitool.yi_download import DownloadActor, DownloadTarget, MinioDownloadUrlProvider, create_download_service

minio_client = ...  # 例如 minio.Minio(...)
downloads = create_download_service(url_provider=MinioDownloadUrlProvider(client=minio_client, bucket_name="exports"))

ticket = downloads.issue_ticket(
    DownloadTarget(
        storage_provider="minio",
        storage_path="exports/orders.csv",
        file_name="orders.csv",
        content_type="text/csv",
    ),
    actor=DownloadActor(actor_id="admin-1", tenant_id="tenant-a"),
)

assert ticket.accepted

生产环境通常把 DownloadTicketStoreProtocol 接到 Redis 或数据库，把 URL provider 接到 MinIO、S3 或 yi_storage。

邮箱验证码

from yitool.yi_cache import YiCacheMemory
from yitool.yi_mail import YiConsoleMailer, create_email_verification_service

cache = YiCacheMemory(max_size=1000, strategy="TTL", ttl=300)
service = create_email_verification_service(mailer=YiConsoleMailer(), cache=cache)

# await service.send_code("[e-mail]", purpose="login")
# service.verify_code("[e-mail]", "123456", purpose="login")

验证码状态可以使用 yi_cache 作为 TTL store；生产环境建议替换为 Redis，避免多进程状态分裂。

业务幂等

from yitool.yi_idempotency import build_idempotency_fingerprint, create_idempotency_service

idempotency = create_idempotency_service()
fingerprint = build_idempotency_fingerprint({"order_id": "order-1", "amount": 99})

result = idempotency.execute(
    "payment:order-1",
    lambda: {"charged": True},
    fingerprint=fingerprint,
)

需要把副作用拆成多步时，可以先 reserve()，调用外部系统后再 complete()； 失败场景可以 fail() 或让 execute() 自动清理本次占位以便重试。

业务配额

from yitool.yi_quota import QuotaLimit, QuotaResetPeriod, QuotaScope, create_quota_service

quota = create_quota_service(
    limits=(QuotaLimit(name="exports", limit=10, reset_period=QuotaResetPeriod.DAILY),)
)
scope = QuotaScope(tenant_id="tenant-a")

decision = quota.consume("exports", amount=1, scope=scope)
assert decision.accepted

生产环境通常把 QuotaStoreProtocol 接到 Redis 或数据库，并确保 consume() 在并发场景下是原子的。

缓存韧性

from yitool.yi_cache import YiCacheMemory, create_cache_aside, create_singleflight

cache = YiCacheMemory(max_size=1000, strategy="TTL", ttl=300)
cache_aside = create_cache_aside(cache, ttl_seconds=300)
singleflight = create_singleflight()

profile = cache_aside.get_or_set("user:1", lambda: {"id": 1, "name": "Tony"})
result = singleflight.do("expensive:user:1", lambda: profile)

yi_cache 还包含 stale-while-revalidate、logical expire 和短 TTL 分布式锁； 幂等能力推荐从 yi_idempotency 进入，yi_cache 保留兼容导出。

FastAPI / Celery 集成

from yitool.yi_fast import create_fast_app
from yitool.yi_celery import create_celery_app, set_yi_celery

app = create_fast_app(title="Admin API")

celery_app = create_celery_app("admin-worker")
set_yi_celery(celery_app)

集成模块提供 factory 和显式 lifecycle API。测试中可以 reset_xxx() 清理默认实例，避免跨测试状态污染。

业务上下文

from yitool.yi_context import context_actor_dict, context_metadata, yi_context

with yi_context(
    actor_id="admin-1",
    tenant_id="tenant-a",
    workspace_id="workspace-1",
    request_id="req-1",
    trace_id="trace-1",
    source="fastapi",
    tags=("admin-api",),
    metadata={"path": "/orders/1"},
):
    actor = context_actor_dict()
    metadata = context_metadata(extra={"operation": "order.update"})

assert actor["actor_id"] == "admin-1"
assert metadata["operation"] == "order.update"

yi_context 只承载标准化的业务/请求上下文，不做认证、不做授权，也不替代 yi_audit。FastAPI middleware、Celery task wrapper、CLI 入口或测试 fixture 可以在边界设置上下文，下游模块按需读取 actor、request 和 metadata 字段。

推荐组合

后台详情页

• yi_audit 记录操作。
• yi_change_history 记录字段变更。
• yi_timeline 聚合审计、变更、附件、任务、审批动态。
• yi_masking 控制敏感字段展示。

导入导出

• yi_file_intake 做上传预检。
• yi_job 管理导入/导出状态和结果。
• yi_progress 暴露异步进度。
• yi_download 发放一次性下载票据。
• yi_audit 记录下载和任务操作。

审批与人工处理

• yi_approval 表达审批语义。
• yi_assignment 把当前审批步骤转成待办。
• yi_resource_lock 防止多人同时处理同一资源。
• yi_notification 发送提醒。
• yi_timeline 展示全过程。

可靠第三方调用

• yi_resilience 包裹 retry/backoff/circuit breaker。
• yi_outbox 做事务后可靠投递。
• yi_event_bus 做进程内事件分发。
• yi_idempotency 做幂等 key 和请求 fingerprint。
• yi_quota 做租户、用户和资源配额。
• yi_cache 做短锁和缓存防击穿。

生产接入边界

yitool 的默认内存实现主要服务于单测、本地脚本和单进程工具。生产环境建议：

• Store protocol 接数据库、Redis 或对象存储，并保证必要的原子性。
• 权限通过 PermissionCheckerProtocol、yi_policy 或宿主项目 ACL 注入。
• 审计事件接 DB、日志系统或事件管道。
• 请求上下文建议在 HTTP、worker、CLI 等边界通过 yi_context 设置，不把 FastAPI Request、Celery task 或 ORM session 直接传进原子服务。
• 下载、附件、导入导出结果文件接 yi_storage、MinIO 或 S3。
• FastAPI、Celery、SQLAlchemy、Redis 等重依赖通过 extras 按需安装。
• 测试中显式调用 reset_xxx() 清理默认实例。

目录结构

yitool/
├── yi_approval/        # 顺序审批
├── yi_assignment/      # 待办和责任分派
├── yi_attachment/      # 附件元数据
├── yi_audit/           # 审计日志
├── yi_bulk_action/     # 批量后台操作
├── yi_cache/           # 缓存、TTL store、锁和防击穿
├── yi_celery/          # Celery 集成
├── yi_change_history/  # 字段变更历史
├── yi_config/          # 配置
├── yi_context/         # 业务和请求上下文
├── yi_db/              # DB / SQLAlchemy / SQLModel
├── yi_download/        # 安全下载票据
├── yi_event_bus/       # 事件总线
├── yi_fast/            # FastAPI 集成
├── yi_feature_flag/    # Feature Flag
├── yi_file_intake/     # 文件接入校验
├── yi_health/          # 健康检查
├── yi_idempotency/     # 业务幂等
├── yi_job/             # 导入导出任务
├── yi_mail/            # 邮件和邮箱验证码
├── yi_masking/         # 数据脱敏
├── yi_monitor/         # 监控
├── yi_notification/    # 通知编排
├── yi_outbox/          # 可靠消息 outbox
├── yi_policy/          # 策略与权限 guard
├── yi_progress/        # 进度
├── yi_quota/           # 业务配额
├── yi_resilience/      # 重试、退避、熔断
├── yi_resource_lock/   # 业务资源锁
├── yi_sequence/        # 业务流水号
├── yi_scheduler/       # 调度编排
├── yi_state_machine/   # 有限状态机
├── yi_storage/         # 存储抽象
├── yi_task/            # 任务抽象
├── yi_template/        # 模板渲染
├── yi_timeline/        # 资源时间线
├── yi_webhook/         # Webhook 安全
├── log/                # 日志
├── shared/             # 共享组件
└── utils/              # 通用工具函数

文档

• 1.0 基建原子能力
• 1.0 迁移说明
• 质量闸门
• 发布检查清单
• 基准测试说明

开发与质量

uv run ruff check
uv run mypy yitool --follow-imports=skip
uv run pytest
uv run python scripts/ci/check_public_api_snapshot.py
uv run python scripts/ci/check_public_api_boundaries.py
uv run python scripts/ci/check_quality_baselines.py --tool all

当前治理策略是：已有历史债务由 baseline 固定，新代码不能新增 lint/type debt；公开 API 通过 snapshot 和 boundary check 管控。

兼容说明

utils、shared、misc、log 等能力仍然保留，但它们不再是 README 的主线。新项目建议优先从 yi_xxx 原子能力选择模块，再按需使用工具函数作为底层补充。