《FastAPI 小白 10 天实战教程（二）：项目拆分版，按企业项目结构重构》

zou54188666

410人浏览 · 2026-03-28 18:19:06

zou54188666 · 2026-03-28 18:19:06 发布

这一版不是教你“再写一个新项目”，而是教你把上一版的 单文件 FastAPI 项目，重构成更像企业项目的结构。

FastAPI 官方专门提供了 Bigger Applications - Multiple Files 章节，核心思路就是用 APIRouter 按模块拆分，把接口、依赖、数据库、配置分别放到不同文件里。APIRouter 本身就是 FastAPI 用来“分组路由、拆分多文件”的官方推荐方式。

一、这一版你会学到什么

你会把上一版的单文件项目，重构成这种企业常见结构：

main.py：应用入口
core/：配置、安全
db/：数据库连接
models/：数据库模型
schemas/：Pydantic 输入输出模型
api/routers/：路由模块
deps/：依赖函数
services/：业务逻辑
uploads/：上传文件目录

这样做的好处是：

文件不会越来越乱
用户、文档、鉴权逻辑能分开
后续加功能更容易
更接近公司项目写法

FastAPI 官方也明确说了：小项目可以一个文件，但真实 Web API 很少能一直塞在同一个文件里，所以提供了多文件组织方式。

二、先理解：为什么要拆分

先看你上一版的单文件项目，里面有：

路由
数据库配置
SQLAlchemy 模型
Pydantic 模型
JWT 认证
上传逻辑
中间件

这些全写在 main.py 里，刚开始很方便，但功能一多就会出现：

一眼看不到重点
修改一个功能容易影响别的地方
同事协作容易冲突
排错困难

所以企业项目一般会按“职责”拆。

你可以先记一句：

路由管入口，service 管业务，model 管表结构，schema 管接口数据格式，deps 管公共依赖，core 管配置和安全。

三、最终目录结构

我们用一个适合小白、又接近企业项目的版本：

document-center/
├── app/
│   ├── main.py
│   ├── core/
│   │   ├── config.py
│   │   └── security.py
│   ├── db/
│   │   ├── base.py
│   │   └── session.py
│   ├── models/
│   │   ├── user.py
│   │   └── document.py
│   ├── schemas/
│   │   ├── user.py
│   │   ├── auth.py
│   │   └── document.py
│   ├── deps/
│   │   └── auth.py
│   ├── services/
│   │   └── document_service.py
│   └── api/
│       └── routers/
│           ├── user.py
│           ├── auth.py
│           └── document.py
├── uploads/
├── test.db
└── requirements.txt

这套结构不是唯一标准，但很适合你当前阶段。

四、重构前先装依赖

还是用上一版那套可运行依赖：

pip install fastapi uvicorn sqlalchemy python-jose passlib[bcrypt] python-multipart

如果你想把依赖保存下来：

pip freeze > requirements.txt

五、先创建目录和文件

在项目根目录执行：

mkdir -p app/core app/db app/models app/schemas app/deps app/services app/api/routers uploads

然后创建这些文件：

touch app/main.py
touch app/core/config.py
touch app/core/security.py
touch app/db/base.py
touch app/db/session.py
touch app/models/user.py
touch app/models/document.py
touch app/schemas/user.py
touch app/schemas/auth.py
touch app/schemas/document.py
touch app/deps/auth.py
touch app/services/document_service.py
touch app/api/routers/user.py
touch app/api/routers/auth.py
touch app/api/routers/document.py

如果你是 Windows，没有 touch，就手动创建空文件。

六、先写数据库层

1）`app/db/session.py`

这个文件只负责数据库连接和 Session。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

2）`app/db/base.py`

这个文件只放 Base。

from sqlalchemy.orm import declarative_base

Base = declarative_base()

七、写数据库模型 models

企业项目里，数据库表结构通常单独放到 models/。

1）`app/models/user.py`

from sqlalchemy import Column, Integer, String
from app.db.base import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    password = Column(String, nullable=False)
    age = Column(Integer, nullable=False)

