后端

用 FastAPI 从零搭建一个企业内部工单系统

2026-07-21 · 约 8 分钟

最近给公司内部 IT 运维场景搭了一套工单系统，需求并不复杂：工单的创建、流转、分配、状态变更，再加一个简单的用户与权限管理。在技术选型上我几乎没有犹豫，直接用了 FastAPI。这篇笔记记录一下整套架构是怎么从零搭起来的，重点放在分层、校验和鉴权这三块。

为什么选 FastAPI

内部系统的特点是迭代快、文档要求高、并发不高但接口多。FastAPI 几乎是为这种场景量身定做的，主要因为三点：

原生异步：基于 Starlette，async def 的写法天然支持高 IO 场景，工单系统里大量操作是查数据库、调外部 API，异步收益明显。
自动生成 OpenAPI 文档：路由写完，/docs 就有一份可交互的 Swagger UI，给前端联调、给运维同学对接省了巨多沟通成本。这是内部系统最香的一点。
Pydantic 校验：请求体的类型、必填、取值范围全部声明式地写在模型里，框架自动校验并返回结构化错误，不用手写一堆 if not field。

项目结构分层

对于会持续迭代的服务，我习惯一开始就分清楚层。一个工单系统大概的结构是这样：

app/
├── main.py              # FastAPI 实例、启动事件、路由挂载
├── routers/             # 路由层：只负责接收请求、调用 service
│   ├── tickets.py
│   └── auth.py
├── services/            # 业务逻辑层：编排 DAO，写真正的业务规则
│   └── ticket_service.py
├── models/              # Pydantic 模型：请求/响应 schema
│   └── ticket.py
├── dao/                 # 数据访问层：SQL 操作封装在这里
│   └── ticket_dao.py
├── deps.py              # 依赖注入：鉴权、数据库连接
└── config.py            # 配置项

核心原则是路由薄、业务厚。routers 里不写 SQL，dao 里不写业务判断，services 负责把两者串起来。这样后续换数据库或者改接口形态，改动都能收敛在一层里。

用 Pydantic 做请求体校验

工单创建接口是最典型的场景。一个工单至少要校验标题、优先级、分类、发起人这几个字段。直接把校验逻辑写在模型里，路由就非常干净：

from pydantic import BaseModel, Field, field_validator
from enum import Enum
from datetime import datetime


class Priority(str, Enum):
    low = "low"
    medium = "medium"
    high = "high"
    urgent = "urgent"


class TicketCreate(BaseModel):
    title: str = Field(..., min_length=2, max_length=100,
                         description="工单标题")
    description: str = Field("", max_length=2000)
    priority: Priority = Priority.medium
    category: str = Field(..., min_length=1, max_length=32)
    assignee_id: int | None = None

    @field_validator("title")
    @classmethod
    def strip_title(cls, v: str) -> str:
        # 标题前后去空格，避免全空格的脏数据
        v = v.strip()
        if not v:
            raise ValueError("标题不能为空")
        return v


class TicketOut(BaseModel):
    id: int
    title: str
    priority: Priority
    status: str
    created_at: datetime

路由里几乎不写校验代码，框架会自动把不合规的请求拦下来，并返回一个 422 + 字段级别的错误明细。前端拿到的就是结构化 JSON，直接渲染到表单就行：

@router.post("/tickets", response_model=TicketOut)
async def create_ticket(
    payload: TicketCreate,
    current_user: User = Depends(get_current_user),
):
    # 校验已由 Pydantic 完成，这里只关心业务
    ticket = await ticket_service.create(payload, creator=current_user)
    return ticket

JWT 鉴权的依赖注入

内部系统一般用 JWT 做无状态鉴权。FastAPI 的依赖注入系统（Depends）让鉴权写起来非常优雅——把"取当前用户"封装成一个依赖，任何需要登录的接口挂上去就行。

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
SECRET_KEY = "change-me-in-prod"
ALGORITHM = "HS256"


async def get_current_user(
    token: str = Depends(oauth2_scheme),
) -> User:
    # 任何路由只要 Depends(get_current_user) 就自带鉴权
    credentials_exc = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str | None = payload.get("sub")
        if user_id is None:
            raise credentials_exc
    except jwt.PyJWTError:
        raise credentials_exc

    user = await user_dao.get_by_id(int(user_id))
    if user is None:
        raise credentials_exc
    return user


async def require_admin(
    current_user: User = Depends(get_current_user),
) -> User:
    # 在 get_current_user 基础上叠加权限校验
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return current_user

使用的时候只要在路由签名上加一个参数：current_user: User = Depends(get_current_user)，鉴权就生效了。需要管理员权限的接口换成 Depends(require_admin)。依赖可以层层嵌套，这是 FastAPI 我觉得最舒服的设计之一。

SQLite + 简单的 DAO 封装

内部系统流量不大，没必要一上来就 Postgres。我直接用了 SQLite，配 WAL 模式后并发读完全够用（这块下一篇文章会单独讲）。数据库访问统一封装到 DAO 层，业务层只面对方法：

from contextlib import asynccontextmanager
import aiosqlite


class TicketDAO:
    def __init__(self, db: aiosqlite.Connection):
        self.db = db

    async def create(self, title: str, priority: str,
                    creator_id: int) -> dict:
        cursor = await self.db.execute(
            """INSERT INTO tickets (title, priority, status, creator_id, created_at)
               VALUES (?, ?, 'open', ?, datetime('now'))""",
            (title, priority, creator_id),
        )
        await self.db.commit()
        return {"id": cursor.lastrowid, "title": title,
                "priority": priority, "status": "open"}

    async def list_open(self, limit=50) -> list[dict]:
        cursor = await self.db.execute(
            "SELECT * FROM tickets WHERE status != 'closed' "
            "ORDER BY created_at DESC LIMIT ?", (limit,))
        return [dict(r) for r in await cursor.fetchall()]


# 用依赖注入把连接分发给每个请求
async def get_dao() -> TicketDAO:
    async with db_session() as conn:
        yield TicketDAO(conn)

业务层只调 dao.create(...)，完全看不到 SQL 字符串。后面如果换 Postgres，只要重写一个 DAO 实现，路由和 service 一行不用改。这就是分层带来的红利。

小提示：SQLite 在异步框架里要用 aiosqlite，别直接用同步的 sqlite3，否则会阻塞整个事件循环，高 IO 时表现会非常糟。

小结

用 FastAPI 搭这套工单系统，整体感受是开发节奏非常快。Pydantic 把校验从业务里剥离，依赖注入把鉴权和数据库连接变成可组合的零件，自动文档让联调成本几乎为零。分层结构（routers / services / dao）让代码边界清晰，半年后回来改需求也不至于一脸懵。

当然它也不是银弹：异步生态里部分第三方库还不完善，遇到阻塞调用要手动 run_in_executor。但对一个内部系统来说，这套组合已经足够稳、足够好维护了。下一篇我会单独聊聊 SQLite 在这个系统里是怎么撑住并发写入的。

← 返回文章列表