FastAPI+MySQL+RAG 实战（五）：从零实现政策文档上传入库完整业务

前言

在前面的章节中，我们已经完整搭建了整套RAG系统的基础运行环境：完成FastAPI项目骨架搭建、全局配置文件封装、DeepSeek大模型远程调用、MySQL数据库连接与会话管理。目前项目已经具备了接口服务能力、模型调用能力、数据持久化能力，整套技术底座完全就绪。

从本章开始，我们正式进入业务功能开发阶段。RAG知识库系统的核心数据源是各类政策文档、技术文档、业务文件，所有的智能问答、文档检索、语义对话功能，全部依赖原始文档数据。因此，文档上传、本地存储、数据库元数据入库是整个RAG项目的第一个核心业务基石。

本章将从零实现一套完整的文档上传业务流程，不依赖任何第三方封装工具，纯原生FastAPI+SQLAlchemy实现，包含配置定义、数据库表设计、数据模型封装、文件校验、本地持久化、数据库入库、接口开发、全流程测试等完整功能。本章仅完成原始文档存储与元数据管理，文档解析、父子分块、向量化、向量库存储等高阶能力，将在后续章节迭代实现。

一、本章核心学习目标

通过本章的开发实践，后端项目将正式具备完整的文档管理基础能力，所有落地目标如下：

本章开发完成后，项目将彻底告别纯基础架构阶段，拥有第一个可落地、可测试、可迭代的真实业务功能，为后续RAG分块、向量检索、智能问答功能提供数据支撑。

二、整体业务流程梳理

在编码开发前，我们先梳理最小可行的文档上传业务闭环，清晰的流程可以规避后续逻辑漏洞，保证代码层级清晰、职责单一。

完整业务链路如下：

前端选择本地文档文件 → 发起POST文件上传请求 → 后端接收文件参数与自定义表单参数 → 后端校验文件后缀、文件大小合法性 → 自动创建本地上传目录 → 数据库预写入文档元数据、获取文档唯一UUID → 基于UUID生成唯一文件名、写入本地磁盘 → 更新数据库文件存储路径 → 提交数据库事务 → 封装标准化响应数据返回前端 → 支持列表接口查询所有已上传文档。

需要重点说明的是：本章仅做原始文件存储+元数据入库，不做文本解析、分块、向量化处理。所有文档默认状态为就绪状态，后续章节会迭代文档解析、异常状态、索引版本管理等能力。

三、扩展全局上传配置

项目所有固定规则、环境变量、全局参数均统一存放在config.py 配置类中，方便统一维护、环境切换、后续修改。本次我们新增文件上传相关全局配置。

3.1 配置类代码改造

修改 backend/app/config.py，在原有配置基础上新增上传目录、最大上传大小配置，基于Path对象实现路径管理，适配Windows、Linux跨平台。

核心代码如下：

from pathlib import Path
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
app_name: str = "政务政策 RAG 系统"
debug: bool = True
api_prefix: str = "/api"
database_url: str = (
"mysql+pymysql://root:password@127.0.0.1:3306/gov_rag?charset=utf8mb4"
)

# 文档上传全局配置
upload_path: Path = Path("data/uploads")
max_upload_mb: int = 20

class Config:
env_file = ".env"

def get_settings() -> Settings:
return Settings()

3.2 环境变量配置

在项目根目录的 .env 文件中添加对应配置，支持环境变量覆盖默认参数，适配开发、测试、生产多环境部署：

UPLOAD_PATH=data/uploads MAX_UPLOAD_MB=20

3.3 配置字段详解

• upload_path：文档本地存储根目录，采用相对路径，项目启动后所有上传文件统一存放在 backend/data/uploads 目录，使用Path对象可以完美适配跨平台路径拼接，自动处理正反斜杠问题。

• max_upload_mb：单文件最大上传大小限制，当前配置20MB，有效避免超大文件占用服务器资源、阻塞接口请求。

3.4 配置有效性测试

进入backend目录，执行测试命令校验配置是否生效：

.\\.venv\\Scripts\\python.exe -c "from app.config import get_settings; s=get_settings(); print(s.upload_path); print(s.max_upload_mb)"

正常输出结果：

data\\uploads
20

