用 FastAPI 搭一个新闻项目，router / crud / model / schema 应该怎么分层

学完路由、参数、响应、依赖注入、ORM 之后很多人会遇到第二个问题。单个知识点都会了但一到项目里就不知道文件该怎么放。新闻接口写哪里SQL 查询写哪里Pydantic 模型写哪里数据库表结构写哪里统一响应、异常处理、认证工具又写哪里如果所有代码都塞进main.py项目很快就会变成一坨。这篇我们用课程里的 AI 掘金头条项目把 FastAPI 项目分层讲清楚。核心思路很简单main.py只做装配router管 HTTPcrud管数据库操作model管表结构schema管输入输出形状utils放横切工具。本篇准备这一篇不引入新的依赖沿用前面已经准备好的 FastAPI SQLAlchemy 环境pipinstallfastapi uvicorn sqlalchemy aiomysql本篇重点不是多装一个包而是看清楚文件边界入口、路由、数据访问、ORM 模型、请求响应模型分别应该放在哪里。1. 为什么不能把所有代码都写进 main.py刚开始学 FastAPI这样写没问题fromfastapiimportFastAPI appFastAPI()app.get(/news/list)asyncdefget_news_list():return[]但真实项目里一个新闻系统至少有这些功能新闻分类新闻列表新闻详情用户注册用户登录获取用户信息收藏新闻浏览历史Redis 缓存异常处理跨域配置如果全部堆在main.py你会得到一个几千行的入口文件。这类代码最可怕的地方不是长而是边界消失。HTTP 参数、数据库查询、业务规则、响应格式、异常处理全混在一起后面想改一个接口都不知道会影响哪里。所以项目必须分层。2. AI 掘金头条的后端结构课程最终项目的后端结构大概是这样toutiao_backend/ ├── main.py ├── config/ │ ├── db_conf.py │ └── cache_conf.py ├── models/ │ ├── users.py │ ├── news.py │ ├── favorite.py │ └── history.py ├── schemas/ │ ├── base.py │ ├── users.py │ ├── news.py │ ├── favorite.py │ └── history.py ├── crud/ │ ├── news.py │ ├── news_cache.py │ ├── users.py │ ├── favorite.py │ └── history.py ├── routers/ │ ├── news.py │ ├── users.py │ ├── favorite.py │ └── history.py ├── cache/ │ └── news_cache.py └── utils/ ├── auth.py ├── security.py ├── response.py ├── exception.py └── exception_handlers.py看起来文件不少但逻辑很清楚。main.py 应用入口routers 路由层crud 数据访问层models ORM 模型MySQLschemas 请求/响应模型utils 横切工具cache 缓存封装Redis这张图比目录本身更重要。读一个 FastAPI 项目时不要先问有多少文件。先问这些文件各自负责什么。3. main.py只做应用级装配最终项目里的main.py做的事情很少。大概是fromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddlewarefromroutersimportnews,users,favorite,historyfromutils.exception_handlersimportregister_exception_handlers appFastAPI()register_exception_handlers(app)app.add_middleware(CORSMiddleware,allow_origins[http://localhost:5173,http://127.0.0.1:5173,],allow_credentialsTrue,allow_methods[*],allow_headers[*],)app.include_router(news.router)app.include_router(users.router)app.include_router(favorite.router)app.include_router(history.router)你会发现main.py基本不写业务逻辑。它只负责创建 FastAPI 应用注册异常处理器注册 CORS 中间件挂载业务路由这就是一个干净入口文件应该有的样子。入口文件越干净项目越容易维护。4. routers负责 HTTP 接口层routers/news.py这类文件负责定义接口。它关心的是URL 是什么HTTP 方法是什么参数从哪里来需要哪些依赖调用哪个 CRUD 函数返回什么响应简化后的新闻列表接口可能长这样fromfastapiimportAPIRouter,Depends,Queryfromsqlalchemy.ext.asyncioimportAsyncSessionfromconfig.db_confimportget_dbfromcrudimportnews routerAPIRouter(prefix/api/news,tags[新闻])router.get(/list)asyncdefget_news_list(category_id:int|NoneQuery(None,aliascategoryId),page:intQuery(1,ge1),page_size:intQuery(10,ge1,le100,aliaspageSize),db:AsyncSessionDepends(get_db)):returnawaitnews.get_news_list(dbdb,category_idcategory_id,pagepage,page_sizepage_size)这里有个边界很重要。router 可以处理 HTTP 参数但不应该堆复杂 SQL。如果 router 里全是select(...)、join(...)、where(...)说明数据访问逻辑已经漏到 HTTP 层了。5. crud负责数据库操作crud/news.py才是写数据库查询的地方。示例fromsqlalchemyimportselect,funcfromsqlalchemy.ext.asyncioimportAsyncSessionfrommodels.newsimportNewsasyncdefget_news_list(db:AsyncSession,category_id:int|None,page:int,page_size:int):stmtselect(News)ifcategory_id:stmtstmt.where(News.category_idcategory_id)total_resultawaitdb.execute(select(func.count()).select_from(stmt.subquery()))totaltotal_result.scalar()offset(page-1)*page_size resultawaitdb.execute(stmt.offset(offset).limit(page_size))return{list:result.scalars().all(),total:total,hasMore:totalpage*page_size}crud 层关心的是数据库。它不应该关心请求头是什么HTTP 状态码怎么返回前端路由叫什么这能让数据访问逻辑更容易复用。比如同一个get_news_list未来可能被接口调用也可能被后台任务调用。6. models负责数据库表结构models/news.py表达的是表结构。简化版fromsqlalchemyimportString,Text,Integer,DateTime,ForeignKeyfromsqlalchemy.ormimportMapped,mapped_columnclassNews(Base):__tablename__newsid:Mapped[int]mapped_column(Integer,primary_keyTrue)title:Mapped[str]mapped_column(String(255))description:Mapped[str|None]mapped_column(String(500))content:Mapped[str]mapped_column(Text)category_id:Mapped[int]mapped_column(ForeignKey(news_category.id))views:Mapped[int]mapped_column(Integer,default0)model 层不写接口逻辑。它只回答一个问题数据库里这张表长什么样。7. schemas负责请求和响应形状数据库字段不等于 API 字段。这句话非常重要。数据库里可能叫publish_time前端可能希望拿到publishTime。数据库里可能有password接口绝不能返回。这时就需要 schema。fromdatetimeimportdatetimefrompydanticimportBaseModel,ConfigDictclassNewsItemResponse(BaseModel):model_configConfigDict(from_attributesTrue)id:inttitle:strdescription:str|Noneimage:str|Noneauthor:str|Noneviews:intpublish_time:datetimePydantic v2 里ConfigDict(from_attributesTrue)可以让模型从 ORM 对象属性里读取数据。这样你就能把 SQLAlchemy 对象转成响应模型。schema 层的价值是定义 API 契约。只要契约稳定数据库内部怎么调整就不会轻易影响前端。8. utils放横切工具项目里的utils包含文件作用auth.py获取当前用户Token 认证依赖security.py密码哈希和校验response.py统一成功响应exception.py异常处理函数exception_handlers.py注册异常处理器这些逻辑不属于某一个具体业务模块。它们横跨多个模块所以放在utils比较合适。但也要注意utils很容易变成杂物间。判断一个函数该不该进utils可以问一句它是不是多个模块都需要而且不依赖某个具体业务如果只是新闻模块内部用就别放utils放crud/news.py或新闻相关模块里更清楚。9. 一次新闻列表请求的完整链路现在把这些层串起来。用户访问GET /api/news/list?categoryId1page1pageSize10项目内部大概这样跑MySQLmodels/news.pycrud/news.pyrouters/news.pymain.pyVue 前端MySQLmodels/news.pycrud/news.pyrouters/news.pymain.pyVue 前端GET /api/news/list路由匹配解析 Query 参数调用 get_news_list使用 News 模型构造查询执行 SQL返回数据返回列表和 totalJSON 响应这条链路看懂了项目结构就不再是死记硬背。10. CORS 为什么放在 main.py前端 Vite 默认可能跑在http://localhost:5173后端 FastAPI 跑在http://127.0.0.1:8000协议、域名、端口只要有一个不同就是不同源。浏览器会触发 CORS 限制。FastAPI 通过CORSMiddleware解决fromfastapi.middleware.corsimportCORSMiddleware app.add_middleware(CORSMiddleware,allow_origins[http://localhost:5173,http://127.0.0.1:5173,],allow_credentialsTrue,allow_methods[*],allow_headers[*],)本地开发时把 Vite 前端常用的两个地址写进去就够了。正式上线时不要这么放。应该明确指定前端域名例如allow_origins[https://your-frontend-domain.com]这是安全边界不是格式问题。尤其要注意如果allow_credentialsTrue就不要再用allow_origins[*]。浏览器携带 Cookie、认证头这类凭证时跨域规则必须更明确。11. 分层不是为了好看是为了控制变化项目分层的核心不是“看起来专业”。而是控制变化范围。变化应该主要影响哪里URL 改了router数据库查询改了crud表字段改了model 和迁移脚本API 返回字段改了schemaToken 认证规则改了utils/auth响应格式改了utils/response这就是分层的意义。当变化来了你知道去哪改也知道不该碰哪里。12. 小结FastAPI 项目分层可以用这句话记main.py 装配应用 router 接 HTTP crud 查数据库 model 映射表 schema 定契约 utils 放通用工具AI 掘金头条项目从 day03 开始做新闻模块day04 加用户模块day05 加收藏和历史day06 加 Redis 缓存。功能越来越多但只要分层边界清楚项目不会失控。下一篇我们继续看用户模块。注册、登录、Token、密码哈希、统一响应、全局异常处理这些东西加上之后一个 FastAPI 项目才真正开始像一个项目。参考资料FastAPI Bigger ApplicationsFastAPI CORS MiddlewarePydantic v2 Models