apiboot

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

• 零依赖: dependencies = [], 所有第三方包 (httpx / pydantic / sqlalchemy / langchain / snowflake-id / PyMuPDF / APScheduler / redis / psutil) 都是可选, 用户项目自己 pip install 才用
• 懒加载: import apiboot 不触发任何第三方包加载, 真正调用时才探测
• Python 3.7+ 兼容: 不使用 PEP 604 (X | None)、PEP 585 (list[int])、海象、match/case 等 3.8+ 语法
• FastAPI 友好: 提供统一响应中间件、错误体系、Schema 基类、CLI 守护进程管理

安装

pip install apiboot

按需安装可选依赖:

# HTTP / FastAPI 中间件
pip install httpx

# ORM / Schema
pip install sqlalchemy pydantic

# LLM
pip install langchain langchain-core

# OCR / 文档解析
pip install pymupdf

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

快速上手

1. 统一响应中间件

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

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

路由返回值自动包成 {"code": 200, "message": "成功", "data": {...}}。 异常自动捕获并包装成业务响应 (HTTP 200 + body code 区分业务/系统错误)。

2. 统一错误体系

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

raise ApiError("uid 不能为空")                  # code = 20000 (默认)
raise ApiError("用户不存在", code=NOT_FOUND.code) # code = 12001
raise SystemError("数据库连接失败")               # code = 10099

3. Schema 基类 (驼峰自动转换)

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"}

populate_by_name=True 让前端 userId 和后端 user_id 双向都接受。

4. 流式响应队列 (SSE / WebSocket)

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

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")

5. .env 配置读取

from apiboot import config

print(config.PYPI_API_TOKEN)  # str
print(config.DB_PORT)         # int (自动类型转换)

支持多环境 (.env.development / .env.production), 自动类型转换 (str → int / bool / float)。

6. CLI 守护进程管理

abt start myproject      # 启动 (后台守护进程)
abt status               # 查看所有项目状态
abt stop myproject       # 优雅停止 (SIGTERM, 等 10s 兜底 SIGKILL)
abt restart myproject    # 重启
abt log myproject        # 实时 tail 日志
abt list                 # 列出所有注册项目

模块清单

开发

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

# 跑测试
uv run pytest tests/ -v

# 发版
uv run python scripts/upload_pypi.py --repository testpypi  # 先测试
uv run python scripts/upload_pypi.py                          # 正式

License

MIT