一、FastAPI到底是什么？

二、FastAPI为什么这么快？

三、环境搭建：5分钟跑起来

	# 使用pyenv管理Python版本（推荐）



	brew install pyenv



	pyenv install 3.11.5



	pyenv global 3.11.5



	 



	# 或直接使用系统Python



	python3 --version

	# 创建项目目录



	mkdir fastapi-demo



	cd fastapi-demo



	 



	# 创建虚拟环境（推荐）



	python3 -m venv venv



	source venv/bin/activate  # Windows: venv\Scripts\activate



	 



	# 安装FastAPI和Uvicorn



	pip install fastapi uvicorn[standard]

	from fastapi import FastAPI



	 



	# 创建应用实例



	app = FastAPI(title="我的第一个FastAPI应用", version="1.0.0")



	 



	# 定义路由



	@app.get("/")



	async def root():



	 return {"message": "Hello World"}



	 



	@app.get("/hello/{name}")



	async def say_hello(name: str):



	 return {"message": f"Hello, {name}!"}

	 uvicorn main:app --reload --host 0.0.0.0 --port 8000

• main:app —— main.py文件中的app实例
• --reload —— 开发模式下自动重启（生产环境不要用）
• --host —— 监听地址
• --port —— 端口号

• API服务：http://localhost:8000
• Swagger文档：http://localhost:8000/docs
• ReDoc文档：http://localhost:8000/redoc

四、核心概念实战

	from fastapi import FastAPI

app = FastAPI()

# 路径参数：从URL路径中提取
@app.get("/users/{user_id}")
async def get_user(user_id: int):  # 自动类型转换 + 校验
    return {"user_id": user_id, "name": f"User_{user_id}"}

# 查询参数：从URL问号后面提取
@app.get("/items")
async def list_items(
    skip: int = 0,      # 默认值
    limit: int = 10,    # 默认值
    category: str | None = None  # 可选参数
):
    return {"skip": skip, "limit": limit, "category": category} 

• 路径参数{user_id}会从URL中提取，user_id: int会自动进行类型转换和校验——传入非数字会返回422错误
• 查询参数从?skip=0&limit=10中提取，有默认值的参数是可选的
• 类型注解让IDE能提供自动补全，也让框架能自动校验

• GET /users/123 → {"user_id": 123, "name": "User_123"}
• GET /items?skip=5&limit=20&category=books

	from fastapi import FastAPI
from pydantic import BaseModel, Field, EmailStr
from typing import Optional
from datetime import datetime

app = FastAPI()

# 定义请求体模型
class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=20, description="用户名")
    email: EmailStr = Field(..., description="邮箱地址")
    password: str = Field(..., min_length=8, description="密码")
    age: Optional[int] = Field(None, ge=0, le=150, description="年龄")
    tags: list[str] = []

# 定义响应体模型
class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    age: Optional[int]
    created_at: datetime

@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate):
    # 业务逻辑：创建用户
    return UserResponse(
        id=1,
        username=user.username,
        email=user.email,
        age=user.age,
        created_at=datetime.now()
    ) 

• UserCreate定义了请求体的结构，Field提供了额外的校验规则（最小长度、最大长度、取值范围等）
• EmailStr会自动校验邮箱格式
• response_model=UserResponse指定了响应的数据结构，框架会自动过滤掉不在模型中的字段
• 如果请求缺少必填字段或字段类型不对，FastAPI会自动返回422错误并说明原因

	from fastapi import FastAPI, Depends, Header, HTTPException

app = FastAPI()

# 定义一个依赖：验证Token
async def verify_token(authorization: str = Header(...)):
    """从请求头中提取并验证Token"""
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="无效的认证格式")
    token = authorization.replace("Bearer ", "")
    if token != "valid-token":
        raise HTTPException(status_code=401, detail="无效的Token")
    return {"user_id": 1, "username": "admin"}

# 使用依赖
@app.get("/protected")
async def protected_route(user: dict = Depends(verify_token)):
    return {"message": f"欢迎, {user['username']}!", "user": user} 

• verify_token是一个依赖函数，从请求头中提取Authorization并验证
• Depends(verify_token)将依赖注入到路由函数中
• 如果验证失败，自动返回401错误
• 验证通过后，返回的用户信息会作为user参数传入路由函数

	import asyncio