手动创建上传目录，避免首次上传因目录不存在报错：

New-Item -ItemType Directory -Force data\\uploads

校验目录是否创建成功：

Test-Path data\\uploads

返回True即代表目录就绪。

四、设计数据库文档模型

数据库模型是业务数据的核心载体，我们基于SQLAlchemy 2.0最新ORM语法，设计政策文档专属数据表，同时封装通用时间戳混入类、状态枚举类，实现代码复用与数据规范化。

4.1 完整模型代码

编写 backend/app/models.py，包含枚举类、通用混入类、业务数据表三层结构：

import enum
import uuid
from datetime import date, datetime

from sqlalchemy import Date, DateTime, Enum, Integer, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base

# 生成唯一UUID主键
def new_id() -> str:
return str(uuid.uuid4())

# 文档状态枚举：字符串枚举，直接与字符串等值匹配
class DocumentStatus(str, enum.Enum):
pending = "pending"
ready = "ready"
failed = "failed"

# 通用时间戳混入类：所有业务表通用创建、更新时间
class TimestampMixin:
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
)

# 政策文档业务表
class PolicyDocument(TimestampMixin, Base):
__tablename__ = "policy_documents"

id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_id)
filename: Mapped[str] = mapped_column(String(255))
stored_path: Mapped[str] = mapped_column(String(500))
content_type: Mapped[str] = mapped_column(String(100))
topic: Mapped[str | None] = mapped_column(String(100), index=True)
effective_date: Mapped[date | None] = mapped_column(Date, index=True)
status: Mapped[DocumentStatus] = mapped_column(
Enum(DocumentStatus), default=DocumentStatus.pending, index=True
)
error_message: Mapped[str | None] = mapped_column(Text)
chunk_count: Mapped[int] = mapped_column(Integer, default=0)
index_version: Mapped[int] = mapped_column(Integer, default=1)

4.2 核心技术点详解

4.2.1 字符串枚举优势

普通枚举继承 enum.Enum 时，枚举对象无法直接和字符串匹配，必须通过 .value 取值。而本次使用 class DocumentStatus(str, enum.Enum) 多继承写法，是官方推荐的字符串枚举方案。

优势：枚举对象可以直接与字符串判等，无需额外取值，适配Pydantic序列化、接口参数校验、数据库存储，代码更简洁。

4.2.2 混入类复用机制

TimestampMixin 是通用数据表模板，封装了带时区的创建时间、更新时间，server_default=func.now() 实现数据库层面自动填充创建时间，onupdate=func.now() 实现数据更新时自动刷新时间戳，所有业务表均可直接继承，避免重复编码。

4.2.3 数据表字段业务设计

• id：UUID唯一主键，替代自增ID，适配分布式部署、避免ID泄露、保证文档全局唯一；

• filename：用户上传的原始文件名，用于前端展示；

• stored_path：服务器本地真实存储路径，后端内部使用，不对外暴露；

• content_type：文件MIME类型，标记文件格式；

• topic：政策文档主题，建立索引，支持后续分类筛选；

• effective_date：政策生效日期，支持业务时间筛选，建立索引优化查询；

• status：文档处理状态，默认待处理，本章上传成功直接置为就绪；

• error_message：文档解析、向量化失败时存储错误日志；

• chunk_count：后续分块完成后记录分块数量，初始为0；

• index_version：索引版本号，用于向量库增量更新、版本回滚。

4.3 修复模型加载问题

SQLAlchemy建表的核心原理：只有模型类被Python解释器加载，才会注册到元数据中。修改 database.py 初始化函数，主动导入models模块，保证建表时识别所有模型：

def init_database() -> None:
from app import models # noqa
Base.metadata.create_all(bind=engine)

4.4 执行建表操作

执行数据库初始化脚本，自动创建业务表：

.\\.venv\\Scripts\\python.exe scripts\\init_db.py

执行校验脚本，确认数据表创建成功：

.\\.venv\\Scripts\\python.exe scripts\\check_database.py

输出包含 policy_documents 即代表建表成功。

五、封装Pydantic响应Schema

在前后端分离项目中，绝对不能直接将数据库ORM对象返回给前端，会存在字段泄露、数据格式不规范、多余字段暴露等问题。因此我们通过Pydantic Schema封装标准化响应结构，实现数据库模型与接口响应解耦。

