apiboot

为组件化构建 API 服务而生的 Python 底层工具包

apiboot 是一个零依赖、按需懒加载的 Python 工具包，专为 FastAPI / 异步服务 / 定时任务 / 数据持久化 / LLM 集成等典型后端场景封装常用基座。它不绑定业务代码，不带任何强制的运行时依赖——你装什么版本，它就用什么版本。

• PyPI: https://pypi.org/project/apiboot/
• GitHub: https://github.com/yanyue/apiboot
• Python: 3.7+（硬约束，源码不使用 3.8+ 独有语法）

---

目录

• 核心特性
• 安装
• 快速上手
• CLI 守护进程管理
• 模块清单
• 可选依赖矩阵
• 开发与测试

---

核心特性

特性	说明
零运行时依赖	dependencies = []，pip install apiboot 不会拖任何第三方包
懒加载第三方包	业务用到的依赖，按需懒加载；没装就不触发 import，环境干净
Python 3.7+ 兼容	不使用 PEP 604（X | None）、PEP 585（list[int]）、海象、match/case 等 3.8+ 语法
同步 + 异步双 API	MySQL / Redis / 定时任务 都提供 sync/async 双版本，按函数名前缀（a 前缀）选用
装饰器自动注册	定时任务 @scheduled_job(...) 写完即生效，无需手动扫描路径
FastAPI 友好	统一响应中间件、错误体系、Schema 基类、SSE 流式队列、CLI 守护进程管理一应俱全
OOM 守护	CLI 内置 supervisor，OOM 自动重启 + 雪崩保护
类型注解友好	支持 pydantic v1/v2、SQLAlchemy DeclarativeBase、SQLModel、dataclass 自动识别

---

安装

pip install apiboot

按需安装可选依赖（只装你用得到的部分）：

# HTTP 客户端（FastAPI 中间件 / 通用 HTTP 调用）
pip install httpx

# ORM / Schema
pip install sqlalchemy pydantic

# MySQL / Redis / 调度
pip install sqlalchemy pymysql redis apscheduler

# LLM
pip install langchain langchain-core

# OCR / 文档解析
pip install pymupdf httpx

# 重试
pip install tenacity

---

快速上手

1. 一行启动：日志 + 配置

from apiboot import config, logger

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

1b. 路径类配置（相对路径 → 绝对路径）

from apiboot import config

# .env:
#   MODEL_DIR = "./models"
#   CACHE_DIR = "~/cache"

model_path = config.path("MODEL_DIR") / "rnn.pt"
# → PosixPath('/abs/path/to/your-project/models/rnn.pt')    ← 默认锚到项目根

cache_path = config.path("CACHE_DIR")
# → PosixPath('/Users/xxx/cache')    # 自动 expanduser

解析规则:

.env 值	行为
"./models" / "models" / "../shared"	按锚点解析（默认项目根，见 APIBOOT_PATH_ANCHOR）
"/abs/path"	原样返回
"~/cache"	expanduser 后返回
未配置 / 空值	抛 KeyError

锚点策略通过 APIBOOT_PATH_ANCHOR 配置切换:

值	锚点	适合场景
project_root (默认)	_find_project_root() 找到的项目根	跨脚本/CWD/CI 一致
env_dir	key 所在 .env 的父目录	嵌套 monorepo 子项目；多环境 .env.prod/.env.test 自动跟随

切换示例:

# .env 或 .env.prod: APIBOOT_PATH_ANCHOR = env_dir
# examples/nlp/03_lstm/.env: DATA_DIR = ./datas
# → 解析为 examples/nlp/03_lstm/datas

显式换锚点：config.path("MODEL_DIR", base=Path("/tmp")) / "x.pt"。

2. FastAPI 统一响应中间件

from fastapi import FastAPI
from apiboot.middlewares import JsonResponseMiddleware, ReqResLoggingMiddleware

app = FastAPI()
app.add_middleware(ReqResLoggingMiddleware)   # 日志外层
app.add_middleware(JsonResponseMiddleware)    # 响应包装内层

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

3. 统一错误体系

from apiboot.error import BusinessError, SystemError
from apiboot.error.code import NOT_FOUND

