后端源码二次开发:加接口、加字段、加迁移

FastAPI + SQLAlchemy 2.0 + Alembic 的分层结构。改之前先搞清五个目录各自负责什么,否则改一处崩三处。

学完能做到 能安全地加一个新接口和一个新字段,并写出可回滚的迁移。

开始之前

  • 私有化部署(有源码)
  • 熟悉 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 调一次新接口。不要只跑单测就宣布完成 —— 这个项目的接口依赖真实租户上下文,很多问题只在真实请求下出现。

代码示例

python加一个自定义接口的完整链路
# 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)
bash改完三项检查
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

示例中的域名、租户编码、密钥请替换成你自己的。生产环境不要把密钥写进前端代码。

容易踩的坑

改了代码但接口不生效?
看服务是否真的重启了。uvicorn 带 --reload 才热加载,生产环境一般不带。另外确认 router.py 里 include_router 的 prefix 与你请求的路径一致。
查询慢,一眼看不出原因?
先确认有没有索引(tenant_id + 时间字段 + 业务外键)。再检查是不是 N+1(循环里查关联表)。本项目默认懒加载,序列化时容易触发 MissingGreenlet 或大量查询。
改了 models 但忘了迁移,报 UndefinedColumnError?
这个错误就是字面意思:代码要一个不存在的列。生成并执行迁移即可。反过来(加了列没改代码)不会报错,但会留下无用的列,长期会混乱。
能不能直接改数据库表结构图省事?
不能。下次执行 alembic upgrade 时会因为状态不一致失败或被覆盖。所有结构变更必须走迁移脚本。

接着看

还有别的问题?

先看常见问题,或者在系统内提工单,我们会回。