5.1 Schema完整代码

编写 backend/app/schemas.py：

from datetime import date, datetime
from pydantic import BaseModel, ConfigDict

# 基础API模型：禁止接收多余未知字段
class APIModel(BaseModel):
model_config = ConfigDict(extra="forbid")

# ORM转换模型：支持ORM对象自动序列化
class ORMModel(APIModel):
model_config = ConfigDict(from_attributes=True, extra="forbid")

# 文档接口响应模型
class DocumentOut(ORMModel):
id: str
filename: str
content_type: str
topic: str | None
effective_date: date | None
status: str
error_message: str | None
chunk_count: int
index_version: int
created_at: datetime

5.2 核心配置解析

• extra="forbid"：严格校验请求/响应字段，禁止传入未定义字段，规避参数污染、非法参数注入问题；

• from_attributes=True：Pydantic V2核心特性，支持直接将SQLAlchemy ORM数据库对象转换为响应模型，无需手动字典映射，极大简化代码；

• 隐藏敏感字段：响应模型不包含 stored_path 服务器存储路径，保护后端文件地址安全。

六、实现文档上传核心服务

业务逻辑统一分层存放在service层，保证接口层只做参数接收、异常捕获，所有核心校验、文件处理、数据库操作全部下沉到服务层，符合分层架构、单一职责设计思想。

6.1 服务层完整代码

编写 backend/app/services/document_service.py：

from datetime import date
from pathlib import Path

from fastapi import UploadFile
from sqlalchemy.orm import Session

from app.config import get_settings
from app.models import DocumentStatus, PolicyDocument

# 允许上传的文档后缀白名单
ALLOWED_EXTENSIONS = {".pdf", ".docx", ".txt"}

async def save_upload(
session: Session,
upload: UploadFile,
topic: str | None = None,
effective_date: date | None = None,
) -> PolicyDocument:
settings = get_settings()
# 1. 校验文件后缀
suffix = Path(upload.filename or "").suffix.lower()
if suffix not in ALLOWED_EXTENSIONS:
raise ValueError("仅支持 PDF、DOCX、TXT 文件")

# 2. 校验文件大小
content = await upload.read()
if len(content) > settings.max_upload_mb * 1024 * 1024:
raise ValueError(f"文件不能超过 {settings.max_upload_mb}MB")

# 3. 自动创建上传目录
settings.upload_path.mkdir(parents=True, exist_ok=True)

# 4. 初始化数据库记录，预写入数据
document = PolicyDocument(
filename=upload.filename or f"policy{suffix}",
stored_path="",
content_type=upload.content_type or "application/octet-stream",
topic=topic,
effective_date=effective_date,
status=DocumentStatus.ready,
)
session.add(document)
session.flush()

# 5. 基于UUID生成唯一文件名，避免重名覆盖
stored_path = settings.upload_path / f"{document.id}{suffix}"
stored_path.write_bytes(content)

# 6. 更新文件真实存储路径，提交事务
document.stored_path = str(stored_path)
session.commit()
session.refresh(document)
return document

6.2 核心业务逻辑解析

七、开发文档业务接口

接口层负责接收前端请求、参数解析、异常捕获、调用服务、返回标准化数据，我们实现文档上传接口和文档列表查询接口两个核心接口。

7.1 接口代码实现

编写 backend/app/api/documents.py：

from datetime import date
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.database import get_session
from app.models import PolicyDocument
from app.schemas import DocumentOut
from app.services.document_service import save_upload

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

# 文档上传接口
@router.post("", response_model=DocumentOut)
async def upload_document(
file: UploadFile = File(…),
topic: str | None = Form(default=None),
effective_date: date | None = Form(default=None),
session: Session = Depends(get_session),
):
try:
return await save_upload(session, file, topic, effective_date)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc

# 文档列表查询接口
@router.get("", response_model=list[DocumentOut])
async def list_documents(session: Session = Depends(get_session)):
return list(
session.scalars(
select(PolicyDocument).order_by(PolicyDocument.created_at.desc())
).all()
)

7.2 接口设计要点

• 文件上传请求必须使用multipart/form-data 格式，因此参数使用 File、Form 接收；

