---

✨ 为什么选择 apiboot

┌─────────────────────────────────────────────────────────────────┐
│ 你的项目                                                          │
│   ├── main.py        FastAPI 路由                                 │
│   ├── .env           配置 (DB / Redis / LLM / OCR ...)             │
│   ├── models/        ORM 模型 (apiboot 自动 import 建表)            │
│   ├── tasks/         cron 任务 (apiboot 自动 register)             │
│   └── requirements   你自己选的依赖版本                            │
│                                                                 │
│ apiboot  (pip install apiboot, 仅 apscheduler 一个运行时依赖)        │
│   ├── config / logger / middlewares / error / schemas             │
│   ├── db.mysql (DB_MODELS_DIR 自动扫) / db.redis / redis.lock     │
│   ├── cron (TASKS_DIR 自动扫) / ocr / llm / torch / data / retry  │
│   └── cli (abtstart/abtstop/abtrestart/abtstatus/... + 滚动重启)   │
└─────────────────────────────────────────────────────────────────┘

痛点	apiboot 的解法
装一个工具包拖一堆用不到的依赖	默认 dependencies = [apscheduler],业务侧第三方全部按需懒加载,没装就不触发 import
sync / async 数据库 API 风格不统一	统一前缀约定:sync 用裸函数名,async 加 a 前缀,13 对 CRUD 一一对应
项目里写一堆 start.sh / stop.sh	abtstart / abtstop / abtstatus / abtlist / abtlog 一个命令搞定
多 worker 进程重启会中断服务	abtrestart 默认走滚动重启(uvicorn workers 逐个 SIGTERM,前端请求 0 中断)
服务被 OOM Killer 杀后无法自动恢复	CLI 内置 supervisor:自动 fork-safe + OOM 自动重启 + 内存阈值预警 + 雪崩保护
业务 / 系统 / HTTP 三类异常混在一起	三个根异常(ApiError / ApiSystemError / HttpError) + 统一错误码常量 + http_status 让业务异常正确映射 HTTP 状态码
Pydantic 字段命名在前后端来回转换	BaseSchema 自动 snake_case ↔ camelCase
models/ / tasks/ 目录漏 import 漏建表	DB_MODELS_DIR / TASKS_DIR 配置驱动自动扫描,业务零感知

---

📦 安装

pip install apiboot

默认只依赖 apscheduler 一个第三方包,不会污染你的依赖图。其余可选依赖按需安装。

按需安装可选依赖(只装你用得到的):

# 最小起步: FastAPI + ORM + Redis
pip install fastapi sqlalchemy pymysql redis httpx pydantic

# AI / OCR 全家桶
pip install langchain langchain-core jieba httpx

# 重试 / 加密 / 雪花 ID
pip install tenacity cryptography snowflake-id

# ML / 数据处理
pip install torch pandas numpy scikit-learn

# 文档解析
pip install pymupdf python-docx python-pptx openpyxl xlrd

# CLI 内存监控 (Linux 用 /proc 不需要, macOS / Windows 必需)
pip install psutil

💡 提示:apiboot 的所有第三方依赖都走懒加载 —— pip install apiboot 之后 import apiboot 永远不会因为缺失依赖而失败。

---

🚀 快速开始

下面六个例子覆盖 90% 的日常场景,每个都能独立运行。

① 一行启动:配置 + 日志

from apiboot import config, logger

logger.info("服务启动")
print(config.DB_HOST)     # 自动读取 .env
print(config.DEBUG)        # 自动类型转换: 'true' → True

② FastAPI 统一响应

from fastapi import FastAPI
from apiboot.middlewares import add_middlewares

app = FastAPI()
add_middlewares(app)   # 一行注册所有内置中间件, 按推荐顺序(日志外层 / 响应包装内层)

@app.get("/users/{uid}")
def get_user(uid: str):
    return {"name": "张三", "age": 18}
    # → 自动包装: {"success": True, "code": 200, "message": "成功", "data": {...}}

add_middlewares 等价于手动写两行 add_middleware(...), 顺序写死, 业务侧只用记一个函数名。也可单独关闭某个中间件或自定义配置:

# 只要响应包装, 不要日志
add_middlewares(app, include_logging=False)

# SSE 接口排除
add_middlewares(app, json_response_exclude_paths=["/api/chat/stream"])

响应里多出 success: true/false 字段, 前端可以直接 if (resp.success) 判断, 不用关心 SUCCESS_CODE 是 0 还是 200。

③ 统一错误体系

from apiboot.error import BusinessError
from apiboot.error.code import USER_NOT_FOUND

class UserNotFoundError(BusinessError):
    default_code = USER_NOT_FOUND.code      # 30001

raise UserNotFoundError("用户不存在")         # code=30001, HTTP 200

④ 通用 SQL CRUD (MySQL / PostgreSQL / SQLite)

from apiboot.db.sql import init_db, get_db_session, save, query_page

init_db()                                   # 启动时建表 + 自动 import models/ 目录
with get_db_session() as session:
    save(session, User(name="张三"))
    users, total = query_page(session, User, page_num=1, page_size=20)

# 异步版 (FastAPI async def 路由)
from apiboot.db.sql import ainit_db, aget_db_session, asave, aquery_page

async with aget_db_session() as session:
    await asave(session, User(name="张三"))
    users, total = await aquery_page(session, User, page_num=1, page_size=20)

在 .env 中通过 DB_TYPE 选择后端：mysql（默认）、postgresql 或 sqlite。MySQL 需要 pymysql / aiomysql，PostgreSQL 需要 psycopg2-binary / asyncpg，SQLite 异步模式需要 aiosqlite。

换驱动通过 DB_DRIVER(同步) / DB_ASYNC_DRIVER(异步)配置覆盖: 各后端有默认 driver, 不配也能跑:

DB_TYPE	同步默认	异步默认	自定义示例
mysql	pymysql	aiomysql	DB_DRIVER=mysqlclient
postgresql	psycopg2	asyncpg	DB_DRIVER=psycopg
sqlite	sqlite	sqlite+aiosqlite	DB_DRIVER=sqlite+pysqlite

# .env: 用 mysqlclient 替代默认 pymysql
DB_DRIVER=mysqlclient

# 异步单独配 (fallback 到 DB_DRIVER, 不配就用异步默认)
DB_ASYNC_DRIVER=asyncpg

自动建表依赖 DB_MODELS_DIR(默认 "models"):目录不存在时静默跳过,走老的"用户自己 import model"逻辑。

跨方言支持(MySQL / PostgreSQL / SQLite)

