开始之前
- 私有化部署(有源码)
- 熟悉 Python 与 FastAPI
- 本地能跑起来一套环境
操作步骤
先认清五个目录
app/api/ 放路由(只做参数校验与编排);app/schemas/ 放 Pydantic 模型(请求/响应结构);app/crud/ 放数据库读写;app/models/ 放 SQLAlchemy 表定义;app/services/ 放业务逻辑(跨表、有规则的都放这里)。路由里写复杂业务是最常见的坏味道。
响应必须走统一信封
所有接口返回 app/core/response.py 的 ok(data, msg) 或 fail(code, msg)。自己 return dict 会破坏前端统一拦截逻辑。抛业务错误用 BizError(code, msg),需要限流语义用 RateLimitExceeded。
多租户是可选项,不是可忘项
所有业务表的查询必须带 tenant_id 过滤(从 get_current_user 的 user.tenant_id 取)。漏一个 tenant_id 就是跨租户数据泄露,这不是性能问题,是安全问题。写新查询时建议直接看同目录现有实现的过滤方式。
加接口的标准五步
① app/schemas/ 加请求/响应模型;② app/crud/ 加数据访问函数(带 tenant_id);③ app/services/ 加业务逻辑(如果需要);④ app/api/admin/xxx/router.py 加路由,记得挂权限依赖;⑤ 在 app/api/router.py 确认已 include_router。少任何一步都会出现「接口有了但 404」或「能调但没有权限校验」。
加字段必须先写迁移
直接改 models 不改库,程序启动就报 UndefinedColumnError。正确流程:改 models → 生成迁移 → 检查迁移脚本 → 执行 upgrade。详见下一篇「数据库迁移与版本升级」。
改完必须自己跑一遍三条命令
语法与导入检查、启动服务、用真实 token 调一次新接口。不要只跑单测就宣布完成 —— 这个项目的接口依赖真实租户上下文,很多问题只在真实请求下出现。
代码示例
# 1) app/schemas/workshop_stat.py
from pydantic import BaseModel
class WorkshopStatOut(BaseModel):
workshop: str
qty_ok: int
qty_ng: int
# 2) app/crud/workshop_stat.py
from sqlalchemy import select, func
from sqlalchemy.orm import Session
from app.models.report import ReportRecord
def stat_by_workshop(db: Session, tenant_id: int, days: int = 7):
stmt = (
select(ReportRecord.workshop, func.sum(ReportRecord.qty_ok), func.sum(ReportRecord.qty_ng))
.where(ReportRecord.tenant_id == tenant_id) # 租户过滤不能漏
.group_by(ReportRecord.workshop)
)
return db.execute(stmt).all()
# 3) app/api/admin/workshop_stat/router.py
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from app.core.deps import get_current_user, get_db
from app.core.response import ok
from app.crud.workshop_stat import stat_by_workshop
from app.models.user import User
router = APIRouter()
@router.get("/workshop-stat")
def workshop_stat(
days: int = Query(default=7, ge=1, le=90),
db: Session = Depends(get_db),
user: User = Depends(get_current_user),
):
rows = stat_by_workshop(db, user.tenant_id, days)
return ok({"items": [
{"workshop": w or "-", "qty_ok": int(okq or 0), "qty_ng": int(ng or 0)}
for w, okq, ng in rows
]})
# 4) app/api/router.py 挂载
# from app.api.admin.workshop_stat.router import router as workshop_stat_router
# api_router.include_router(workshop_stat_router,
# prefix="/admin/workshop-stat",
# tags=["admin-workshop"],
# dependencies=_admin_deps)cd /www/wwwroot/lightmes/backend
# 1) 导入与语法
./venv/bin/python -c "import app.main; print('import ok')"
# 2) 重启(按你们的进程管理方式)
supervisorctl restart lightmes-backend # 或 systemctl / 宝塔进程守护
# 3) 用真实 token 调一次
curl -s "http://127.0.0.1:8000/api/admin/workshop-stat?days=7" \
-H "Authorization: Bearer $TOKEN" | head -c 300示例中的域名、租户编码、密钥请替换成你自己的。生产环境不要把密钥写进前端代码。