• 统一捕获服务层自定义异常，转换为400标准请求异常，前端可统一处理错误提示；

• 列表接口按创建时间倒序排列，最新上传文档置顶展示；

• 通过 scalars 直接查询ORM对象，配合Pydantic自动序列化，代码极简高效。

7.3 注册全局路由

修改 backend/app/factory.py，注册文档路由，拼接全局API前缀：

from fastapi import FastAPI
from app.api import documents
from app.config import get_settings

settings = get_settings()

def create_app() -> FastAPI:
app = FastAPI(
title=settings.app_name,
description="FastAPI + LangChain 政策知识库问答系统",
version="1.0.0",
)
# 注册文档接口路由
app.include_router(documents.router, prefix=settings.api_prefix)

@app.get("/")
async def root():
return {
"name": settings.app_name,
"docs": "/docs",
"message": "服务已启动",
}
return app

最终接口完整路径：

• 文档上传：POST /api/documents

• 文档列表：GET /api/documents

八、编写全流程校验脚本

为了避免手动排查BUG，我们编写专属校验脚本，一次性校验模型注册、Schema转换、数据库数据、接口路由四大核心能力，极大提升调试效率。

编写 backend/scripts/check_documents.py：

import sys
from datetime import datetime
from pathlib import Path
from sqlalchemy.sql.expression import select

BACKEND_ROOT = Path(__file__).resolve().parents[1]
if str(BACKEND_ROOT) not in sys.path:
sys.path.insert(0, str(BACKEND_ROOT))

from app.database import Base, SessionLocal
from app.models import PolicyDocument, DocumentStatus
from app.schemas import DocumentOut

# 校验ORM模型是否成功注册
def check_metadata() -> None:
from app import models
print("ORM 注册表：")
print(Base.metadata.tables.keys())

# 校验数据库文档数据
def check_database_rows() -> None:
with SessionLocal() as session:
rows = list(session.scalars(select(PolicyDocument)).all())
print("文档记录：")
print(
[
(row.filename, row.topic, row.status.value, row.chunk_count)
for row in rows
]
)

# 校验ORM转Schema是否正常
def check_schema() -> None:
document = PolicyDocument(
id="demo-document-id",
filename="a.txt",
stored_path="data/uploads/a.txt",
content_type="text/plain",
topic="人才政策",
effective_date=None,
status=DocumentStatus.ready,
chunk_count=0,
index_version=1,
created_at=datetime.now(),
)
output = DocumentOut.model_validate(document)
print("Schema 转换结果：")
print(output)

# 校验API路由注册
def check_routes() -> None:
import main
routes = [route.path for route in main.app.routes if route.path.startswith("/api")]
print("API 路由：")
print(routes)

if __name__ == '__main__':
check_schema()
check_metadata()
check_database_rows()
check_routes()

执行脚本，无报错、路由包含 /api/documents、Schema正常转换即代表所有底层逻辑就绪。

九、全流程功能测试

9.1 准备测试文件

创建测试政策文档，用于上传测试：

Set-Content -Path sample_data\\policy.txt -Encoding UTF8 -Value "青年人才租房补贴申请条件包括年龄、就业、社保和租房状态等要求。"

9.2 启动后端服务

.\\.venv\\Scripts\\python.exe -m uvicorn main:app –reload

访问接口文档地址：http://127.0.0.1:8000/docs，可可视化调试接口。

9.3 测试文档上传

curl.exe -X POST "http://127.0.0.1:8000/api/documents" `
-F "file=@sample_data\\policy.txt" `
-F "topic=人才政策"

返回完整标准化文档数据，状态为ready、chunk_count为0即为正常。

9.4 测试文档列表查询

curl.exe "http://127.0.0.1:8000/api/documents"

成功返回所有已上传文档数组。

9.5 校验本地文件与数据库

• 查看 data/uploads 目录，存在UUID命名的文档文件；

• 执行校验脚本，数据库可查询到完整文档记录。

十、本章总结

本章是RAG智能问答系统从架构到业务的第一个里程碑，我们彻底完成了文档管理的基础底座开发，实现了配置管理、数据表设计、文件校验、本地存储、数据入库、接口开发、全流程测试的完整闭环。

本章核心亮点：