raise BusinessError("uid 不能为空")                  # code = 20000（业务异常，自动包装成 HTTP 200）
raise SystemError("数据库连接失败")                   # code = 10099（系统异常，HTTP 500）

class UserNotFoundError(BusinessError):
    default_code = NOT_FOUND.code                    # code = 12001

raise UserNotFoundError("用户不存在")

4. Schema 基类（snake_case ↔ camelCase 自动转换）

from apiboot.schemas.base import 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"}
u.model_dump(by_alias=False)   # {"user_id": 1, "user_name": "alice"}

5. MySQL CRUD（sync + async 双 API，26 个函数）

from apiboot.db.mysql import init_db, get_db, save, query_page
# 异步版: ainit_db / aget_db / asave / aquery_page

init_db()                           # 启动时建表（自动扫 SQLAlchemy / SQLModel metadata）
with get_db() as session:
    save(User(name="张三"), session=session)
    users, total = query_page(User, page_num=1, page_size=20, session=session)

6. Redis KV（sync + async 双 API）

from apiboot.db.redis import init_redis, set_json, get_json
# 异步版: ainit_redis / aset_json / aget_json

init_redis()
set_json("user:1", {"name": "张三"}, ttl=3600)
print(get_json("user:1"))     # {'name': '张三'}

7. 定时任务（装饰器自动注册）

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 contextlib import asynccontextmanager
from fastapi import FastAPI
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: FastAPI):
    await astart_scheduler()
    yield
    await astop_scheduler()

app = FastAPI(lifespan=lifespan)

8. LLM 接入（LangChain chat model 一键构造）

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)

9. OCR / 文档解析（MinerU API）

import asyncio
from apiboot.ocr import MinerU

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

10. SSE / WebSocket 流式队列

from fastapi.responses import StreamingResponse
from apiboot.utils.queue_utils import StreamQueue

channel = StreamQueue()

async def background():
    try:
        for chunk in llm.stream(prompt):
            await channel.put({"content": chunk})
    finally:
        channel.close()     # 推哨兵，异常路径也会触发

asyncio.create_task(background())

@app.get("/stream")
async def stream():
    return StreamingResponse(channel.iter_sse(), media_type="text/event-stream")

---

CLI 守护进程管理

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

cd /path/to/your-project    # 有 main.py + .env 的目录
abtstart                     # 启动服务（后台）
abtstatus                    # 查看状态
abtstop                      # 停止服务
abtrestart                   # 重启
abtlist                      # 列出所有 abtstart 启动的项目
abtlog                       # tail -f .abt.log

也可以用子命令形式 abt start / stop / restart / status / list / log。

常用参数

abtstart --module api.main:app      # 自定义入口模块
abtstart --host 127.0.0.1 --port 9000
abtstart --workers 4                 # 多 worker 进程（uvicorn multiprocessing）
abtstart --reload                    # 开发模式自动 reload
abtstart --env production            # 加载 .env.production
abtstart --clean                     # 端口被占时强杀清理
abtstart --no-supervisor             # 关闭 OOM 守护，回到旧行为

关键能力

• 自动发现：自动找 .env + main.py / app.py / server.py
• 多环境：APP_ENV=production 自动加载 .env.production
• 端口预检：启动前检测端口占用，给出明确提示（--clean 可强杀）
• OOM 守护（默认）：子进程被 OOM Killer 杀，supervisor 自动重启
• 雪崩保护：短窗口内重启次数超限，自动放弃（防内存泄漏反复重启）
• 多 worker：通过 abtstart --workers N 把进程模型切成 uvicorn multiprocessing
• 全局 registry：abt list 一台机器混着多个项目也清晰可查
• 配置文件热加载：uvicorn 不重启不生效，但 .env 重新读

workers 与 OOM 边界（必读）

--workers N 模式下，supervisor 的 RSS 防御只对 uvicorn master 可见，看不到后代 worker 的内存（/proc/<master_pid>/status 只算自己）。

• 单个 worker 被 OOM Killer 杀 → uvicorn master 自己重启该 worker（supervisor 无感知）
• 整套（master + 全部 workers）同归于尽 → supervisor 检测到 master 死，重启整套
• worker 内存慢慢爬升没到 OOM Kill 阈值 → 没人管，靠 cgroup / K8s / systemd 内存限制兜底