apiboot.db.sql 在 3 个数据库后端都能跑通,业务代码零感知切换 DB_TYPE:

# .env
# DB_TYPE=mysql         # 默认
# DB_TYPE=postgresql
DB_TYPE=sqlite

手写原生 SQL 时(如 db_utils.encrypt(model_cls) 批量加密脚本)已自动跨方言, 通过 apiboot.db.sql.backends Protocol 实现:

方言差异	解决方式
标识符引用 `tbl` / "tbl"	backend.quote_ident('tbl') (MySQL/SQLite 反引号, PG 双引号)
JSON 值写入 CAST(:v AS JSON) / :v::jsonb	backend.json_value_expr('v') (各方言专属表达式)
UPSERT ON DUPLICATE KEY / ON CONFLICT / OR IGNORE	backend.apply_upsert(...) (自动派发)
各 dialect 私有属性 / 函数	禁止跨方言污染, 全部封装在 backends/{mysql,postgresql,sqlite}.py 内部

业务侧需要写原生 SQL 时, 必须通过 backend.quote_ident(...) 和 backend.json_value_expr(...) 拼接, 详见 CLAUDE.md "SQL 跨方言规范" 章节。

新增方言(如 Oracle / SQL Server)只需在 apiboot/db/sql/backends/ 加一个实现 DatabaseBackend Protocol 的文件, 注册即可。

对象 ↔ dict 互转规范

业务函数入参往往是"什么类型都行"(dict / SQLAlchemy ORM / Pydantic v1+v2 / SQLModel / dataclass / namedtuple ...), 禁止每个业务函数自己写 if/elif isinstance 分支判断。统一委托 apiboot.utils.obj_utils:

from apiboot.utils.obj_utils import obj_to_dict, dict_to_obj

def save(model: type, data) -> int:
    """data 可以是 dict / ORM / pydantic / dataclass / namedtuple, 都规范化成 dict."""
    data_dict = obj_to_dict(data)  # ← 一行替代所有 if/elif isinstance
    stmt = sql_insert(model).values(data_dict)
    return session.execute(stmt).rowcount or 0

def parse(raw: dict, cls: type):
    """把 dict 转回 Pydantic / dataclass 实例."""
    return dict_to_obj(raw, cls)

函数	方向	支持输入
obj_to_dict(obj, *, deep=False)	对象 → dict	None / 基础类型 / dict / pydantic v1+v2 / SQLModel / SQLAlchemy ORM / dataclass / namedtuple / 普通对象 (vars() 兜底)
dict_to_obj(data, cls)	dict → 类实例	已是 cls 实例原样返回; pydantic v1+v2 (model_validate / parse_obj); dataclass / namedtuple / 普通类 (cls(**data))

反模式(禁止):

# ❌ 自己写 isinstance 分支
if isinstance(data, dict):
    return data
if hasattr(data, "__table__"):
    return {c.name: getattr(data, c.name) for c in data.__table__.columns}
if hasattr(data, "model_dump"):
    return data.model_dump()
raise TypeError(...)

# ❌ 字符串嗅探 __module__.startswith("pydantic")
model_module = type(data).__module__ or ""
if model_module.startswith("pydantic"):
    if hasattr(data, "model_dump"):
        return data.model_dump()

# ❌ 自己写 duck-typing 判 v1/v2
if hasattr(obj, "model_dump"):
    return obj.model_dump()
return obj.dict()  # v1

完整规范、检测命令、例外清单见 CLAUDE.md "对象 ↔ dict 互转规范" 章节。

⑤ Redis 分布式锁(Redisson 风格)

from apiboot.db.redis import init_redis
from apiboot.db.redis.lock import with_lock, LockAcquireError

init_redis()
try:
    with with_lock("order:123", ttl=30, timeout=5) as lock:
        do_critical()                      # Lua CAS 释放 + watchdog 自动续期
except LockAcquireError:
    logger.warning("5 秒内没拿到锁, 跳过")

⚠️ 务必 release 或用上下文管理器:watchdog 后台线程会一直续期,业务异常路径也要 finally 释放,否则锁 key 长时间不被释放。

⑥ LLM 接入

from apiboot.llm import init_model
from apiboot.llm.agent import ChatAgent

llm = init_model()                          # 从 .env 读 LLM_MODEL / LLM_API_KEY / LLM_BASE_URL
agent = ChatAgent(llm)

---

🛡️ CLI 守护进程管理

装上 apiboot 后,6 个全局命令直接可用,无需写 start.sh / stop.sh:

cd /path/to/your-project           # 有 main.py + .env 的目录

abtstart                            # 启动 (后台 + supervisor 守护)
abtstatus                           # 查看状态
abtstop                             # 停止
abtrestart                          # 重启 (默认走滚动重启)
abtlist                             # 列出所有 abtstart 启动的项目
abtlog                              # tail -f .abt.log

或子命令形式:abt start / stop / restart / status / list / log。

常用参数

abtstart --module api.main:app     # 自定义入口
abtstart --port 9000 --host 127.0.0.1
abtstart --workers 4                # 多 worker (uvicorn multiprocessing)
abtstart --reload                   # 开发模式
abtstart --env production           # 加载 .env.production
abtstart --clean                    # 端口被占时强杀

# Supervisor 内存 / 重启阈值
abtstart --mem-warn-ratio 0.80     # RSS 达到 80% 时 WARN
abtstart --mem-hard-ratio 0.95     # RSS 达到 95% 时主动 SIGTERM (防 OOM Killer)
abtstart --max-restarts 5           # 短窗口最大重启次数
abtstart --restart-window 300       # 统计窗口秒数
abtstart --no-supervisor            # 关闭 OOM 守护 (回到旧行为)

# 重启策略 (abtrestart)
abtrestart                          # workers>1 默认滚动重启 (前端请求 0 中断)
abtrestart --force-reload           # 强制 stop+start (改了代码后 reload master)
abtrestart --graceful-timeout 60    # 滚动重启每个 worker 最多等 60s
abtrestart --force                  # 滚动重启跳过 graceful, 直接 SIGKILL worker

关键能力

