FastAPI 入门:Python 现代 Web 框架

在 Python Web 开发的世界里，FastAPI 正如一颗新星冉冉升起。它以惊人的速度、简洁的语法和强大的自动文档功能，迅速赢得了开发者的青睐。本文将带你从零开始学习 FastAPI，掌握这个现代 Web 框架的核心特性和实战技巧。

FastAPI 是一个现代、高性能的 Web 框架，用于使用 Python 3.8+ 构建 API。它基于标准 Python 类型提示，具有以下核心优势：

• 🚀 极速性能 - 基于 Starlette 和 Pydantic，性能媲美 NodeJS 和 Go
• 📝 自动文档 - 自动生成 OpenAPI 和 ReDoc 文档
• ✅ 类型检查 - 基于 Python 类型提示，自动验证数据
• 🎯 智能提示 - IDE 自动补全，开发体验极佳
• 🔒 数据验证 - 自动验证请求数据，返回清晰错误
• 🌐 异步支持 - 原生支持 async/await 异步编程

• Python 3.8 或更高版本
• pip 或 uv 包管理器

1# 使用 pip 安装2pip install fastapi uvicorn[standard]34# 或使用 uv（推荐）5uv add fastapi uvicorn[standard]

创建 main.py：

1from fastapi import FastAPI23app = FastAPI()45@app.get("/")6async def root():7    return {"message": "Hello World"}89@app.get("/items/{item_id}")10async def read_item(item_id: int):11    return {"item_id": item_id}

1uvicorn main:app --reload23# 访问 http://127.0.0.1:8000

FastAPI 自动生成两种交互式文档：

1. Swagger UI
    - http://127.0.0.1:8000/docs
2. ReDoc
    - http://127.0.0.1:8000/redoc

打开浏览器访问，即可看到完整的 API 文档和测试界面！

FastAPI 使用装饰器定义路由：

1from fastapi import FastAPI23app = FastAPI()45@app.get("/users")           # GET 请求6async def get_users():7    return {"method": "GET"}89@app.post("/users")          # POST 请求10async def create_user():11    return {"method": "POST"}1213@app.put("/users/{id}")      # PUT 请求14async def update_user(id: int):15    return {"method": "PUT", "id": id}1617@app.delete("/users/{id}")   # DELETE 请求18async def delete_user(id: int):19    return {"method": "DELETE", "id": id}2021@app.patch("/users/{id}")    # PATCH 请求22async def patch_user(id: int):23    return {"method": "PATCH", "id": id}

1from fastapi import FastAPI23app = FastAPI()45@app.get("/users/{user_id}")6async def get_user(user_id: int):7    """获取用户信息"""8    return {"user_id": user_id}910@app.get("/users/{user_id}/posts/{post_id}")11async def get_user_post(user_id: int, post_id: int):12    """获取用户的文章"""13    return {"user_id": user_id, "post_id": post_id}1415# 路径参数类型转换16@app.get("/files/{file_path:path}")17async def read_file(file_path: str):18    """读取文件路径（包含斜杠）"""19    return {"file_path": file_path}