真要把每个 worker 内存也管起来，用 systemd MemoryMax= 或 K8s resources.limits.memory，不要指望本 supervisor。

---

模块清单

顶层快捷入口

符号	说明
apiboot.config	一行导入 .env 配置（自动类型转换）
apiboot.logger	一行拿到 logger（控制台 + 文件双输出，按天切割）

CLI（apiboot.cli）

子模块	功能
cli/__init__.py	abtstart / abtstop / abtrestart / abtstatus / abtlist / abtlog 实现
cli/supervisor.py	OOM 守护 + 雪崩保护 + RSS 检测
cli/registry.py	~/.abt/registry.json 多项目注册表（fcntl 文件锁）
cli/server.py	端口预检 / PID 文件 / main.py 自动发现
cli/env.py	多环境 .env 系列文件加载
cli/memory.py	内存上限探测（cgroup / 物理机）

FastAPI 中间件（apiboot.middlewares）

组件	功能
JsonResponseMiddleware	路由返回值 + 异常统一包装成 {code, message, data}
ReqResLoggingMiddleware	请求/响应日志（每个请求两行：进入 / 离开）

特性：

• 路由返回值 / ApiError / HTTPException / 未预期异常 → 自动包装
• SSE / 流式接口自动跳过（识别 StreamingResponse）或通过 exclude_paths 显式排除
• 非 JSON 响应（文件下载、HTML 页面）原样放行
• 已包装过的响应（顶层已有 code/message/data）原样放行，不重复包装
• 业务错误 HTTP 200（靠 body 的 code 区分），系统错误 HTTP 500
• Datetime 自动格式化为 "%Y-%m-%d %H:%M:%S"

错误体系（apiboot.error）

类 / 码	说明
ApiError / BusinessError	业务侧异常（HTTP 200，code ≥ 10000）
HttpError	外部 HTTP 接口调用失败
SystemError / ApiSystemError	系统侧异常（HTTP 500，code ≥ 10099）
ErrorCode	错误码常量工厂

Schema（apiboot.schemas）

类	功能
BaseSchema	pydantic Schema 基类，自动 snake_case ↔ camelCase（pydantic ≥ 2.0）
JsonResult[T]	统一 API 响应格式 {"code": 0, "message": "success", "data": ...}
PageReq	FastAPI 分页请求基类（page_num ≥ 1, page_size 1–500）

数据库（apiboot.db）

子模块	功能
db.mysql	MySQL sync + async 双 API，26 个 CRUD 函数
db.redis	Redis sync + async 双 API，支持单节点 + 集群
db._base	get_db_prefix / get_redis_prefix 表名/key 前缀管理

MySQL 26 个 CRUD 函数（sync + async 一一对应）：

类别	函数
增	save / save_batch
改	update / update_batch / save_or_update / save_or_update_batch（MySQL ON DUPLICATE KEY）
查	query / query_primary_key / query_page（rows + total）
删	delete / delete_batch / soft_delete / soft_delete_batch（UPDATE deleted_at = NOW()）

Redis API 摘要：

init_redis / close_redis / get_redis
set_value / get_value / delete / exists / expire / mget / set_json / get_json / ping
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（自动启用）。

定时任务（apiboot.cron）

API	说明
同步：scheduled_job / start_scheduler / stop_scheduler / get_scheduler	def 程序用
异步：async_scheduled_job / astart_scheduler / astop_scheduler	async def 程序用
约定扫描：register_jobs_from_dir / register_jobs_from_files	与传统 def register(scheduler) 风格兼容

核心改进：装饰器在模块顶层求值时自动注册，业务方不用手动调用 register_jobs_from_dir("task")；start_scheduler() / astart_scheduler() 无需任何参数即可启动。

LLM / Agent（apiboot.llm）

模块	功能
llm.init_model	LangChain chat model 一键构造（自动派发 provider：openai / deepseek / anthropic）
llm.agent.ChatAgent	流式聊天 Agent 封装（适配 FastAPI StreamingResponse）

OCR（apiboot.ocr）

类 / 函数	功能
MinerU	MinerU API 异步客户端（PDF / 图片 / Office → markdown）
MinerUAPIError	MinerU 调用失败的异常
mineru_is_supported	判断文件扩展名是否被 MinerU 支持