• 🔍 自动发现 .env + main.py / app.py / server.py
• 🌍 多环境 APP_ENV=production 自动加载 .env.production
• 🚧 端口预检 启动前检测占用,--clean 可强杀
• 🛡️ OOM 守护(默认开启) 子进程被 OOM Killer 杀,自动重启
• 📊 内存阈值 RSS 达到 mem-hard-ratio 时主动 SIGTERM,避免被 OOM Killer 突然终止
• 🚨 雪崩保护 短窗口内重启次数超限,自动放弃
• 👷 多 worker --workers N 切到 uvicorn multiprocessing
• 🔁 滚动重启 abtrestart 在 workers>1 时默认走滚动重启,worker 逐个 SIGTERM → uvicorn master 自动 fork 新 worker 顶上 → 验证就绪 → 处理下一个,整组服务在滚动过程中始终有 N 个 worker 在跑,前端请求 0 中断
• 🧬 fork-safe 子进程标记 APIBOOT_FORKED=1,业务代码可借此 dispose + 重建 SQLAlchemy engine,避免多 worker 共享父进程 FD 导致 "MySQL server has gone away"
• 📋 全局 registry abt list 一台机器混多个项目也清晰可查
• 🧹 残留进程清理 自动杀掉上次没杀干净的 uvicorn 孤儿

⚠️ --workers N 与 OOM 边界:supervisor 的 RSS 防御只对 uvicorn master 可见,看不到后代 worker 的内存。单个 worker 被 OOM Killer 杀由 uvicorn master 自己重启;整套同归于尽才由 supervisor 重启。要管每个 worker 内存,用 systemd MemoryMax= 或 K8s resources.limits.memory,不要指望 supervisor。

⚠️ 滚动重启不 reload 代码:uvicorn worker 是 master 的 fork,继承 master 的代码映像。改了 main.py 等业务文件必须用 --force-reload 走 stop+start 让 master 重启,新 worker 才会 fork 自新 master。滚动重启只 graceful 重启 worker(重置进程内状态、重读 .env 等)。

abtrestart 行为矩阵

场景	路径	中断窗口
workers > 1(默认)	滚动重启 worker	0(整组滚动,服务始终可用)
workers > 1 + 改模块 / env	stop + start	几秒 ~ 30s(不可避免)
workers == 1	stop + start	几秒 ~ 30s(不可避免)
任何场景 + --force-reload	stop + start	几秒 ~ 30s

滚动重启需要项目装 psutil,Linux 用 /proc 也可不装(走 stdlib 实现)。

---

🧩 模块一览

apiboot/
├── config/        # .env 加载器 + 类型化访问器 (env.py)
├── log/           # 控制台 + 文件双输出 logger
├── middlewares/   # FastAPI 中间件 (统一响应 + 请求日志 + 大响应体保护)
├── error/         # 三个根异常 + 错误码常量 + http_status 支持
├── schemas/       # BaseSchema (驼峰) + JsonResult + PageReq
├── db/
│   ├── sql/       # 通用 SQL backend (MySQL / PostgreSQL / SQLite)
│   ├── redis/     # sync + async KV (含 REDIS_MAX_CONNECTIONS)
│   └── redis/lock # Redisson 风格分布式锁
├── cron/          # 定时任务 (装饰器自动注册 + TASKS_DIR 自动扫)
├── llm/           # LangChain chat model + Agent
│   └── model/     # init_model 实现拆到 base.py
├── ocr/           # MinerU API 异步客户端
├── torch/         # CUDA / MPS / CPU 设备探测
├── data/          # pandas / sklearn 常用工具 (大量扩展)
├── retry/         # tenacity 懒加载透传 (PEP 562 lazy module)
├── utils/         # 字符串 / 时间 / JSON / HTTP / 加密 / ...
└── cli/           # abtstart / abtstop / abtrestart / ... (+ 滚动重启)

顶层入口

符号	作用
apiboot.config	一行拿到 .env 配置(自动类型转换)
apiboot.logger	一行拿到 logger(控制台 + 文件双输出)

FastAPI 中间件(apiboot.middlewares)

组件	功能
JsonResponseMiddleware	路由返回值 + 异常统一包装成 {success, code, message, data}
ReqResLoggingMiddleware	请求 / 响应日志(每请求两行,含敏感字段脱敏)
add_middlewares(app)	一行注册所有内置中间件,按推荐顺序(日志外层 / 响应包装内层)

特性:

• 路由返回值 / ApiError / HTTPException / 未预期异常 → 自动包装
• HTTP 状态码映射:业务异常走异常自身的 http_status 类属性(默认 200), 系统异常默认 500;子类可声明 http_status = 401 把 401 业务异常正确返回
• SSE / 流式接口自动跳过(识别 StreamingResponse)
• 非 JSON 响应(文件下载、HTML 页面)原样放行
• 已包装过的响应(顶层已有 code/message/data)原样放行
• 大响应体保护:响应超过 MAX_BODY_SIZE (默认 10MB) 直接 pass-through, 避免 OOM + 不必要的 body 解析
• 双层 try-except:异常路径里中间件自身处理失败会用 INTERNAL_ERROR_CODE 兜底, 永远不会让"框架崩了"代替"业务异常"
• Datetime 自动格式化为 "%Y-%m-%d %H:%M:%S"(可配置)
• 响应 body 超过 LOG_MAX_CHARS 自动截断;password/token/secret/api_key 自动脱敏

跳过某些路径:

app.add_middleware(JsonResponseMiddleware, exclude_paths=["/api/chat/stream"])
# 精确匹配 + 前缀匹配: "/api/llm" → 匹配 /api/llm, /api/llm/chat

或者用 add_middlewares 一行注册(自动按推荐顺序, 顺序写错也不会出错):

from apiboot.middlewares import add_middlewares

app = FastAPI()
add_middlewares(app)
# 等价于:
#   app.add_middleware(ReqResLoggingMiddleware)   # 外层
#   app.add_middleware(JsonResponseMiddleware)    # 内层

# 可选: SSE 排除 + 关闭 RESP body 日志
add_middlewares(
    app,
    json_response_exclude_paths=["/api/chat/stream"],
    log_response_body=False,
)

中间件顺序是 Starlette 栈 LIFO 决定:先 add 的在外层(先被调用)。 本函数内 add 顺序固定为 日志 → 响应包装, 用户不用关心。

业务异常映射 HTTP 状态码:

from apiboot.error import BusinessError
from apiboot.error.code import UNAUTHORIZED

class UnauthorizedError(BusinessError):
    default_code = UNAUTHORIZED.code   # 11001
    http_status = 401                  # 同时覆盖 HTTP 状态码(默认 200)

@app.get("/me")
def me():
    raise UnauthorizedError("登录已过期")
    # → HTTP 401 + {"code": 11001, "message": "登录已过期", "data": null}

错误体系(apiboot.error)