1from fastapi import FastAPI2from typing import Optional34app = FastAPI()56@app.get("/items")7async def read_items(8    skip: int = 0,           # 默认值9    limit: int = 10,10    search: Optional[str] = None  # 可选参数11):12    """获取物品列表"""13    return {14        "skip": skip,15        "limit": limit,16        "search": search17    }1819# 访问：/items?skip=5&limit=20&search=python

1from fastapi import FastAPI2from pydantic import BaseModel, EmailStr, Field34app = FastAPI()56# 定义数据模型7class UserCreate(BaseModel):8    username: str = Field(..., min_length=3, max_length=50)9    email: EmailStr10    password: str = Field(..., min_length=8)11    full_name: Optional[str] = None12    is_active: bool = True1314class UserResponse(BaseModel):15    id: int16    username: str17    email: EmailStr18    full_name: Optional[str]1920    class Config:21        from_attributes = True2223@app.post("/users", response_model=UserResponse)24async def create_user(user: UserCreate):25    """创建用户"""26    # user 自动验证和解析27    return {28        "id": 1,29        "username": user.username,30        "email": user.email,31        "full_name": user.full_name32    }

1from pydantic import BaseModel, validator, Field23class Product(BaseModel):4    name: str5    price: float = Field(..., gt=0)  # 必须大于 06    quantity: int = Field(default=1, ge=0)  # 必须大于等于 078    @validator(&#x27;name&#x27;)9    def name_must_not_be_empty(cls, v):10        if not v.strip():11            raise ValueError(&#x27;名称不能为空&#x27;)12        return v.strip()1314    @validator(&#x27;price&#x27;)15    def price_must_be_positive(cls, v):16        if v <= 0:17            raise ValueError(&#x27;价格必须大于 0&#x27;)18        return v

1from fastapi import FastAPI, status2from pydantic import BaseModel34app = FastAPI()56class Item(BaseModel):7    name: str8    price: float910# 自定义状态码11@app.post("/items", status_code=status.HTTP_201_CREATED)12async def create_item(item: Item):13    return item1415# 简化写法16@app.post("/products", status_code=201)17async def create_product(product: Item):18    return product1920# 响应模型过滤21@app.post("/users", response_model=UserResponse)22async def register_user(user: UserCreate):23    # 自动过滤掉 password 等字段24    return user_data

1from fastapi import FastAPI, HTTPException, status2from pydantic import BaseModel34app = FastAPI()56fake_db = {}78@app.get("/items/{item_id}")9async def get_item(item_id: int):10    if item_id not in fake_db:11        raise HTTPException(12            status_code=status.HTTP_404_NOT_FOUND,13            detail=f"Item {item_id} not found",14            headers={"X-Error": "Not Found"}15        )16    return fake_db[item_id]1718# 自定义异常处理19from fastapi.exceptions import RequestValidationError20from fastapi.responses import JSONResponse2122@app.exception_handler(RequestValidationError)23async def validation_exception_handler(request, exc):24    return JSONResponse(25        status_code=400,26        content={"error": "验证失败", "details": exc.errors()}27    )

1from fastapi import FastAPI, Depends, HTTPException, status2from typing import Optional34app = FastAPI()56# 简单的依赖7def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):8    return {"q": q, "skip": skip, "limit": limit}910@app.get("/items")11async def read_items(commons: dict = Depends(common_parameters)):12    return commons1314# 数据库会话依赖15async def get_db():16    db = Database()17    try:18        yield db19    finally:20        db.close()2122@app.get("/users")23async def get_users(db: Database = Depends(get_db)):24    return db.query(User).all()2526# 认证依赖27async def get_current_user(token: str = Header(...)):28    user = decode_token(token)29    if not user:30        raise HTTPException(status_code=401, detail="Invalid token")31    return user3233@app.get("/me")34async def read_me(current_user: User = Depends(get_current_user)):35    return current_user

1from fastapi import FastAPI, Request2from fastapi.middleware.cors import CORSMiddleware3import time45app = FastAPI()67# CORS 中间件8app.add_middleware(9    CORSMiddleware,10    allow_origins=["*"],  # 生产环境应该限制具体域名11    allow_credentials=True,12    allow_methods=["*"],13    allow_headers=["*"],14)1516# 自定义中间件17@app.middleware("http")18async def add_process_time_header(request: Request, call_next):19    start_time = time.time()20    response = await call_next(request)21    process_time = time.time() - start_time22    response.headers["X-Process-Time"] = str(process_time)23    return response

1from fastapi import FastAPI, HTTPException, status2from pydantic import BaseModel3from typing import List, Optional4from datetime import datetime56app = FastAPI(title="Todo API")78class TodoBase(BaseModel):9    title: str10    description: Optional[str] = None11    completed: bool = False1213class TodoCreate(TodoBase):14    pass1516class TodoUpdate(BaseModel):17    title: Optional[str] = None18    description: Optional[str] = None19    completed: Optional[bool] = None2021class Todo(TodoBase):22    id: int23    created_at: datetime2425    class Config:26        from_attributes = True2728# 模拟数据库29todos_db = []30next_id = 13132@app.post("/todos", response_model=Todo, status_code=status.HTTP_201_CREATED)33async def create_todo(todo: TodoCreate):34    global next_id35    new_todo = {36        "id": next_id,37        **todo.dict(),38        "created_at": datetime.now()39    }40    todos_db.append(new_todo)41    next_id += 142    return new_todo4344@app.get("/todos", response_model=List[Todo])45async def get_todos(skip: int = 0, limit: int = 10):46    return todos_db[skip:skip+limit]4748@app.get("/todos/{todo_id}", response_model=Todo)49async def get_todo(todo_id: int):50    for todo in todos_db:51        if todo["id"] == todo_id:52            return todo53    raise HTTPException(status_code=404, detail="Todo not found")5455@app.put("/todos/{todo_id}", response_model=Todo)56async def update_todo(todo_id: int, todo_update: TodoUpdate):57    for i, todo in enumerate(todos_db):58        if todo["id"] == todo_id:59            update_data = todo_update.dict(exclude_unset=True)60            todos_db[i].update(update_data)61            return todos_db[i]62    raise HTTPException(status_code=404, detail="Todo not found")6364@app.delete("/todos/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)65async def delete_todo(todo_id: int):66    for i, todo in enumerate(todos_db):67        if todo["id"] == todo_id:68            todos_db.pop(i)69            return70    raise HTTPException(status_code=404, detail="Todo not found")

1from fastapi import FastAPI, Depends, HTTPException, status2from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm3from pydantic import BaseModel, EmailStr4from datetime import datetime, timedelta5from typing import Optional6from passlib.context import CryptContext7import jwt89app = FastAPI()1011# 配置12SECRET_KEY = "your-secret-key"13ALGORITHM = "HS256"14ACCESS_TOKEN_EXPIRE_MINUTES = 301516# 密码加密17pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")18oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")1920class UserCreate(BaseModel):21    username: str22    email: EmailStr23    password: str2425class Token(BaseModel):26    access_token: str27    token_type: str2829class User(BaseModel):30    username: str31    email: EmailStr32    disabled: Optional[bool] = None3334# 模拟数据库35users_db = {}3637def verify_password(plain_password, hashed_password):38    return pwd_context.verify(plain_password, hashed_password)3940def get_password_hash(password):41    return pwd_context.hash(password)4243def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):44    to_encode = data.copy()45    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))46    to_encode.update({"exp": expire})47    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)4849async def get_current_user(token: str = Depends(oauth2_scheme)):50    credentials_exception = HTTPException(51        status_code=status.HTTP_401_UNAUTHORIZED,52        detail="Could not validate credentials",53        headers={"WWW-Authenticate": "Bearer"},54    )55    try:56        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])57        username: str = payload.get("sub")58        if username is None:59            raise credentials_exception60    except:61        raise credentials_exception62    user = users_db.get(username)63    if user is None:64        raise credentials_exception65    return user6667@app.post("/register", response_model=User)68async def register(user: UserCreate):69    if user.username in users_db:70        raise HTTPException(status_code=400, detail="Username already registered")7172    hashed_password = get_password_hash(user.password)73    user_data = {74        "username": user.username,75        "email": user.email,76        "hashed_password": hashed_password,77        "disabled": False78    }79    users_db[user.username] = user_data80    return {"username": user.username, "email": user.email}8182@app.post("/token", response_model=Token)83async def login(form_data: OAuth2PasswordRequestForm = Depends()):84    user = users_db.get(form_data.username)85    if not user or not verify_password(form_data.password, user["hashed_password"]):86        raise HTTPException(87            status_code=status.HTTP_401_UNAUTHORIZED,88            detail="Incorrect username or password",89            headers={"WWW-Authenticate": "Bearer"},90        )9192    access_token = create_access_token(93        data={"sub": user["username"]},94        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)95    )96    return {"access_token": access_token, "token_type": "bearer"}9798@app.get("/users/me", response_model=User)99async def read_users_me(current_user: User = Depends(get_current_user)):100    return current_user

1from fastapi import FastAPI, File, UploadFile, HTTPException2from fastapi.responses import FileResponse3from typing import List4import shutil5from pathlib import Path67app = FastAPI()89UPLOAD_DIR = Path("uploads")10UPLOAD_DIR.mkdir(exist_ok=True)1112@app.post("/upload")13async def upload_file(file: UploadFile = File(...)):14    """上传单个文件"""15    if not file.filename:16        raise HTTPException(status_code=400, detail="No file provided")1718    file_path = UPLOAD_DIR / file.filename1920    with open(file_path, "wb") as buffer:21        shutil.copyfileobj(file.file, buffer)2223    return {"filename": file.filename, "size": file_path.stat().st_size}2425@app.post("/upload/multiple")26async def upload_multiple_files(files: List[UploadFile] = File(...)):27    """上传多个文件"""28    uploaded = []29    for file in files:30        file_path = UPLOAD_DIR / file.filename31        with open(file_path, "wb") as buffer:32            shutil.copyfileobj(file.file, buffer)33        uploaded.append({"filename": file.filename})34    return uploaded3536@app.get("/download/{filename}")37async def download_file(filename: str):38    """下载文件"""39    file_path = UPLOAD_DIR / filename40    if not file_path.exists():41        raise HTTPException(status_code=404, detail="File not found")42    return FileResponse(file_path)

1from fastapi import FastAPI2import asyncio3import httpx45app = FastAPI()67# 同步端点（阻塞）8@app.get("/sync")9def sync_endpoint():10    time.sleep(2)  # 阻塞 2 秒11    return {"type": "sync"}1213# 异步端点（非阻塞）14@app.get("/async")15async def async_endpoint():16    await asyncio.sleep(2)  # 非阻塞 2 秒17    return {"type": "async"}1819# 异步 HTTP 请求20@app.get("/fetch-data")21async def fetch_data():22    async with httpx.AsyncClient() as client:23        responses = await asyncio.gather(24            client.get("https://api.example.com/data1"),25            client.get("https://api.example.com/data2"),26            client.get("https://api.example.com/data3")27        )28    return [r.json() for r in responses]

1from fastapi import FastAPI, BackgroundTasks2import time34app = FastAPI()56def send_email(email: str, message: str):7    """模拟发送邮件"""8    time.sleep(2)9    print(f"发送邮件到 {email}: {message}")1011@app.post("/send-email")12async def send_email_endpoint(13    email: str,14    background_tasks: BackgroundTasks15):16    """发送邮件（后台任务）"""17    background_tasks.add_task(send_email, email, "欢迎注册！")18    return {"message": "邮件将在后台发送"}

1from fastapi import FastAPI, Depends, HTTPException2from sqlalchemy import create_engine, Column, Integer, String3from sqlalchemy.ext.declarative import declarative_base4from sqlalchemy.orm import sessionmaker, Session5from pydantic import BaseModel6from typing import List, Optional78DATABASE_URL = "sqlite:///./test.db"910engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})11SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)12Base = declarative_base()1314# 数据库模型15class User(Base):16    __tablename__ = "users"1718    id = Column(Integer, primary_key=True, index=True)19    username = Column(String, unique=True, index=True)20    email = Column(String, unique=True, index=True)2122# 创建表23Base.metadata.create_all(bind=engine)2425# Pydantic 模型26class UserBase(BaseModel):27    username: str28    email: str2930class UserCreate(UserBase):31    password: str3233class UserResponse(UserBase):34    id: int3536    class Config:37        from_attributes = True3839# 依赖40def get_db():41    db = SessionLocal()42    try:43        yield db44    finally:45        db.close()4647app = FastAPI()4849@app.post("/users", response_model=UserResponse)50def create_user(user: UserCreate, db: Session = Depends(get_db)):51    db_user = User(username=user.username, email=user.email)52    db.add(db_user)53    db.commit()54    db.refresh(db_user)55    return db_user5657@app.get("/users", response_model=List[UserResponse])58def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):59    users = db.query(User).offset(skip).limit(limit).all()60    return users