from fastapi import FastAPI

app = FastAPI()

# 同步函数（适用于CPU密集型操作）
@app.get("/sync")
def sync_endpoint():
    # 同步操作，会阻塞线程
    return {"result": "done"}

# 异步函数（适用于I/O密集型操作）
@app.get("/async")
async def async_endpoint():
    # 模拟I/O操作：数据库查询、外部API调用等
    await asyncio.sleep(1)
    # 在等待期间，事件循环可以处理其他请求
    return {"result": "done after 1 second"}

# 混合使用：在异步函数中调用同步代码
@app.get("/mixed")
async def mixed_endpoint():
    # 使用 run_in_executor 将同步代码放到线程池执行
    result = await asyncio.to_thread(sync_heavy_work)
    return {"result": result}

def sync_heavy_work():
    # CPU密集型操作
    return sum(range(1000000)) 

• 异步函数用async def定义，在等待I/O时释放线程资源
• await asyncio.sleep(1)模拟I/O等待，期间事件循环可以处理其他请求
• 对于CPU密集型操作，使用asyncio.to_thread放到线程池执行，避免阻塞事件循环

五、数据库集成

	from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import declarative_base, Mapped, mapped_column
from sqlalchemy import select

# 数据库配置
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(unique=True)
    email: Mapped[str]

# 依赖：获取数据库会话
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

app = FastAPI()

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    users = result.scalars().all()
    return [{"id": u.id, "username": u.username, "email": u.email} for u in users] 

• create_async_engine创建异步数据库引擎
• AsyncSessionLocal是异步会话工厂
• get_db是一个依赖，每个请求创建一个数据库会话，请求结束后自动关闭
• 所有数据库操作都是异步的，不会阻塞事件循环

六、统一响应与异常处理

	from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Generic, TypeVar, Optional

T = TypeVar("T")

class ApiResponse(BaseModel, Generic[T]):
    """统一API响应格式"""
    code: int = 200
    msg: str = "success"
    data: Optional[T] = None

app = FastAPI()

@app.get("/users/{user_id}", response_model=ApiResponse)
async def get_user(user_id: int):
    if user_id <= 0:
        return ApiResponse(code=400, msg="用户ID必须大于0")
    # 模拟查询
    user = {"id": user_id, "name": f"User_{user_id}"}
    return ApiResponse(data=user) 

七、中间件

	from fastapi import FastAPI, Request
import time

app = FastAPI()

# 日志中间件：记录每个请求的耗时
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start_time = time.time()
    
    # 记录请求信息
    print(f"收到请求: {request.method} {request.url.path}")
    
    # 继续处理请求
    response = await call_next(request)
    
    # 记录响应信息
    process_time = time.time() - start_time
    print(f"请求完成: {process_time:.4f}秒")
    
    # 在响应头中添加处理时间
    response.headers["X-Process-Time"] = str(process_time)
    return response 

八、自动文档：写代码=写文档

• Swagger UI（/docs）：交互式文档，可以在线调试API
• ReDoc（/redoc）：更美观的静态文档

	from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(
    title="电商API",
    description="这是一个电商平台的API文档",
    version="1.0.0",
    contact={"name": "技术团队", "email": "dev@example.com"}
)

class Product(BaseModel):
    name: str
    price: float
    stock: int

@app.post("/products", 
          summary="创建商品",
          description="创建一个新的商品，需要提供名称、价格和库存",
          response_description="创建成功的商品信息")
async def create_product(product: Product):
    """创建商品接口"""
    return {"id": 1, **product.model_dump()} 

九、优缺点

十、一张图看懂适用场景

十一、和Spring Boot的真实对比

• FastAPI：响应时间50分位45ms，吞吐量2400次/秒，内存占用180MB
• Spring Boot：响应时间50分位80ms，吞吐量1800次/秒

十二、写在最后

• 需要快速部署一个AI模型推理服务
• 需要搭建一个高性能的数据API
• 需要快速验证一个产品原型
• 团队中已经有Python能力

• FastAPI：github.com/tiangolo/fa… （80K+ Star）
• 官方文档：fastapi.tiangolo.com