zWebApi
一个功能丰富、开箱即用的 Python Web 框架,基于 FastAPI 构建。它旨在通过约定优于配置的原则,简化 API 开发流程,提供自动路由、统一异常处理、日志记录和可扩展工具集。
目录
特性
- 🚀** 快速启动**: 基于 FastAPI 和 Uvicorn,提供异步高性能。
- 🧭** 自动路由注册**: 只需按约定在
action/目录下组织代码,路由自动生效。 - 🔒** 路由签名强制**: 确保所有路由函数遵循统一的
query/body参数规范。 - 🛡️** 统一异常处理**: 全局捕获异常,返回格式统一的 JSON 错误响应。
- 🚨** 自定义
Panic异常**: 简单易用的自定义异常类,用于主动返回错误。 - 📋** 全面日志记录**: 自动记录应用启动、路由、异常等信息,支持文件和控制台输出,并提供日志查看接口。
- 🛠️** 可扩展工具模块**: 通过
zWebApi.tools.*轻松访问和扩展框架功能。 - 🌐** CORS 支持**: 内置 CORS 中间件,轻松配置跨域资源共享。
- 📦** 易于打包和分发**: 标准 Python 包,可通过
pip安装。
安装
pip install zWebApi
pip install zWebApi -i https://pypi.tuna.tsinghua.edu.cn/simple/
打包
python -m build
twine upload dist/*
rm -rf dist src/*.egg-info
脚本打包
./py_build.sh <env_name> build
./py_build.sh <env_name> upload <pypi api token>
./py_build.sh <env_name> delete
快速开始
1. 项目结构
创建一个符合框架约定的项目目录结构:
my_project/
├── main.py # 应用入口点
├── action/ # 路由定义目录 (必须)
│ └── user/ # 模块目录 (例如 'user')
│ └── user.py # 路由文件 (必须与模块目录同名)
├── domain/ # 业务逻辑层 (可选,但推荐)
├── dao/ # 数据访问层 (可选,但推荐)
├── utils/ # 项目通用工具 (可选)
└── zwebApi.log # (运行后自动生成) 日志文件
2. 创建应用 (main.py)
这是你应用的入口文件。
# main.py
from zWebApi import create_app
# 创建应用实例,并设置 API 标题和基础路径前缀
app = create_app(title="我的酷炫API")
# --- 可选:添加自定义中间件或配置 ---
# from fastapi.middleware import Middleware
# app.add_middleware(SomeMiddleware)
if __name__ == "__main__":
# 使用框架封装的 run 方法启动服务器
# 等效于 uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
app.run(host="0.0.0.0", port=8000, reload=True) # reload=True 适用于开发环境
3. 定义路由 (action/)
在 action/ 目录下创建模块和路由文件。
**创建 **action/user/user.py
# action/user/user.py
# 导入自定义异常
from zWebApi import zRouter, Panic
# 导入校验参数
from schema.user import (
UserQueryParams,
UserCreate,
UserResponse
)
# 必须创建一个 APIRouter 实例,变量名必须为 'router'
router = zRouter(tags=["用户管理"]) # tags 用于 API 文档分组
# --- 定义路由处理函数 ---
# 注意:函数签名必须遵循规范,只使用 'query' 和 'body' 作为参数名,
# 且它们必须是 Pydantic BaseModel 或 None。
@router.get("/info")
async def get_user_info(query: UserQueryParams = None):
"""获取用户信息"""
if query and query.user_id:
# 模拟业务逻辑
if query.user_id == 999:
# 使用 Panic 异常返回自定义错误
raise Panic(code=404, msg="用户未找到", error="请求的用户ID不存在。")
return {
"user_id": query.user_id,
"name": f"User_{query.user_id}",
"filter_used": query.name_filter
}
return {"message": "请提供 user_id 查询参数"}
@router.post("/create")
async def create_user(body: UserCreate = None):
"""创建新用户"""
if body is None:
# 使用 Panic 异常返回错误
raise Panic(code=400, msg="请求体缺失", error="创建用户必须提供请求体。")
# 模拟创建用户
new_user = UserResponse(id=123, name=body.name, age=body.age)
return {"message": "用户创建成功", "user": new_user}
在schema/目录下创建校验参数
**创建 **schema/user.py
# --- 定义 Pydantic 模型用于参数校验 ---
from pydantic import BaseModel
from typing import Optional
# 查询参数模型
class UserQueryParams(BaseModel):
user_id: Optional[int] = None
name_filter: Optional[str] = None
# 请求体模型
class UserCreate(BaseModel):
name: str
age: int
# 响应体模型 (可选但推荐)
class UserResponse(BaseModel):
id: int
name: str
age: int
4. 运行应用
在你的项目根目录 (your_project/) 下打开终端,运行:
python main.py
应用将在 http://0.0.0.0:8000 启动。
- API 根路径:
http://localhost:8000/ - 用户模块路径:
http://localhost:8000/我的酷炫api/user/... - API 文档:
http://localhost:8000/docs - 日志查看:
http://localhost:8000/我的酷炫api/api/error/logs
核心概念
应用创建与配置
使用 zWebApi.create_app 工厂函数创建 FastAPI 应用实例。
app = create_app(
title="My App", # API 标题,也用作基础路径前缀
enable_cors=True, # 是否启用 CORS
cors_origins=["*"], # CORS 允许的源
cors_allow_credentials=True,
cors_allow_methods=["*"],
cors_allow_headers=["*"]
)
路由自动注册
框架启动时会自动扫描 action/ 目录。
- 遍历
action/下的每个子目录(如user/)。 - 在每个子目录中查找与目录同名的 Python 文件(如
user.py)。 - 导入该文件,并查找名为
router的APIRouter实例。 - 将该
router挂载到以目录名(如/user)为前缀的路径下。 - 最终的基础路径是
/title(title 转为小写并用下划线替换空格)。
路由函数签名规范
为了保持一致性并利用框架的校验功能,路由处理函数 必须 遵循以下签名:
- 只接受
query和body两个命名参数。 - 参数类型 必须是 Pydantic
BaseModel的子类或None。 - 默认值 应为
None以使其成为可选参数。 - 参数名 必须严格是
query和/或body。
正确示例:
# 只有 query
async def get_items(query: ItemQueryParams = None): ...
# 只有 body
async def create_item(body: CreateItemRequest = None): ...
# 两者都有
async def update_item(query: UpdateQuery = None, body: UpdateItemRequest = None): ...
错误示例 (会导致应用启动失败):
# 错误:使用了不允许的参数名 'item_id'
async def get_item(item_id: int): ...
# 错误:没有使用 BaseModel
async def search(name: str = ""): ...
统一响应与异常处理
全局异常处理
框架自动注册了多个全局异常处理器,确保所有错误都返回统一的 JSON 格式:
{
"code": 400,
"msg": "错误的请求",
"error": "具体错误信息...",
"data": null
}
HTTPException: 处理 FastAPI/Starlette 抛出的 HTTP 异常 (如 404, 403)。RequestValidationError: 处理 Pydantic 模型校验失败 (422)。Exception: 捕获所有未处理的服务器内部错误 (500)。Panic: 处理用户自定义的Panic异常。
自定义 Panic 异常
用户可以在任何地方(路由、domain、dao)主动抛出 Panic 来返回自定义错误。
from zWebApi import Panic
# 在路由、服务或数据访问层
def some_business_logic(user_id):
if user_id <= 0:
# 主动抛出 Panic 异常
raise Panic(
code=400, # HTTP 状态码和业务码
msg="无效的用户ID", # 用户友好信息
error="用户ID必须是正整数。", # 技术错误详情
data={"provided_id": user_id} # 可选的附加数据
)
日志记录
框架使用 Python logging 模块提供全面的日志功能。
- 格式:
[级别][年月日时分秒][文件名][行号]: 消息- 例如:
[INFO][20240521180000][app.py][150]: 应用创建完成。 - 例如:
[ERROR][20240521180001][user.py][30]: 无效的用户ID
- 例如:
- 输出: 同时记录到控制台(开发)和项目根目录下的
weblog.log文件。 - 轮转: 使用
TimedRotatingFileHandler,默认每10天轮转一次日志文件。 - 查看: 提供内置 API 接口
GET /<title>/api/error/logs查看日志内容。- 可通过
?lines=N参数指定返回最后 N 行。
- 可通过
工具模块
框架提供了一个可扩展的 tools 包,用于存放通用功能模块。
导入方式:
# 从框架内置工具导入
from zWebApi.tools.db.mysql import testsql, MySQLHelper
# 未来可扩展
# from zWebApi.tools.cache.redis_client import RedisManager
创建自定义工具:
在框架源码的 src/zWebApi/tools/ 下创建新的子目录和 .py 文件即可。用户安装更新后的包即可使用。
高级用法
CORS 配置
在 create_app 时配置 CORS:
app = create_app(
title="API",
enable_cors=True,
cors_origins=["http://localhost:3000", "https://myfrontend.com "],
cors_allow_credentials=True,
cors_allow_methods=["GET", "POST", "PUT", "DELETE"],
cors_allow_headers=["*"],
)
使用框架日志
在你的项目代码中,可以使用框架配置好的日志记录器:
# 在你的 action, domain, dao 等模块中
from zWebApi import get_logger
logger = get_logger()
@router.get("/some-path")
async def my_endpoint():
logger.info("处理 /some-path 请求")
try:
# ... 业务逻辑 ...
logger.debug("业务逻辑执行成功")
return {"result": "ok"}
except Exception as e:
logger.error(f"处理请求时出错: {e}", exc_info=True)
raise # 让全局异常处理器捕获
API 文档
框架完全兼容 FastAPI 的自动生成文档功能。
- Swagger UI:
http://<your-host>:<port>/docs - ReDoc:
http://<your-host>:<port>/redoc
贡献
欢迎提交 Issue 和 Pull Request!
- Fork 本仓库
- 创建你的特性分支 (
git checkout -b feature/AmazingFeature) - 提交你的更改 (
git commit -m 'Add some AmazingFeature') - 推送到分支 (
git push origin feature/AmazingFeature) - 开启一个 Pull Request
许可证
本项目采用 MIT 许可证。详情请见 LICENSE 文件。
Metadata
Release files for KairoCore 1.6.2
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| kairocore-1.6.2.tar.gz | 3.9 MB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| kairocore-1.6.2-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 7.8 MB
Release files / kairocore-1.6.2.tar.gz
| Download URL | kairocore-1.6.2.tar.gz |
|---|---|
| Size | 3.9 MB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
907bd1f6ffa72134b1685960302172cb5c4c4b5e33fc2e0c363cce5a12e38922
|
|
BLAKE2b-256 checksum How to use checksums |
73e8a68056021e7c221a2b1308fe7be90dcad443a58b3562f98a199438a8cb65
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.1.0 CPython/3.12.11
|
Release files / kairocore-1.6.2-py3-none-any.whl
| Download URL | kairocore-1.6.2-py3-none-any.whl |
|---|---|
| Size | 3.9 MB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
0ecc583dbb4b6d58b34746f3580f73fc544aae5c7b6d5dc8198795e74ccf24a8
|
|
BLAKE2b-256 checksum How to use checksums |
2f76fb9e61bed715301d2b9fdc1852e9cc1964c76a3d7a19188eeba56af9f0c2
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/6.1.0 CPython/3.12.11
|