1from fastapi.testclient import TestClient2from main import app34client = TestClient(app)56def test_read_main():7    response = client.get("/")8    assert response.status_code == 2009    assert response.json() == {"message": "Hello World"}1011def test_create_item():12    response = client.post(13        "/items",14        json={"name": "Test Item", "price": 10.0}15    )16    assert response.status_code == 20117    assert response.json()["name"] == "Test Item"1819def test_get_user_not_found():20    response = client.get("/users/999")21    assert response.status_code == 404

1# 安装测试依赖2pip install pytest httpx34# 运行测试5pytest test_main.py -v67# 查看覆盖率8pytest --cov=main test_main.py

1FROM python:3.11-slim23WORKDIR /app45# 安装依赖6COPY requirements.txt .7RUN pip install --no-cache-dir -r requirements.txt89# 复制代码10COPY . .1112# 运行应用13CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

1version: '3.8'23services:4  api:5    build: .6    ports:7      - "8000:8000"8    environment:9      - DATABASE_URL=postgresql://user:pass@db:5432/mydb10    depends_on:11      - db1213  db:14    image: postgres:1515    environment:16      - POSTGRES_USER=user17      - POSTGRES_PASSWORD=pass18      - POSTGRES_DB=mydb19    volumes:20      - postgres_data:/var/lib/postgresql/data2122volumes:23  postgres_data:

1# config.py2from pydantic_settings import BaseSettings34class Settings(BaseSettings):5    app_name: str = "My API"6    debug: bool = False7    database_url: str8    secret_key: str910    class Config:11        env_file = ".env"1213settings = Settings()

1# main.py2from fastapi import FastAPI3from config import settings45app = FastAPI(6    title=settings.app_name,7    debug=settings.debug8)

1my_project/2├── app/3│   ├── __init__.py4│   ├── main.py5│   ├── config.py6│   ├── dependencies.py7│   ├── models/8│   │   ├── __init__.py9│   │   └── user.py10│   ├── schemas/11│   │   ├── __init__.py12│   │   └── user.py13│   ├── routers/14│   │   ├── __init__.py15│   │   ├── users.py16│   │   └── items.py17│   └── utils/18│       ├── __init__.py19│       └── security.py20├── tests/21│   └── test_api.py22├── requirements.txt23└── README.md

1# app/routers/users.py2from fastapi import APIRouter34router = APIRouter(prefix="/users", tags=["users"])56@router.get("/")7async def get_users():8    return []910@router.post("/")11async def create_user():12    return {}1314# app/main.py15from fastapi import FastAPI16from app.routers import users1718app = FastAPI()19app.include_router(users.router)

1# .env2DATABASE_URL=postgresql://user:pass@localhost:5432/mydb3SECRET_KEY=your-secret-key4DEBUG=false