Exception
└─── ApiError ─────────── BusinessError (别名)
    ├── HttpError       # 外部 HTTP 调用失败
    └── ApiSystemError  # 系统异常(HTTP 500,业务侧一般不直接抛)

为什么没有 SystemError 别名:Python 内建 SystemError(解释器内部错误用)如果被遮蔽, 用户的 except SystemError: 会永远捕获不到真正的解释器内部错误。本模块只导出 ApiSystemError,历史用了 from apiboot.error import SystemError 的请改 ApiSystemError。

根类	默认 code	默认 HTTP	用途
ApiError / BusinessError	20000	200	业务侧异常
HttpError	20000	200	外部 HTTP 调用失败
ApiSystemError	10099	500	系统侧异常

子类可同时覆盖 default_code 和 http_status,让业务异常正确映射到 HTTP 状态码 (如 UnauthorizedError.default_code=11001, http_status=401)。 没声明 http_status 时,业务异常一律 200,系统异常一律 500。

错误码常量(apiboot.error.code):

码段	常量
成功	SUCCESS(默认 200, .env 配 SUCCESS_CODE 可改成 0 等)
系统 10xxx	UNKNOWN_ERROR / PARAM_ERROR / INTERNAL_ERROR / SYSTEM_ERROR
认证 11xxx	UNAUTHORIZED / TOKEN_INVALID / PERMISSION_DENIED
数据 12xxx	NOT_FOUND / DATA_EXISTS / DATA_VALIDATION_FAILED
业务 2xxxx	BUSINESS_ERROR / OPERATION_FAILED / STATE_ILLEGAL
用户 30xxx	USER_NOT_FOUND / USER_PASSWORD_ERROR

升级注意:本模块移除了 SystemError 别名(避免遮蔽 Python 内建 SystemError)。 历史用了 from apiboot.error import SystemError 的代码,改成:

from apiboot.error import ApiSystemError

类实例化和抛异常的语法完全不变(raise ApiSystemError("...")), isinstance 判断也兼容(因为 ApiSystemError 是类本身,不是别名)。

Schema(apiboot.schemas)

类	功能	依赖
BaseSchema	pydantic Schema 基类,自动 snake_case ↔ camelCase	pydantic ≥ 2.0
JsonResult[T]	dataclass 实现的统一响应 {success, code, message, data}, success 是 boolean 字段, 前端可 if (resp.success) 直判	stdlib
PageReq	分页请求基类(page_num ≥ 1,page_size 1–500)	pydantic

# BaseSchema 自动驼峰
class UserSchema(BaseSchema):
    user_id: int
    user_name: str

u = UserSchema(user_id=1, user_name="alice")
u.model_dump(by_alias=True)    # → {"userId": 1, "userName": "alice"}

# JsonResult 统一响应 (含 success 字段, 前端 boolean 判断)
from apiboot.schemas.json_result import JsonResult
from apiboot.error.code import NOT_FOUND

# 成功 (success=True, code=SUCCESS.code)
JsonResult.ok(data={"id": 1}).to_dict()
# → {"success": True, "code": 200, "message": "成功", "data": {"id": 1}}

# 失败 (success=False, code=12001)
JsonResult.fail(code=NOT_FOUND).to_dict()
# → {"success": False, "code": 12001, "message": "资源不存在", "data": None}

SUCCESS.code 默认 200, 在 .env 配 SUCCESS_CODE=0 可改成 0 等其他值。 前端只需用 resp.success 字段判断, 不用关心 SUCCESS_CODE 是 0 还是 200。

数据库(apiboot.db)

通用 SQL CRUD (26 个函数)

类别	函数
增	save / save_batch
改	update / update_batch / save_or_update / save_or_update_batch (各后端原生 UPSERT)
查	query / query_primary_key / query_page(rows + total)— 均支持 fields=[...]
删	delete / delete_batch / soft_delete / soft_delete_batch(UPDATE deleted_at = NOW())

异步版一一对应,加 a 前缀:asave / aquery_page / asoft_delete …

自动建表 + models 目录扫描

项目结构
├── models/
│   ├── __init__.py
│   ├── user.py          # class User(Base): ...
│   └── order.py         # class Order(Base): ...
└── .env  →  DB_MODELS_DIR=models   # 默认值, 目录不存在静默跳过

from apiboot.db.sql import init_db
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

init_db(Base)      # 启动时自动 import models/ 下所有 .py + create_all

• 每个 engine 独立标记 lazy 建表状态(WeakKeyDictionary),多 engine 场景互不干扰
• DB_MODELS_DIR= 显式禁用扫描
• 单个 model 文件 import 失败仅 WARN,不阻断其他文件

多 engine 支持

# 测试场景:换 engine 后,期望重新跑 lazy 建表 —— 现在做到了
engine_a = create_engine("mysql://a")
engine_b = create_engine("mysql://b")
lazy_ensure_tables_sync(engine_a, Base)   # 跑
lazy_ensure_tables_sync(engine_b, Base)   # 也跑 (之前 process-global bool 会跳过)

数据库配置项(.env)

通过 :mod:apiboot.config.env 里的类型化访问器读取,常用配置如下:

Key	默认值	适用
DB_ENABLE	true	通用 SQL 子系统总开关。CI / 单元测试无 DB 时显式设 false
DB_TYPE	mysql	数据库后端:mysql / mysql+mysqlclient / postgresql / postgres / pg / sqlite / sqlite3。未知值会在 get_engine() 时直接 ValueError
DB_DRIVER	后端默认	同步数据库驱动(覆盖 backend 类属性)。MySQL 默认 pymysql, PG 默认 psycopg2, SQLite 默认 sqlite。e.g. mysqlclient / psycopg
DB_ASYNC_DRIVER	异步默认	异步数据库驱动, 默认按后端: aiomysql / asyncpg / sqlite+aiosqlite。未配时 fallback 到 DB_DRIVER
DB_HOST	localhost	数据库主机;SQLite 下表示文件路径(可相对 CWD)
DB_PORT	按 DB_TYPE 推断	mysql=3306 / postgresql=5432 / sqlite=0
DB_USERNAME	root	数据库用户名,SQLite 不使用
DB_PASSWORD	""	数据库密码(⚠️ 敏感)。包含 @/: 等特殊字符时 SQLAlchemy URL 会自动 quote
DB_NAME	""	MySQL/PostgreSQL 表示逻辑库;SQLite 下表示数据库文件路径,留空走 :memory:
DB_CHARSET	utf8mb4	MySQL/PostgreSQL 客户端编码;SQLite 不使用
DB_PREFIX	APP_NAME + "_"	SQL 表名前缀,字符规则 [A-Za-z][A-Za-z0-9_]*,多环境共享 DB 强烈建议设置
DB_MODELS_DIR	models	init_db / ainit_db 启动时自动 import 的 ORM 文件目录
DB_USE_MODERN_UPSERT	false	MySQL 专属,只有确认用现代别名语法(> 8.0.20)才设为 true