2）`app/models/document.py`

from sqlalchemy import Column, Integer, String
from app.db.base import Base

class Document(Base):
    __tablename__ = "documents"

    id = Column(Integer, primary_key=True, index=True)
    filename = Column(String, index=True, nullable=False)
    filepath = Column(String, nullable=False)
    owner_id = Column(Integer, nullable=False)
    status = Column(String, default="processing")

八、写接口数据模型 schemas

这一层很重要。

你要记住：

models 是数据库表结构
schemas 是接口输入输出结构

这两者不要混。

1）`app/schemas/user.py`

from pydantic import BaseModel, Field

class UserCreate(BaseModel):
    username: str = Field(min_length=2, max_length=20)
    password: str = Field(min_length=6, max_length=50)
    age: int = Field(ge=1, le=120)

class UserPublic(BaseModel):
    id: int
    username: str
    age: int

    class Config:
        from_attributes = True

2）`app/schemas/auth.py`

from pydantic import BaseModel

class Token(BaseModel):
    access_token: str
    token_type: str

3）`app/schemas/document.py`

from pydantic import BaseModel

class DocumentPublic(BaseModel):
    id: int
    filename: str
    filepath: str
    owner_id: int
    status: str

    class Config:
        from_attributes = True

九、写核心配置 core

1）`app/core/config.py`

这个文件先放基础配置。
后面你再升级成 .env 读取版。

FastAPI 官方高级指南里专门有 Settings / Environment Variables，推荐用 Pydantic Settings 处理配置，并可结合 .env 文件。当前你先写死，后面再升级。

SECRET_KEY = "your_secret_key_123456"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60
UPLOAD_DIR = "uploads"

2）`app/core/security.py`

这个文件只处理密码和 token。

FastAPI 官方安全教程说明：可以基于 OAuth2 Password + JWT + 安全哈希做一套真正可用的登录系统。

from datetime import datetime, timedelta
from jose import jwt
from passlib.context import CryptContext

from app.core.config import SECRET_KEY, ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str):
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str):
    return pwd_context.verify(plain_password, hashed_password)

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

十、写依赖 deps

FastAPI 的依赖系统是核心设计之一，官方明确说它很强大，而且非常适合复用数据库连接、认证、权限等逻辑。yield 依赖也正是官方推荐的数据库连接写法。

1）`app/deps/auth.py`

from jose import JWTError, jwt
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

from app.core.config import SECRET_KEY, ALGORITHM
from app.db.session import SessionLocal
from app.models.user import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=401,
        detail="无效的身份凭证"
    )

    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = db.query(User).filter(User.username == username).first()
    if user is None:
        raise credentials_exception
    return user

为什么 `tokenUrl="/auth/login"`？

因为我们后面会把登录接口放进 auth 路由模块里。
FastAPI 安全工具会把这个路径集成到 OpenAPI 文档中，这也是官方教程反复强调的用法。

十一、写业务层 services

不是所有项目都必须有 services/，但企业项目里很常见。

因为路由文件最好不要写一大坨业务细节。

`app/services/document_service.py`

import os
from app.core.config import UPLOAD_DIR

def save_upload_file(filename: str, content: bytes) -> str:
    os.makedirs(UPLOAD_DIR, exist_ok=True)
    save_path = os.path.join(UPLOAD_DIR, filename)

    with open(save_path, "wb") as f:
        f.write(content)

    return save_path

def process_document(doc_id: int):
    print(f"后台正在处理文档：{doc_id}")

小白理解

这个文件就是把“文件保存”和“后台处理”这些业务动作抽出来。
以后如果你要接 PDF 解析、分块、向量化，就继续往这里加。

十二、写路由 routers

FastAPI 官方的 Bigger Applications 核心就是：
使用 APIRouter 把不同模块的接口拆开，然后再在主应用里 include_router()。

1）`app/api/routers/user.py`

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session

from app.deps.auth import get_db, get_current_user
from app.models.user import User
from app.schemas.user import UserPublic

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/", response_model=list[UserPublic])
def list_users(db: Session = Depends(get_db)):
    return db.query(User).all()