1# config.py2from pydantic_settings import BaseSettings34class Settings(BaseSettings):5    database_url: str6    secret_key: str7    debug: bool = False89    class Config:10        env_file = ".env"1112settings = Settings()

1import logging23logging.basicConfig(4    level=logging.INFO,5    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'6)78logger = logging.getLogger(__name__)910@app.get("/items")11async def get_items():12    logger.info("获取物品列表")13    return []

FastAPI 适合：

• 构建 API 服务
• 需要高性能
• 需要自动文档
• 需要类型安全

Flask 适合：

• 简单 Web 应用
• 需要完整模板引擎
• 项目已有 Flask 生态

1from fastapi.middleware.cors import CORSMiddleware23app.add_middleware(4    CORSMiddleware,5    allow_origins=["http://localhost:3000"],6    allow_credentials=True,7    allow_methods=["*"],8    allow_headers=["*"],9)

1@app.get("/items")2async def get_items(page: int = 1, page_size: int = 10):3    skip = (page - 1) * page_size4    items = db.query(Item).offset(skip).limit(page_size).all()5    return {"items": items, "total": total_count}

• 📚 FastAPI 官方文档
• 🎓 FastAPI 中文教程
• 💻 FastAPI GitHub
• 🎥 FastAPI 视频教程
• 💬 FastAPI Discord

通过本文的学习，你应该掌握了：✅ FastAPI 的核心特性和优势  ✅ 路径参数、查询参数和请求体  ✅ Pydantic 数据验证  ✅ 依赖注入系统  ✅ 异步编程和后台任务  ✅ 数据库集成  ✅ 测试和部署  ✅ 最佳实践  FastAPI 是一个强大且易用的现代 Web 框架。现在就开始构建你的第一个 FastAPI 应用吧！🚀