完整配置项和默认值见 apiboot/config/env.py 顶部 docstring。

Redis KV

init_redis / close_redis / get_redis
set_value / get_value / delete / exists / expire / mget / set_json / get_json / ping
get_value_raw / read_json / scanned_keys / prefixed_pattern
ainit_redis / aclose_redis / aget_redis / aget_redis_session
aset_value / aget_value / adelete / aexists / aexpire / amget / aset_json / aget_json / aping

支持 redis 3.5+(仅 sync),4.2+(sync + async),5.0+ / 6.0+ 推荐。

集群模式:REDIS_CLUSTER_NODES=host1:port,host2:port,host3:port(自动启用)。

连接池大小:REDIS_MAX_CONNECTIONS=N(高并发 FastAPI 建议 100~200,默认 redis-py 自定)。

Redis 分布式锁(apiboot.db.redis.lock)

类型	API
同步	RedisLock / LockAcquireError / acquire_lock / with_lock / locked
异步	AsyncRedisLock / AsyncLockAcquireError / aacquire_lock / awith_lock / alocked
常量	DEFAULT_TTL=30 / DEFAULT_RETRY_INTERVAL=0.1 / DEFAULT_RENEW_RATIO=1/3

Redisson 风格保证:

• 🛡️ Lua compare-and-delete — 只删自己 token 的锁,防止误删别人的锁
• 🔄 watchdog 自动续期 — 后台线程(sync)/ asyncio.Task(async),默认开启
• ⏳ 阻塞等锁 — spin 循环 + 超时控制

定时任务(apiboot.cron)

from apiboot.cron import scheduled_job, start_scheduler

@scheduled_job("cron", hour=6, minute=0)
def _daily_pipeline_6am():
    print("早上 6 点跑批")

if __name__ == "__main__":
    start_scheduler()    # 无需传扫描路径 —— 装饰器已自动注册

异步版(FastAPI lifespan):

from apiboot.cron import async_scheduled_job, astart_scheduler, astop_scheduler

@async_scheduled_job("cron", hour=6, minute=0)
async def _daily_pipeline_async():
    print("async 跑批")

@asynccontextmanager
async def lifespan(app):
    await astart_scheduler()
    yield
    await astop_scheduler()

约定目录自动扫描(TASKS_DIR)

项目结构
├── tasks/
│   ├── sync_data.py    # def register(scheduler): scheduler.add_job(...)
│   └── report.py
└── .env  →  TASKS_DIR=tasks    # 默认值, 目录不存在静默跳过

# start_scheduler 启动时会自动调 register_jobs_from_dir(tasks_dir)
# 等价于在 main.py 里手动写 register_jobs_from_dir("tasks", scheduler_obj)

API	说明
同步	scheduled_job / start_scheduler / stop_scheduler / get_scheduler / scheduler
异步	async_scheduled_job / ainit_scheduler / astart_scheduler / astop_scheduler / scheduler_async
约定扫描	register_jobs_from_dir / register_jobs_from_files(与传统 def register(scheduler) 风格兼容)
配置 helper	build_job_defaults / default_timezone / is_scheduler_enabled

LLM / Agent(apiboot.llm)

模块	功能	依赖
llm.init_model	LangChain chat model 一键构造(自动派发 provider)	langchain ≥ 1.0
llm.agent.ChatAgent	流式聊天 Agent 封装(适配 FastAPI StreamingResponse)	langchain
data.utils.jieba_utils	jieba 中文分词懒加载封装	jieba

from apiboot.data.utils.jieba_utils import lcut, load_userdict

load_userdict("./jieba自定义词典.txt")
words = lcut("小明毕业于北京大学计算机系")
# → ['小明', '毕业', '于', '北京大学', '计算机系']

迁移说明: 原 apiboot.llm.utils.jieba_utils 已迁移到 apiboot.data.utils.jieba_utils。 旧路径下不再保留兼容 shim, 项目内无外部引用, 直接迁移。

📦 模块组织:init_model 的实现拆到 apiboot.llm.model.base,apiboot.llm.model.__init__ 仅做导出。业务方 from apiboot.llm import init_model 不受影响。

OCR(apiboot.ocr)

import asyncio
from apiboot.ocr import MinerU

client = MinerU()                                       # 从 .env 读 MINERU_URL
md = asyncio.run(client.parse_file("/path/to/report.pdf"))

类 / 函数	功能
MinerU	MinerU API 异步客户端(PDF / 图片 / Office → markdown)
MinerUAPIError	MinerU 调用失败的异常(含 status_code / url / method)
mineru_is_supported	判断文件扩展名是否被 MinerU 支持
SUPPORTED_EXTENSIONS / UNSUPPORTED_EXTENSIONS	支持 / 不支持的扩展名常量

支持格式:.pdf / .jpg / .jpeg / .png / .gif / .bmp / .webp / .tiff / .tif / .xlsx / .docx / .pptx / .ofd(.xls 不支持)。

后端可配:MINERU_BACKEND=hybrid-engine(默认,平衡)/ vlm-engine(精度高,速度慢);MINERU_EFFORT=medium / high。

数据 / ML 工具

apiboot.torch

函数	功能	依赖
get_torch_device()	返回当前最优 torch.device(cuda > mps > cpu)	torch
get_device_info()	返回详细探测 dict	torch

apiboot.data.utils.pd_utils

分类	函数
空值处理	drop_empty_rows / fill_empty / fillna_with_strategy / ffill_column / bfill_column / missing_summary
类型转换	safe_to_numeric / safe_to_datetime / safe_to_categorical
去重	drop_duplicate_rows
字符串	normalize_string_column / truncate_string_column / extract_numbers_from_string / split_column / contains_pattern
列操作	select_columns / rename_columns / drop_columns / reorder_columns / cast_columns / move_column
日期处理	extract_date_parts / fill_missing_dates
数值处理	clip_outliers_iqr / bin_numeric_column
断言	assert_columns_exist / assert_no_nulls / assert_unique
报告 / 统计	value_counts_pct / describe_extended
重塑 / 抽样	melt_to_long / pivot_to_wide / stratified_sample / groupby_agg
I/O	safe_read_csv
通用	df_to_dict