@router.get("/me", response_model=UserPublic)
def read_me(current_user: User = Depends(get_current_user)):
    return current_user

@router.get("/{user_id}", response_model=UserPublic)
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user

2）`app/api/routers/auth.py`

from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from app.deps.auth import get_db
from app.models.user import User
from app.schemas.user import UserCreate, UserPublic
from app.schemas.auth import Token
from app.core.security import hash_password, verify_password, create_access_token

router = APIRouter(prefix="/auth", tags=["auth"])

@router.post("/register", response_model=UserPublic)
def register(user: UserCreate, db: Session = Depends(get_db)):
    existing_user = db.query(User).filter(User.username == user.username).first()
    if existing_user:
        raise HTTPException(status_code=400, detail="用户名已存在")

    db_user = User(
        username=user.username,
        password=hash_password(user.password),
        age=user.age
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@router.post("/login", response_model=Token)
def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db)
):
    user = db.query(User).filter(User.username == form_data.username).first()

    if not user or not verify_password(form_data.password, user.password):
        raise HTTPException(status_code=401, detail="用户名或密码错误")

    access_token = create_access_token(data={"sub": user.username})
    return {
        "access_token": access_token,
        "token_type": "bearer"
    }

3）`app/api/routers/document.py`

from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, BackgroundTasks
from sqlalchemy.orm import Session

from app.deps.auth import get_db, get_current_user
from app.models.user import User
from app.models.document import Document
from app.schemas.document import DocumentPublic
from app.services.document_service import save_upload_file, process_document

router = APIRouter(prefix="/documents", tags=["documents"])

