大家好，我是阿浪。干了几年前端，又写一段时间Java，最近几年彻底入了Python的坑。今天不聊虚的，就说说我为什么在众多Python框架里，最终把FastAPI当作主力工具，以及我亲手写的一个完整项目怎么一步步搭起来的。一、我的“转行”心路
从刚毕业做前端，Vue、React都折腾过，那时候觉得后端好神秘。后来自学接触Java，Spring Boot一套下来，确实能扛住大流量，但每次写个简单接口都要建一堆包、配置一堆注解，心里总犯嘀咕：我就想从数据库查十条数据返回给前端，至于这么隆重吗？
最近这段时间因为公司要接一个AI模型，Java调用Python模型各种别扭，干脆让阿浪牵头用Python写个独立服务。我先试了Flask，轻量是真轻量，但一上生产就发现参数校验得自己手写、文档得用Postman维护、异步支持不够丝滑。直到遇见FastAPI，我才发现：原来写后端也可以像写前端那样“爽”。
二、FastAPI到底好在哪？我说三点
很多文章会列一堆基准测试数据，什么每秒几万请求，我不太想复读。从我的实际体感出发，就三个字：
1. 快 —— 不止是运行快，更是开发快
它基于Starlette（异步Web框架）和Pydantic（数据校验），底层是asyncio，所以并发能力天生强。但对我这种“半路出家”的Python选手来说，更吸引我的是写代码不用纠结：
路径参数、查询参数、请求体，全用Python类型提示搞定
写一个def，加个类型，自动就有校验、自动就有Swagger文档
修改代码后，热重载立即生效，不用像Spring Boot那样重启半天
2. 稳 —— 类型安全帮我拦住了好多低级错误
以前写Flask时，经常遇到前端传了个字符串"3"，我当成数字拿去计算，结果报错。FastAPI配合Pydantic，在请求进入函数之前就把类型转换和校验做完了，如果传错了，直接返回400，并带清楚的原因。这让我这种粗心的人省了不少心。
3. 全 —— 生产级功能几乎开箱即用
依赖注入（Depends）让数据库连接、用户鉴权等逻辑复用起来非常优雅
后台任务（BackgroundTasks）处理发邮件、写日志等非阻塞操作
WebSocket、文件上传、CORS配置、中间件……大部分你在Spring Boot里见过的功能，它都有，而且代码量少一半
三、实战：从0搭建一个“用户管理系统”
光说不练假把式。下面我把我最近写的一个小项目拿出来，完整代码都附上。这个项目是一个简单的用户管理API，支持：
创建用户（POST）
查询所有用户（GET）
根据ID查询用户（GET）
更新用户（PUT）
删除用户（DELETE）
数据库用SQLite（方便测试），ORM用SQLAlchemy异步版。
整体流程图（文字版）
客户端请求 ↓FastAPI路由层（验证路径/查询/请求体） ↓依赖注入（获取数据库会话） ↓服务层（业务逻辑，如检查用户是否存在） ↓异步数据库操作（SQLAlchemy） ↓响应序列化（Pydantic模型） ↓返回JSON给客户端
如果要用一句话概括：从请求到响应，全异步，而且每一步都有类型保护。
第一步：环境与依赖
创建虚拟环境，安装依赖：
pip install fastapi uvicorn sqlalchemy aiomysql # 这里我用SQLite，只需sqlalchemy
实际我用的是databases和sqlalchemy，但为了更贴近生产，我用async sqlalchemy。
项目结构：
my_fastapi_project/├── main.py # 应用入口├── database.py # 数据库配置与模型├── schemas.py # Pydantic模型（请求/响应）├── crud.py # 数据库操作函数├── routers.py # 路由（也可以写在main里，但分开放更清晰）└── .env # 环境变量（非必须）
第二步：数据库与模型（database.py）
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSessionfrom sqlalchemy.orm import declarative_base, sessionmakerfrom sqlalchemy import Column, Integer, String, DateTimeimport datetime# 使用SQLite异步驱动，生产可换PostgreSQLDATABASE_URL = "sqlite+aiosqlite:///./test.db"engine = create_async_engine(DATABASE_URL, echo=True)AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)Base = declarative_base()class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) name = Column(String(50), nullable=False, index=True) email = Column(String(100), unique=True, nullable=False) age = Column(Integer, nullable=True) created_at = Column(DateTime, default=datetime.datetime.utcnow)# 异步初始化表async def init_db(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all)
我的想法：这里我特意用了aiosqlite，因为之前用同步SQLite在高并发下会阻塞事件循环。虽然SQLite本身不太适合高并发，但作为本地开发或小项目完全够用。如果上线，把DATABASE_URL换成postgresql+asyncpg://...即可。
第三步：Pydantic模型（schemas.py）
from pydantic import BaseModel, EmailStr, Fieldfrom typing import Optionalfrom datetime import datetime# 创建用户时的请求体class UserCreate(BaseModel): name: str = Field(..., min_length=1, max_length=50, description="姓名") email: EmailStr = Field(..., description="邮箱，自动校验格式") age: Optional[int] = Field(None, ge=0, le=150, description="年龄，0-150")# 更新用户时的请求体（所有字段可选）class UserUpdate(BaseModel): name: Optional[str] = Field(None, min_length=1, max_length=50) email: Optional[EmailStr] = None age: Optional[int] = Field(None, ge=0, le=150)# 返回给客户端的用户信息（去掉敏感字段，也可以加额外字段）class UserOut(BaseModel): id: int name: str email: str age: Optional[int] created_at: datetime class Config: orm_mode = True # 支持ORM对象直接转换
我的想法：EmailStr是Pydantic提供的，能自动校验邮箱格式，再也不用自己写正则了。Field还能加校验条件（如年龄区间），这比Java的@Valid注解要直观得多。而且UserOut里定义orm_mode=True，后续从SQLAlchemy查询出的对象可以直接.dict()转换，省去了手动映射的枯燥工作。第四步：数据库操作层（crud.py）
from sqlalchemy.ext.asyncio import AsyncSessionfrom sqlalchemy import select, update, deletefrom sqlalchemy.exc import IntegrityErrorfrom .models import Userfrom .schemas import UserCreate, UserUpdateasync def get_user(db: AsyncSession, user_id: int): result = await db.execute(select(User).where(User.id == user_id)) return result.scalar_one_or_none()async def get_users(db: AsyncSession, skip: int = 0, limit: int = 100): result = await db.execute(select(User).offset(skip).limit(limit)) return result.scalars().all()async def create_user(db: AsyncSession, user: UserCreate): db_user = User(name=user.name, email=user.email, age=user.age) db.add(db_user) try: await db.commit() await db.refresh(db_user) except IntegrityError: await db.rollback() return None # 邮箱重复等情况 return db_userasync def update_user(db: AsyncSession, user_id: int, user_update: UserUpdate): # 动态构造更新字段 update_data = user_update.dict(exclude_unset=True) # 只传需要更新的字段 if not update_data: return await get_user(db, user_id) # 没有要更新的，直接返回原对象 stmt = update(User).where(User.id == user_id).values(**update_data) await db.execute(stmt) await db.commit() return await get_user(db, user_id)async def delete_user(db: AsyncSession, user_id: int): user = await get_user(db, user_id) if not user: return False await db.delete(user) await db.commit() return True
我的想法：这里有个小细节——update_user里用了exclude_unset=True，这样前端只传name，我就只更新name，不会把email重置为空。这种写法比Spring Boot里用@PatchMapping再判空要简洁得多。第五步：依赖注入与路由（routers.py 或直接写在 main.py）
我习惯把路由单独放一个文件，但为了方便展示，这里直接写在main.py：
from fastapi import FastAPI, Depends, HTTPException, statusfrom sqlalchemy.ext.asyncio import AsyncSessionfrom typing import Listfrom .database import AsyncSessionLocal, init_dbfrom . import crud, schemasapp = FastAPI(title="我的用户管理系统", version="1.0.0")# 依赖：获取数据库会话async def get_db() -> AsyncSession: async with AsyncSessionLocal() as session: yield session# 启动时初始化表@app.on_event("startup")async def startup(): await init_db()# ---------- 路由 ----------@app.post("/users/", response_model=schemas.UserOut, status_code=status.HTTP_201_CREATED)async def create_user(user: schemas.UserCreate, db: AsyncSession = Depends(get_db)): db_user = await crud.create_user(db, user) if not db_user: raise HTTPException(status_code=400, detail="邮箱已被注册") return db_user@app.get("/users/", response_model=List[schemas.UserOut])async def read_users(skip: int = 0, limit: int = 100, db: AsyncSession = Depends(get_db)): users = await crud.get_users(db, skip=skip, limit=limit) return users@app.get("/users/{user_id}", response_model=schemas.UserOut)async def read_user(user_id: int, db: AsyncSession = Depends(get_db)): db_user = await crud.get_user(db, user_id) if not db_user: raise HTTPException(status_code=404, detail="用户不存在") return db_user@app.put("/users/{user_id}", response_model=schemas.UserOut)async def update_user(user_id: int, user_update: schemas.UserUpdate, db: AsyncSession = Depends(get_db)): db_user = await crud.update_user(db, user_id, user_update) if not db_user: raise HTTPException(status_code=404, detail="用户不存在") return db_user@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)async def delete_user(user_id: int, db: AsyncSession = Depends(get_db)): success = await crud.delete_user(db, user_id) if not success: raise HTTPException(status_code=404, detail="用户不存在") return # 204 No Content
我的想法：
Depends(get_db) 是FastAPI依赖注入的灵魂。它会在每个请求进来时自动创建数据库会话，请求结束后自动关闭，不用我们手动try...finally。
response_model 会自动过滤掉模型里没有的字段，比如我返回来User对象包含created_at，但前端不需要，我就在UserOut里不定义它（如果我不想暴露）。实际上我定义了，但你可以灵活控制。
错误处理用HTTPException，配合状态码，和Spring Boot的ResponseStatusException有异曲同工之妙，但更轻量。
第六步：启动服务
在项目根目录执行：
uvicorn main:app --reload --host 0.0.0.0 --port 8000
然后浏览器打开 http://localhost:8000/docs，你会看到自动生成的交互式API文档，可以直接在上面测试所有接口，连Postman都省了。四、完整代码运行演示
我把上面所有代码合并成一个文件（便于复制测试），但实际项目建议分包。合并版main.py如下：
# 合并版（为了演示，把所有内容放一起）from fastapi import FastAPI, Depends, HTTPException, statusfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSessionfrom sqlalchemy.orm import declarative_base, sessionmakerfrom sqlalchemy import Column, Integer, String, DateTimefrom pydantic import BaseModel, EmailStr, Fieldfrom typing import Optional, Listfrom datetime import datetimefrom sqlalchemy import select, update, deletefrom sqlalchemy.exc import IntegrityErrorimport uvicorn# ------- 数据库 -------DATABASE_URL = "sqlite+aiosqlite:///./test.db"engine = create_async_engine(DATABASE_URL, echo=True)AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)Base = declarative_base()class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) name = Column(String(50), nullable=False) email = Column(String(100), unique=True, nullable=False) age = Column(Integer, nullable=True) created_at = Column(DateTime, default=datetime.utcnow)async def init_db(): async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all)# ------- Pydantic -------class UserCreate(BaseModel): name: str = Field(..., min_length=1, max_length=50) email: EmailStr age: Optional[int] = Field(None, ge=0, le=150)class UserUpdate(BaseModel): name: Optional[str] = Field(None, min_length=1, max_length=50) email: Optional[EmailStr] = None age: Optional[int] = Field(None, ge=0, le=150)class UserOut(BaseModel): id: int name: str email: str age: Optional[int] created_at: datetime class Config: orm_mode = True# ------- CRUD -------async def get_user(db: AsyncSession, user_id: int): result = await db.execute(select(User).where(User.id == user_id)) return result.scalar_one_or_none()async def get_users(db: AsyncSession, skip: int = 0, limit: int = 100): result = await db.execute(select(User).offset(skip).limit(limit)) return result.scalars().all()async def create_user(db: AsyncSession, user: UserCreate): db_user = User(name=user.name, email=user.email, age=user.age) db.add(db_user) try: await db.commit() await db.refresh(db_user) except IntegrityError: await db.rollback() return None return db_userasync def update_user(db: AsyncSession, user_id: int, user_update: UserUpdate): update_data = user_update.dict(exclude_unset=True) if not update_data: return await get_user(db, user_id) stmt = update(User).where(User.id == user_id).values(**update_data) await db.execute(stmt) await db.commit() return await get_user(db, user_id)async def delete_user(db: AsyncSession, user_id: int): user = await get_user(db, user_id) if not user: return False await db.delete(user) await db.commit() return True# ------- FastAPI App -------app = FastAPI(title="用户管理API")async def get_db() -> AsyncSession: async with AsyncSessionLocal() as session: yield session@app.on_event("startup")async def startup(): await init_db()@app.post("/users/", response_model=UserOut, status_code=status.HTTP_201_CREATED)async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)): db_user = await crud.create_user(db, user) if not db_user: raise HTTPException(status_code=400, detail="邮箱已被注册") return db_user@app.get("/users/", response_model=List[UserOut])async def read_users(skip: int = 0, limit: int = 100, db: AsyncSession = Depends(get_db)): users = await crud.get_users(db, skip=skip, limit=limit) return users@app.get("/users/{user_id}", response_model=UserOut)async def read_user(user_id: int, db: AsyncSession = Depends(get_db)): db_user = await get_user(db, user_id) if not db_user: raise HTTPException(status_code=404, detail="用户不存在") return db_user@app.put("/users/{user_id}", response_model=UserOut)async def update_user(user_id: int, user_update: UserUpdate, db: AsyncSession = Depends(get_db)): db_user = await update_user(db, user_id, user_update) if not db_user: raise HTTPException(status_code=404, detail="用户不存在") return db_user@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)async def delete_user(user_id: int, db: AsyncSession = Depends(get_db)): success = await delete_user(db, user_id) if not success: raise HTTPException(status_code=404, detail="用户不存在") returnif __name__ == "__main__": uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)
把这个文件保存为main.py，安装好依赖（pip install fastapi uvicorn sqlalchemy aiosqlite pydantic[email]），然后运行python main.py，打开/docs即可测试。五、跟Spring Boot的硬核对比（我的真实感受）
我也不是盲目吹Python。我拿Spring Boot写过支付网关，那套事务管理、AOP、监控确实成熟。但换成FastAPI后，有些维度确实拉开了差距：
| | |
|---|
| 启动速度 | | |
| 代码行数 | 一个简单CRUD得写Controller/Service/Repository/DTO，至少300行 | |
| 文档生成 | 需要集成Swagger或SpringDoc，配置较多 | |
| 参数校验 | | |
| 异步支持 | | async/await |
| 类型安全 | | 运行时Pydantic保证，但借助IDE插件也很强 |
| 生态成熟度 | | |
我个人看法：
六、我在生产环境踩过的坑（以及解决方案）
坑1：异步SQLAlchemy的事务管理
刚开始我没用AsyncSessionLocal的上下文管理器，而是手动begin()，结果有一次并发请求导致会话状态错乱。后来老老实实用async with，每个请求独立会话，完美解决。
坑2：Pydantic模型里用了orm_mode=True，但忘记设置from_attributes=True（新版本）
在Pydantic V2中，orm_mode改成了from_attributes，需要留意。我用的是V2，所以上面代码应该写from_attributes = True，但我为了兼容旧版写了orm_mode，实际使用时根据你的版本调整。
坑3：依赖注入的循环依赖
比如get_db依赖了某个配置类，而配置类又需要数据库读取，这就绕了。我解决的办法是把配置类做成单例，提前加载，不依赖请求周期。
七、我的学习建议（给同样转行的朋友）
别把FastAPI当黑盒：花一下午把官方教程（大概60页）通读一遍，尤其tutorial部分，比看任何博客都管用。
动手改写：把我上面的代码换成PostgreSQL，加上JWT鉴权，再集成Redis缓存，你就基本掌握生产级写法了。
拥抱类型提示：虽然Python是动态语言，但用了类型提示后，IDE的补全和错误检测会让你写Python像写Java一样自信。
善用依赖注入：Depends可以嵌套，比如Depends(get_current_user)依赖Depends(get_db)，这样复用性极高，摆脱了Spring里那套复杂的@Autowired和构造器注入的选择困难。
八、最后说两句
从Java转到Python，我并没有抛弃前者，而是多了一把趁手的工具。FastAPI让我重新找回了写代码的爽快感——不用重启、不用等编译、不用写那些没营养的getter/setter。如果你也厌倦了繁琐的配置，或者你正在搭建一个AI服务、一个数据可视化后端，我真的建议你花一天时间试试FastAPI。
10个月宝宝每天需要喝多少奶粉？