apiboot.data.utils.sklean_utils

分类	函数
数据缩放	standard_scale / minmax_scale / robust_scale / maxabs_scale
编码	label_encode / onehot_encode
填充缺失	simple_impute
模型评估	classification_metrics / regression_metrics / confusion_matrix_df / cross_validate
特征	polynomial_features
数据切分	train_test_split / train_val_test_split / kfold_split / stratified_kfold_split
模型持久化	save_model / load_model(用 joblib)
重采样	oversample_minority / undersample_majority

from apiboot.torch import get_torch_device, get_device_info
from apiboot.data.utils import train_test_split, standard_scale
from apiboot.data.utils.pd_utils import drop_empty_rows, missing_summary

device = get_torch_device()               # torch.device("mps") / "cuda" / "cpu"
model.to(device)

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# sklearn 工具(走 joblib 持久化,不要用 pickle)
from apiboot.data.utils.sklean_utils import save_model, load_model
save_model(model, "model.joblib")
model = load_model("model.joblib")

重试(apiboot.retry)

Tenacity 的懒加载透传,行为完全等价:

分类	符号
核心	retry / Retrying / AsyncRetrying / RetryError / TryAgain / NO_RESULT
停止策略	stop_after_attempt / stop_after_delay / stop_before_delay / stop_any / stop_all / stop_never / stop_when_event_set
等待策略	wait_fixed / wait_random / wait_random_exponential / wait_incrementing / wait_exponential / wait_exponential_jitter / wait_full_jitter / wait_combine / wait_chain / wait_none / wait_exception
重试条件	retry_if_result / retry_if_not_result / retry_if_exception / retry_if_exception_type / retry_if_not_exception_type / retry_if_exception_cause_type / retry_if_exception_message / retry_if_not_exception_message / retry_unless_exception_type / retry_any / retry_all / retry_always / retry_never
回调钩子	before_log / after_log / before_sleep / before_sleep_log / before_nothing / after_nothing / before_sleep_nothing

📦 实现:本模块用 PEP 562 lazy module(__getattr__ + __dir__)实现按需导出,Python 3.7+ 标准 pattern。业务方 from apiboot.retry import ... 无感知。

from apiboot.retry import retry, stop_after_attempt, wait_fixed, retry_if_exception_type

@retry(stop=stop_after_attempt(3), wait=wait_fixed(1), retry=retry_if_exception_type(ConnectionError))
def fetch():
    ...

工具集合(apiboot.utils)

顶层一行导入(纯 stdlib)

分类	函数
路径与文件	get_project_root / get_parent_path / ensure_dir / path_join / normalize_path / get_path_segments / to_relative_path / to_absolute_path / is_relative_to / change_extension / safe_join / find_executable / get_file_type / get_file_name / get_file_stem / get_file_dir / get_file_size / get_modified_time / get_created_time / file_exists / is_file / is_directory / read_text / read_text_safe / write_text / read_bytes / write_bytes / read_lines / write_lines / delete_file / copy_file / move_file / touch / list_files / list_dirs
对象 / 字典	obj_to_dict / dict_to_obj / obj_is_null / obj_is_not_null / obj_get_attr / obj_filter_none / obj_pick / obj_omit / obj_merge / obj_deep_copy / is_dataclass / is_pydantic_model
字符串	str_len / str_is_empty / str_is_not_empty / str_is_blank / str_is_not_blank / str_preview / to_snake_case / to_camel_case / to_pascal_case / to_kebab_case / truncate / truncate_middle / pad_left / pad_right / pad_center / collapse_whitespace / strip_chars / remove_all_whitespace / mask_email / mask_phone / mask_id_card / is_chinese / contains_chinese / is_digits / is_ascii / is_uuid / contains_any / contains_all / default_if_empty / default_if_blank / slugify / reverse_str / repeat_str / remove_prefix / remove_suffix / generate_uuid
日期时间	now / today / yesterday / timestamp / timestamp_ms / timestamp_us / now_utc / timestamp_to_datetime / timestamp_to_str / format_time / parse_time / try_parse_datetime / parse_to_timestamp / start_of_day / end_of_day / start_of_month / end_of_month / add_days / add_hours / add_minutes / add_seconds / add_months / add_years / is_same_day / is_today / is_yesterday / days_between / hours_between / format_iso / parse_iso / to_utc / to_local / humanize_duration / is_leap_year / days_in_month
JSON	to_json_str / from_json_str / safe_from_json / to_json_bytes / from_json_bytes / load_json_file / save_json_file / pretty_json / minify_json / validate_json / json_equal / jsonl_iter / jsonl_save
列表	list_len / list_is_empty / list_is_not_empty / list_is_blank / list_union / list_intersection / list_difference / list_symmetric_difference / list_contains / list_contains_any / list_contains_all / list_contains_none / list_index_of / list_distinct / list_reverse / list_sort / list_flatten / list_chunk / list_partition / list_take / list_skip / list_concat / list_push / list_compact / list_remove / list_remove_blank / list_first / list_last / list_get / list_safe / list_join / list_to_tuple / list_count
数学	add / sub / mul / div / round_to / percent / change_percent / random_int / random_float / random_choice / random_sample / random_seed / clamp / lerp / is_close / gcd / lcm / is_prime / sum_safe / mean / median / stdev
依赖探测	get_installed_version

需要第三方包的子模块

子模块	功能	可选依赖
utils.http_utils	同步 + 异步 HTTP 客户端(连接池复用 / 自动重试 / 流式下载)	requests / httpx / aiohttp
utils.queue_utils.StreamQueue	异步流式队列(SSE / WebSocket 哨兵结束)	stdlib
utils.obj_utils	对象 ↔ dict 互转(pydantic / SQLAlchemy / dataclass 自动识别)	pydantic / sqlalchemy / sqlmodel
utils.json_utils	pydantic 兼容 JSON(pydantic_to_json / pydantic_from_json 等)	pydantic
utils.image_utils	PDF / PPTX → 图片(PyMuPDF + LibreOffice), 含单页 / 缩略图 / base64	pymupdf + 系统 LibreOffice
utils.poi_utils	PDF / PPTX / DOCX / Excel → 纯文本	pymupdf / python-docx / python-pptx / openpyxl / xlrd / pandas
utils.snowflake_utils	雪花 ID 生成器(线程安全全局单例)	snowflake-id
utils.password_utils	用户密码哈希(PBKDF2)+ 字段加密(Fernet)+ HMAC + 随机 token	stdlib + cryptography
utils.base_utils	懒加载工具(require_module / _get_optional_module / get_installed_version)	stdlib