@router.post("/upload", response_model=DocumentPublic)
async def upload_document(
    background_tasks: BackgroundTasks,
    file: UploadFile = File(...),
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    if not file.filename.endswith(".pdf"):
        raise HTTPException(status_code=400, detail="只允许上传 PDF 文件")

    content = await file.read()
    save_path = save_upload_file(file.filename, content)

    db_doc = Document(
        filename=file.filename,
        filepath=save_path,
        owner_id=current_user.id,
        status="processing"
    )
    db.add(db_doc)
    db.commit()
    db.refresh(db_doc)

    background_tasks.add_task(process_document, db_doc.id)

    return db_doc

@router.get("/", response_model=list[DocumentPublic])
def list_documents(
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    docs = db.query(Document).filter(Document.owner_id == current_user.id).all()
    return docs

@router.get("/{doc_id}", response_model=DocumentPublic)
def get_document(
    doc_id: int,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    doc = db.query(Document).filter(
        Document.id == doc_id,
        Document.owner_id == current_user.id
    ).first()

    if not doc:
        raise HTTPException(status_code=404, detail="文档不存在")

    return doc

@router.delete("/{doc_id}")
def delete_document(
    doc_id: int,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    doc = db.query(Document).filter(
        Document.id == doc_id,
        Document.owner_id == current_user.id
    ).first()

    if not doc:
        raise HTTPException(status_code=404, detail="文档不存在")

    db.delete(doc)
    db.commit()

    return {"message": "删除成功"}

十三、写主入口 main.py

main.py 的任务应该尽量简单：

创建应用
注册中间件
注册路由
初始化数据库表

这也是多文件项目最核心的思路。

`app/main.py`

import time
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware

from app.db.base import Base
from app.db.session import engine
from app.api.routers import user, auth, document

Base.metadata.create_all(bind=engine)

app = FastAPI(title="Document Center API - Structured")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.perf_counter()
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    print(f"{request.method} {request.url.path} - {process_time:.4f}s")
    response.headers["X-Process-Time"] = str(process_time)
    response.headers["X-Project-Name"] = "document-center"
    return response

@app.get("/")
async def root():
    return {"message": "欢迎来到结构化版 Document Center API"}

@app.get("/ping")
async def ping():
    return {"status": "ok"}

app.include_router(auth.router)
app.include_router(user.router)
app.include_router(document.router)

十四、补一个 `init.py` 说明

为了让 Python 更稳定地识别包，建议在这些目录下都加 __init__.py：

app/__init__.py
app/core/__init__.py
app/db/__init__.py
app/models/__init__.py
app/schemas/__init__.py
app/deps/__init__.py
app/services/__init__.py
app/api/__init__.py
app/api/routers/__init__.py

其中 app/api/routers/__init__.py 可以写：

from . import user, auth, document

十五、启动方式

因为现在入口文件变成了 app/main.py，所以启动命令要改成：

uvicorn app.main:app --reload

然后打开：

http://127.0.0.1:8000/docs

十六、现在你该怎么测试

顺序和上一版类似，只是路径更规范了。

1）注册

POST /auth/register

请求体：

{
  "username": "alice",
  "password": "123456",
  "age": 20
}

2）登录

POST /auth/login

表单：

username: alice
password: 123456

3）授权

点 /docs 右上角 Authorize

输入 token。

4）获取当前用户

GET /users/me

5）上传 PDF

POST /documents/upload

6）查文档列表

GET /documents/

十七、你要理解这套结构背后的“企业思维”

这里最重要的不是把文件拆开，而是学会“按职责拆”。

1）为什么 `schemas` 和 `models` 分开

因为它们不是一回事。

`models`

是数据库表长什么样。

`schemas`

是接口输入输出长什么样。

比如数据库里有密码字段，但接口返回给前端时不应该带密码。
所以 User 和 UserPublic 必须分开。

2）为什么 `router` 和 `service` 分开

`router`

只负责：

接参数
调业务
返回结果

`service`

只负责：

真正执行业务逻辑

例如上传文件：

router 负责接收上传请求
service 负责把文件写入磁盘、后续处理

这样以后逻辑变复杂时，不会把接口文件写成一锅粥。

3）为什么 `deps` 单独放出来

FastAPI 依赖系统就是为了复用公共逻辑。官方明确把数据库连接、认证、权限这些场景作为依赖系统的典型用途。

比如：

get_db()：所有接口都能拿数据库连接
get_current_user()：所有需要登录的接口都能直接复用

这就是企业项目里最常见的写法之一。

4）为什么 `core` 单独放安全和配置

因为这些东西是“全项目共享能力”。

比如：

SECRET_KEY
JWT 算法
密码哈希
token 生成

这些不属于某个具体业务模块，所以不应该塞进 user.py 或 document.py。

十八、从“小白能跑”升级到“更像公司项目”

你现在这版已经比单文件强很多了。
接下来还可以继续升级。

升级 1：把配置改成 `.env`

官方高级文档建议用 Settings 处理环境变量和 .env 文件，并可配合缓存避免每次请求重复读取。

你后面可以改成：

开发环境一个配置
生产环境一个配置
本地不把密钥写死在代码里

升级 2：把 SQLite 换 PostgreSQL

学习期 SQLite 很合适。
真实项目更常见的是 PostgreSQL / MySQL。

升级 3：把同步数据库改成异步

你现在先跑通最重要。
后面再升级 async SQLAlchemy / asyncpg。

升级 4：给 `services/` 增加文档解析逻辑

比如：

PDF 提取文本
文本切片
向量化
建索引
RAG 检索

这样你的 document_service.py 就会慢慢变成真正的知识库处理模块。

升级 5：补测试

FastAPI 官方测试章节说明，可以很方便地基于 pytest/httpx 测接口。这个是你后面进入更规范开发时再补。

十九、这一版和上一版的区别

单文件版适合

第一次学习
快速理解 FastAPI 基本结构
先把功能跑通

项目拆分版适合

开始做真实项目
功能越来越多
后续会持续维护
希望代码更像公司写法

二十、你现在可以这样练

最适合你的练法不是“重新看一遍”，而是直接重构：

第一步

先把上一版单文件项目复制一份。

第二步

新建 app/ 目录，按我上面结构开始拆。

第三步

先拆数据库、schema、security。

第四步

再拆 auth、user、document 三个 router。

第五步

最后把 main.py 清空到只剩：

app 创建
middleware
include_router

你做完这一步，项目结构感就会明显提升。

二十一、你已经进入什么阶段了

如果你能把单文件版成功拆成这一版，你就已经不只是“会抄 FastAPI demo”了。

你已经开始具备这些能力：

理解后端项目结构
区分数据库模型和接口模型
理解依赖注入的实际价值
理解路由层和业务层的分工
能搭出一个可扩展的 AI 后端基础骨架

这已经非常接近你后面做：

RAG 知识库后端
智能体后端
文档管理系统后端
登录鉴权后台系统

的基础框架了

AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

基于SpringBoot+Vue的在线装修管理系统管理系统设计与实现【Java+MySQL+MyBatis完整源码】

AtomGit开源社区

SpringBoot+Vue 银行客户管理系统平台完整项目源码+SQL脚本+接口文档【Java Web毕设】

AtomGit开源社区

【毕业设计】SpringBoot+Vue+MySQL 银行客户管理系统平台源码+数据库+论文+部署文档

AtomGit开源社区

所有评论(0)

查看更多评论

zou54188666

@zou54188666

已为社区贡献4条内容

《FastAPI 小白 10 天实战教程（二）：项目拆分版，按企业项目结构重构》

zou54188666

一、这一版你会学到什么

二、先理解：为什么要拆分

三、最终目录结构

四、重构前先装依赖

五、先创建目录和文件

六、先写数据库层

1）app/db/session.py

2）app/db/base.py

七、写数据库模型 models

1）app/models/user.py

2）app/models/document.py

八、写接口数据模型 schemas

1）app/schemas/user.py

2）app/schemas/auth.py

3）app/schemas/document.py

九、写核心配置 core

1）app/core/config.py

2）app/core/security.py

十、写依赖 deps

1）app/deps/auth.py

为什么 tokenUrl="/auth/login"？

十一、写业务层 services

app/services/document_service.py

小白理解

十二、写路由 routers

1）app/api/routers/user.py

2）app/api/routers/auth.py

3）app/api/routers/document.py

十三、写主入口 main.py

app/main.py

十四、补一个 __init__.py 说明

十五、启动方式

十六、现在你该怎么测试

1）注册

2）登录

3）授权

4）获取当前用户

5）上传 PDF

6）查文档列表

十七、你要理解这套结构背后的“企业思维”

1）为什么 schemas 和 models 分开

models

schemas

2）为什么 router 和 service 分开

router

service

3）为什么 deps 单独放出来

4）为什么 core 单独放安全和配置

十八、从“小白能跑”升级到“更像公司项目”

升级 1：把配置改成 .env

升级 2：把 SQLite 换 PostgreSQL

升级 3：把同步数据库改成异步

升级 4：给 services/ 增加文档解析逻辑

升级 5：补测试

十九、这一版和上一版的区别

单文件版适合

项目拆分版适合

二十、你现在可以这样练

第一步

第二步

第三步

第四步

第五步

二十一、你已经进入什么阶段了

所有评论(0)

zou54188666

1）`app/db/session.py`

2）`app/db/base.py`

1）`app/models/user.py`

2）`app/models/document.py`

1）`app/schemas/user.py`

2）`app/schemas/auth.py`

3）`app/schemas/document.py`

1）`app/core/config.py`

2）`app/core/security.py`

1）`app/deps/auth.py`

为什么 `tokenUrl="/auth/login"`？

`app/services/document_service.py`

1）`app/api/routers/user.py`

2）`app/api/routers/auth.py`

3）`app/api/routers/document.py`

`app/main.py`

十四、补一个 `init.py` 说明

1）为什么 `schemas` 和 `models` 分开

`models`

`schemas`

2）为什么 `router` 和 `service` 分开

`router`

`service`

3）为什么 `deps` 单独放出来

4）为什么 `core` 单独放安全和配置

升级 1：把配置改成 `.env`

升级 4：给 `services/` 增加文档解析逻辑