支持格式：.pdf / .jpg / .jpeg / .png / .gif / .bmp / .webp / .tiff / .xlsx / .docx / .pptx / .ofd（.xls 不支持）。

后端可配：MINERU_BACKEND=hybrid-engine（默认，平衡） / vlm-engine（精度高，速度慢）。

重试（apiboot.retry）

Tenacity 的薄封装 + 懒加载再导出，行为完全等价：

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 / get_file_type / get_file_name / get_file_stem
对象/字典	obj_to_dict / dict_to_obj / obj_is_null / obj_is_not_null
字符串	str_len / str_is_empty / str_is_blank / str_preview
日期时间	now / today / yesterday / timestamp / timestamp_ms / format_time / parse_time / add_days 等
JSON	to_json_str / from_json_str / load_json_file / save_json_file
列表	list_union / list_intersection / list_difference / list_symmetric_difference
数学	add / sub / mul / div / round_to / percent / random_int / random_float
依赖探测	get_installed_version

需要第三方包的子模块（按需从子模块导入）：

子模块	功能	可选依赖
utils.http_utils	同步 + 异步 HTTP 客户端（连接池复用、自动重试、统一 HttpError）	requests / httpx / aiohttp
utils.queue_utils.StreamQueue	异步流式队列（SSE / WebSocket，哨兵结束）	stdlib（asyncio）
utils.obj_utils	对象 ↔ dict 互转（pydantic / SQLAlchemy / sqlmodel / dataclass 自动识别）	pydantic / sqlalchemy / sqlmodel
utils.image_utils	PDF / PPTX → 图片（PyMuPDF + LibreOffice）	pymupdf + 系统 LibreOffice（PPTX 必需）
utils.poi_utils	PDF / PPTX / DOCX / Excel → 纯文本	pymupdf / python-docx / python-pptx / openpyxl / xlrd / pandas
utils.snowflake_utils	雪花 ID 生成器（线程安全全局单例）	snowflake-id
utils.base_utils	懒加载第三方模块工具（require_module / _get_optional_module / get_installed_version）	stdlib

日志（apiboot.log）

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

特性：

• 默认 ./logs/<name>.log，按天切割，保留 7 天
• 控制台（stdout）+ 文件双输出
• 同名 logger 幂等（多次 setup 不会重复挂 handler）
• 通过 .env 中 LOG_LEVEL 配置级别（默认 INFO）
• 通过 .env 中 LOG_ENABLE=false 关闭文件日志（仅保留控制台）
• Surrogate 安全：自动清洗 LLM 流式输出里偶发的未配对 UTF-16 代理对，避免 UnicodeEncodeError 静默吞日志 / 中断 SSE 流
• 目录创建推迟到首次写入（delay=True + 自定义 _LazyDirTimedRotatingFileHandler）

高级用法：

from apiboot.log.logger import setup_logger
logger = setup_logger("my_app", level=logging.DEBUG)

---

可选依赖矩阵

核心原则：dependencies = []。任何 import 第三方包的代码都走懒加载，没装就不触发。用户项目自己装什么版本，apiboot 就用什么版本。

业务需求	需要装的可选依赖
HTTP 客户端（同步）	requests 或 httpx
HTTP 客户端（异步）	httpx 或 aiohttp
MySQL ORM	sqlalchemy + pymysql（同步）/ aiomysql 或 asyncmy（异步）
SQLModel 兼容（旧项目）	sqlmodel
Redis	redis（3.5+ 同步 / 4.2+ 异步）
定时任务	apscheduler
重试	tenacity
FastAPI 中间件	fastapi + httpx
Pydantic Schema	pydantic ≥ 2.0（BaseSchema）/ 1.x / 2.x 都可以（PageReq）
LLM	langchain + langchain-core（≥ 1.0）
OCR（MinerU）	httpx
文档解析（PDF / Office）	pymupdf / python-docx / python-pptx / openpyxl / xlrd / pandas
图片转换（PDF / PPTX → 图片）	pymupdf + 系统 LibreOffice
雪花 ID	snowflake-id
CLI 内存检测	psutil（macOS / Windows 必需，Linux 用 /proc 不需要）

---

开发与测试

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

# 跑测试
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