字符串工具(apiboot.utils.str_utils)

参考 hutool (Java) 的 :class:StringUtil 思路, 按场景分组, 全部 None 安全 (判空 / 默认值 / 搜索), 类型严格 (大小写转换 / 校验遇到非 str 抛 TypeError):

分类	函数
长度 / 判空	str_len / str_is_empty / str_is_not_empty / str_is_blank / str_is_not_blank / str_preview
大小写转换	to_snake_case / to_camel_case / to_pascal_case / to_kebab_case
截断	truncate / truncate_middle (中间省略, 适合长 ID/Hash 显示)
填充	pad_left / pad_right / pad_center
空白归一化	collapse_whitespace / strip_chars / remove_all_whitespace
掩码 (日志脱敏)	mask_email / mask_phone / mask_id_card
校验	is_chinese / contains_chinese / is_digits / is_ascii / is_uuid
搜索	contains_any / contains_all
默认值	default_if_empty / default_if_blank
URL slug	slugify (NFKC 归一化 + ASCII-only, 中文会被 drop, 需拼音接 pypinyin)
反转 / 重复	reverse_str / repeat_str
前后缀移除	remove_prefix / remove_suffix (兼容 3.7+)
UUID	generate_uuid

from apiboot.utils import (
    to_snake_case, mask_phone, slugify,
    contains_any, default_if_blank, repeat_str,
)

to_snake_case("HelloWorld")          # → "hello_world"
mask_phone("13800138000")            # → "138****8000"
slugify("Hello World! 你好")           # → "hello-world"  (中文 drop)
contains_any("error.log", ".log", ".err")  # → True
default_if_blank(None, "N/A")        # → "N/A"
repeat_str("-", 30)                  # → "------------------------------"

列表工具(apiboot.utils.list_utils)

参考 hutool (Java) 的 :class:CollUtil 思路, 按场景分组, 全部 None 安全 / 不可变优先 (排序 / 反转 / 去重 / 移除 都返回新列表, 不修改原列表):

分类	函数
长度 / 判空	list_len / list_is_empty / list_is_not_empty / list_is_blank
集合运算	list_union / list_intersection / list_difference / list_symmetric_difference (元素需可哈希)
包含 / 搜索	list_contains / list_contains_any / list_contains_all / list_contains_none / list_index_of
排序 / 反转 / 去重 / 扁平	list_distinct (保序) / list_reverse / list_sort (不修改原列表) / list_flatten (支持 depth)
分块 / 切片 / 拆分	list_chunk / list_partition / list_take / list_skip
添加 / 删除 / 合并	list_concat / list_push / list_compact / list_remove / list_remove_blank
安全访问	list_first / list_last / list_get / list_safe
转换 / 实用	list_join / list_to_tuple / list_count

from apiboot.utils import (
    list_distinct, list_chunk, list_partition,
    list_remove_blank, list_compact, list_get,
)

list_distinct([1, 2, 2, 3, 1])              # → [1, 2, 3]  (保序)
list_chunk([1, 2, 3, 4, 5], 2)              # → [[1, 2], [3, 4], [5]]
list_partition([1,2,3,4], lambda x: x%2==0) # → ([2, 4], [1, 3])
list_remove_blank([None, "", "x", 0])       # → ["x", 0]   (None 和空白都去, 但保留 0)
list_compact([1, None, 2])                  # → [1, 2]     (只去 None)
list_get([1, 2, 3], 10, default="X")        # → "X"        (越界返回 default)

加密 / 密码工具(apiboot.utils.password_utils)

术语区分 (代码里这 3 个词含义完全不同, 别混用):

中文	参数名	用途	用在
用户密码	password	登录认证, 单向哈希	hash_password / verify_password
加密口令	passphrase	派生 Fernet key 的种子	encrypt / decrypt 系列
共享密钥	key	HMAC 签名的对称密钥	hmac_sign / hmac_verify

分类	函数	依赖
用户密码哈希	hash_password / verify_password (PBKDF2-HMAC-SHA256, 200k 轮)	stdlib
字段对称加密	encrypt / decrypt (Fernet, AES-128-CBC + HMAC-SHA256)	cryptography
NULL 友好版	encrypt_or_none / decrypt_or_none (None/空串直接透传)	cryptography
批量加解密	encrypt_dict / decrypt_dict (dict 指定字段批量加解密)	cryptography
HMAC 签名	hmac_sign / hmac_verify (HMAC-SHA256, 常数时间比较)	stdlib
随机 token	generate_token (URL-safe base64)	stdlib

典型场景: 数据库字段加密 → API 返回前端前解密

# .env
ENCRYPT_SECRET=<48 字节高熵字符串>

# 业务代码
from apiboot.config import env as _env
from apiboot.utils.password_utils import encrypt_dict, decrypt_dict

PASSPHRASE = _env.get_encrypt_secret()
ENCRYPT_FIELDS = ["phone", "id_card", "address"]

# 写库前 (model → dict 后批量加密)
db.execute(
    "INSERT INTO users ...",
    encrypt_dict(
        {"name": "张三", "phone": "13800138000", "address": "北京市..."},
        fields=ENCRYPT_FIELDS,
        passphrase=PASSPHRASE,
    ),
)

# 读出后 (DB row → dict → 批量解密 → 返回前端)
row = db.fetchone("SELECT * FROM users WHERE id = %s", uid)
return decrypt_dict(dict(row), fields=ENCRYPT_FIELDS, passphrase=PASSPHRASE)

⚠️ 运维红线:

• 长度建议 ≥ 32 字符; 生成命令:

  ENCRYPT_SECRET=$(python -c "from apiboot.utils.password_utils import generate_token; print(generate_token(48))")
  

• ⚠️ 丢失 = 历史加密数据永久无法恢复, 务必多处离线备份 (密码管理工具 / 加密保险柜)
• ⚠️ 更换密钥需要批量重新加密存量数据, 务必先做灰度
• 敏感度: ⚠️ 高 (supervisor 子进程环境变量白名单不会透传)

日志(apiboot.log)

from apiboot import logger
logger.info("hello")

特性一览:

• 📁 默认 ./logs/<name>.log,按天切割,保留 7 天
• 🖥️ 控制台(stdout) + 文件双输出
• 🔒 同名 logger 幂等(多次 setup 不会重复挂 handler)
• 📍 智能锚定日志目录:显式 log_dir > .env LOG_DIR > <项目根>/logs > ./logs
• 🏷️ 智能解析 logger 名:显式 name > .env LOG_NAME > .env APP_NAME > "app"
• ⚙️ .env 中 LOG_LEVEL 配置级别(默认 INFO)
• 🚫 .env 中 LOG_ENABLE=false 关闭文件日志
• 🛡️ Surrogate 安全 — 自动清洗 LLM 流式输出里偶发的未配对 UTF-16 代理对
• 🪝 自动接管 apiboot 内部 logger — capture_internal=True 让 apiboot.config.config 等子模块的日志也走统一 handler
• 🐢 目录创建推迟到首次写入(delay=True + _LazyDirTimedRotatingFileHandler)

---

🔌 可选依赖矩阵

核心原则:默认 dependencies = [apscheduler]。业务侧任何 import 第三方包的代码都走懒加载,用户项目自己装什么版本,apiboot 就用什么版本。

业务需求	需要装的可选依赖
HTTP 客户端(同步)	requests 或 httpx
HTTP 客户端(异步)	httpx 或 aiohttp
通用 SQL 数据库	sqlalchemy + pymysql/aiomysql（MySQL）、psycopg2-binary/asyncpg（PostgreSQL）、aiosqlite（SQLite 异步）
SQLModel 兼容(旧项目)	sqlmodel
Redis	redis(3.5+ 同步 / 4.2+ 异步)
定时任务	已自带 apscheduler(硬依赖)
重试	tenacity
FastAPI 中间件	fastapi + httpx
Pydantic Schema(BaseSchema)	pydantic ≥ 2.0
Pydantic 分页(PageReq)	pydantic ≥ 1.x
LLM	langchain + langchain-core(≥ 1.0)
中文分词	jieba
OCR(MinerU)	httpx
文档解析(PDF / Office)	pymupdf / python-docx / python-pptx / openpyxl / xlrd / pandas
图片转换(PDF / PPTX → 图片)	pymupdf + 系统 LibreOffice
雪花 ID	snowflake-id
Fernet 对称加密	cryptography
深度学习(torch 设备探测)	torch
数据处理(pandas)	pandas + numpy
sklearn 工具	scikit-learn + joblib
CLI 内存检测	psutil(macOS / Windows 必需,Linux 用 /proc 不需要)

新增配置项(.env)

Key	默认	用途
SUCCESS_CODE	200	业务成功状态码(SUCCESS.code),前端用 resp.success 判断时可不关心此值
SUCCESS_MESSAGE	成功	业务成功默认文案(SUCCESS.message)
DB_MODELS_DIR	models	init_db 启动时自动 import 的模型目录(不存在静默跳过)
DB_DRIVER	后端默认	同步数据库驱动覆盖;MySQL 默认 pymysql, PG psycopg2, SQLite sqlite
DB_ASYNC_DRIVER	异步默认	异步数据库驱动;fallback 到 DB_DRIVER
TASKS_DIR	tasks	start_scheduler 启动时自动 register 的任务目录(不存在静默跳过)
REDIS_MAX_CONNECTIONS	(redis-py 默认)	Redis 连接池大小;高并发 FastAPI 建议 100~200
ENCRYPT_SECRET	(空)	数据库字段加密口令(Fernet 派生种子);通过 get_encrypt_secret() 读取,建议长度 ≥ 32 字符

---

🐍 Python 3.7 兼容性

apiboot 承诺在 Python 3.7+ 上可用 (pyproject.toml 中 requires-python = ">=3.7")。 包内部代码禁止使用 3.8+ 才引入的语法。

工具包内禁止的语法

语法	引入版本	替代写法
:= 海象	3.8 (PEP 572)	拆成普通赋值 + 判断
f"{x=}" debug	3.8 (PEP 498)	f"{x}={x}"
def f(a, /, b) positional-only	3.8 (PEP 570)	用 keyword-only def f(a, *, b)
dict[str, int] / list[int]	3.9 (PEP 585)	Dict[K, V] / List[T] (从 typing 导入)
str | None / int | str	3.10 (PEP 604)	Optional[T] / Union[T, U] (从 typing 导入)
match / case	3.10 (PEP 634)	if / elif / else
@functools.cache	3.9	@functools.lru_cache(maxsize=None)
importlib.metadata	3.8	3.7 用 pkg_resources 或 importlib_metadata 第三方包

typing 导入完整性 (硬性)

用到的 typing 名必须显式 import, 不能依赖 from __future__ import annotations 兜底:

# ✅ 正确
from typing import Any, List, Optional
def obj_pick(obj: Any, keys: List[str]) -> Any:
    ...

# ❌ 错误 (IDE 报红, 去掉 future 就 NameError)
from typing import Any
def obj_pick(obj: Any, keys: List[str]) -> Any:
    ...

通用 import 模板(满足 95% 场景):

from typing import (
    Any, Callable, Dict, List, Optional, Tuple, Type, Union,
)

给贡献者的检测脚本

python -c "
import ast, pathlib
TYPING_NAMES = {
    'Any','Callable','ClassVar','Dict','FrozenSet','Generic','Iterable',
    'Iterator','List','Literal','Mapping','NoReturn','Optional','Sequence',
    'Set','Tuple','TYPE_CHECKING','Type','TypeVar','TypedDict','Union',
    'cast','Protocol',
}
for py in pathlib.Path('apiboot').rglob('*.py'):
    if '__pycache__' in str(py): continue
    src = py.read_text()
    tree = ast.parse(src)
    imported = set()
    for n in ast.walk(tree):
        if isinstance(n, ast.ImportFrom) and n.module == 'typing':
            imported |= {a.name for a in n.names}
    used = set()
    for n in ast.walk(tree):
        if isinstance(n, ast.Name) and n.id in TYPING_NAMES: used.add(n.id)
        if isinstance(n, ast.Attribute) and isinstance(n.value, ast.Name) and n.value.id == 'typing':
            used.add(n.attr)
    missing = used - imported
    if missing: print(f'{py}: 缺失 {sorted(missing)}')
"

期望输出:空。

完整规范(包括 from __future__ 的副作用分析、Literal / TypedDict / Protocol 等 3.7 可放心用的现代语法)见 CLAUDE.md "Python 兼容性约束 (硬性)" 章节。

---

🛠️ 开发与测试

# 安装开发依赖
uv add --dev pytest mypy

# 跑测试
uv run pytest tests/ -v

# 发版流程
uv run python scripts/upload_pypi.py --repository testpypi  # 先 Test PyPI
uv run python scripts/upload_pypi.py                          # 正式 PyPI

---

📄 License

MIT © yanyue