FastAPI 入门教程：构建高性能 Python API

FastAPI 是一个现代、快速（高性能）的 Python Web 框架，用于构建 API。它基于标准 Python 类型提示，提供了自动数据验证、序列化和 API 文档生成等强大功能。

为什么选择 FastAPI？

FastAPI 在 2025 年已成为构建 Python API 的首选框架，主要优势包括：

• 高性能：性能媲美 NodeJS 和 Go，基于 Starlette 和 Pydantic
• 快速开发：开发速度提升约 200-300%
• 更少的 Bug：减少约 40% 的人为错误
• 直观易用：优秀的编辑器支持，自动补全
• 自动文档：自动生成交互式 API 文档（Swagger UI 和 ReDoc）
• 类型安全：基于 Python 类型提示，提供编译时类型检查
• 异步支持：原生支持 async/await，处理高并发场景

环境准备

安装 Python

FastAPI 需要 Python 3.7 或更高版本（推荐 Python 3.10+）。

# 检查 Python 版本
python --version

安装 FastAPI

推荐使用完整安装，包含所有可选依赖：

# 完整安装（包含 uvicorn 服务器）
pip install "fastapi[standard]"

# 或者分别安装
pip install fastapi
pip install "uvicorn[standard]"

第一个 FastAPI 应用

创建一个名为 main.py 的文件：

from fastapi import FastAPI

# 创建 FastAPI 应用实例
app = FastAPI()

# 定义根路径的 GET 请求处理器
@app.get("/")
async def root():
    return {"message": "Hello World"}

# 带路径参数的示例
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

运行应用

# 启动开发服务器
uvicorn main:app --reload

# --reload 参数启用热重载，代码修改后自动重启

访问 http://127.0.0.1:8000，你会看到 JSON 响应：

{"message": "Hello World"}

自动生成的 API 文档

FastAPI 自动生成交互式 API 文档：

• Swagger UI：访问 http://127.0.0.1:8000/docs
• ReDoc：访问 http://127.0.0.1:8000/redoc

核心概念

1. 路径操作（Path Operations）

路径操作就是 HTTP 方法与路径的组合：

from fastapi import FastAPI

app = FastAPI()

# GET 请求
@app.get("/users")
async def get_users():
    return {"users": []}

# POST 请求
@app.post("/users")
async def create_user():
    return {"message": "User created"}

# PUT 请求
@app.put("/users/{user_id}")
async def update_user(user_id: int):
    return {"user_id": user_id, "message": "User updated"}

# DELETE 请求
@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    return {"user_id": user_id, "message": "User deleted"}

2. 路径参数（Path Parameters）

路径参数是 URL 路径的一部分：

from fastapi import FastAPI

app = FastAPI()

# 基本路径参数
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# 多个路径参数
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: str):
    return {"user_id": user_id, "item_id": item_id}

# 路径参数类型转换和验证
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

3. 查询参数（Query Parameters）

查询参数是 URL 中 ? 后面的键值对：

from fastapi import FastAPI
from typing import Optional

app = FastAPI()

# 可选查询参数
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

# 可选参数（使用 Optional）
@app.get("/users/")
async def read_users(q: Optional[str] = None):
    if q:
        return {"q": q}
    return {"q": "No query provided"}

# 必需查询参数
@app.get("/search/")
async def search(query: str):
    return {"query": query}

4. 请求体（Request Body）

使用 Pydantic 模型定义请求体：

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

app = FastAPI()

# 定义数据模型
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

# 使用请求体
@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

# 带字段验证的模型
class User(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(..., regex=r"^[\w\.-]+@[\w\.-]+\.\w+$")
    age: int = Field(..., ge=0, le=120)
    is_active: bool = True

@app.post("/users/")
async def create_user(user: User):
    return user

5. 数据验证

FastAPI 使用 Pydantic 进行自动数据验证：

from fastapi import FastAPI, Query, Path, Body
from pydantic import BaseModel, Field, EmailStr
from typing import Optional

app = FastAPI()

# 查询参数验证
@app.get("/items/")
async def read_items(
    q: Optional[str] = Query(None, min_length=3, max_length=50),
    page: int = Query(1, ge=1),
    size: int = Query(10, ge=1, le=100)
):
    return {"q": q, "page": page, "size": size}

# 路径参数验证
@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., title="The ID of the item", ge=1)
):
    return {"item_id": item_id}

# 复杂数据验证
class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    password: str = Field(..., min_length=8)
    age: Optional[int] = Field(None, ge=0, le=120)

@app.post("/register/")
async def register(user: UserCreate):
    return {"username": user.username, "email": user.email}

6. 响应模型

定义 API 响应的数据结构：

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    description: Optional[str] = None

class ItemResponse(BaseModel):
    id: int
    name: str
    price: float

# 指定响应模型
@app.post("/items/", response_model=ItemResponse)
async def create_item(item: Item):
    # 模拟数据库保存，返回带 ID 的对象
    return {"id": 1, "name": item.name, "price": item.price}

# 返回列表
@app.get("/items/", response_model=List[ItemResponse])
async def read_items():
    return [
        {"id": 1, "name": "Item 1", "price": 10.5},
        {"id": 2, "name": "Item 2", "price": 20.0}
    ]

7. 依赖注入

FastAPI 的依赖注入系统非常强大：

from fastapi import FastAPI, Depends, HTTPException
from typing import Optional

app = FastAPI()

# 简单依赖
def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

# 类作为依赖
class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/users/")
async def read_users(commons: CommonQueryParams = Depends()):
    return commons

# 认证依赖示例
def get_current_user(token: str = Depends(lambda: "fake-token")):
    if token != "fake-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"username": "john_doe"}

@app.get("/me")
async def read_current_user(current_user: dict = Depends(get_current_user)):
    return current_user

8. 异步编程

FastAPI 原生支持异步操作：

from fastapi import FastAPI
import asyncio
import httpx

app = FastAPI()

# 异步路径操作
@app.get("/async-items/")
async def read_async_items():
    # 模拟异步数据库查询
    await asyncio.sleep(1)
    return {"items": ["item1", "item2"]}

# 异步 HTTP 请求
@app.get("/external-api/")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

# 同步操作（如果不需要异步）
@app.get("/sync-items/")
def read_sync_items():
    return {"items": ["item1", "item2"]}

实战示例：构建 TODO API

让我们构建一个完整的 TODO 应用 API：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime

app = FastAPI(title="TODO API", version="1.0.0")

# 数据模型
class TodoCreate(BaseModel):
    title: str
    description: Optional[str] = None
    completed: bool = False

class TodoUpdate(BaseModel):
    title: Optional[str] = None
    description: Optional[str] = None
    completed: Optional[bool] = None

class Todo(BaseModel):
    id: int
    title: str
    description: Optional[str] = None
    completed: bool
    created_at: datetime

# 模拟数据库
todos_db = []
todo_id_counter = 1

# API 端点
@app.get("/", tags=["Root"])
async def root():
    return {"message": "Welcome to TODO API"}

@app.get("/todos", response_model=List[Todo], tags=["Todos"])
async def get_todos(skip: int = 0, limit: int = 10):
    """获取所有 TODO 项"""
    return todos_db[skip : skip + limit]

@app.get("/todos/{todo_id}", response_model=Todo, tags=["Todos"])
async def get_todo(todo_id: int):
    """根据 ID 获取单个 TODO 项"""
    for todo in todos_db:
        if todo["id"] == todo_id:
            return todo
    raise HTTPException(status_code=404, detail="Todo not found")

@app.post("/todos", response_model=Todo, status_code=201, tags=["Todos"])
async def create_todo(todo: TodoCreate):
    """创建新的 TODO 项"""
    global todo_id_counter
    new_todo = {
        "id": todo_id_counter,
        "title": todo.title,
        "description": todo.description,
        "completed": todo.completed,
        "created_at": datetime.now()
    }
    todos_db.append(new_todo)
    todo_id_counter += 1
    return new_todo

@app.put("/todos/{todo_id}", response_model=Todo, tags=["Todos"])
async def update_todo(todo_id: int, todo_update: TodoUpdate):
    """更新 TODO 项"""
    for index, todo in enumerate(todos_db):
        if todo["id"] == todo_id:
            updated_todo = todo.copy()
            if todo_update.title is not None:
                updated_todo["title"] = todo_update.title
            if todo_update.description is not None:
                updated_todo["description"] = todo_update.description
            if todo_update.completed is not None:
                updated_todo["completed"] = todo_update.completed
            todos_db[index] = updated_todo
            return updated_todo
    raise HTTPException(status_code=404, detail="Todo not found")

@app.delete("/todos/{todo_id}", status_code=204, tags=["Todos"])
async def delete_todo(todo_id: int):
    """删除 TODO 项"""
    for index, todo in enumerate(todos_db):
        if todo["id"] == todo_id:
            todos_db.pop(index)
            return
    raise HTTPException(status_code=404, detail="Todo not found")

测试 API

启动服务器后，访问 http://127.0.0.1:8000/docs 可以在 Swagger UI 中测试所有端点。

使用 curl 测试：

# 创建 TODO
curl -X POST "http://127.0.0.1:8000/todos" \
  -H "Content-Type: application/json" \
  -d '{"title": "学习 FastAPI", "description": "完成入门教程"}'

# 获取所有 TODO
curl "http://127.0.0.1:8000/todos"

# 获取单个 TODO
curl "http://127.0.0.1:8000/todos/1"

# 更新 TODO
curl -X PUT "http://127.0.0.1:8000/todos/1" \
  -H "Content-Type: application/json" \
  -d '{"completed": true}'

# 删除 TODO
curl -X DELETE "http://127.0.0.1:8000/todos/1"

最佳实践

1. 项目结构

my_fastapi_app/
├── app/
│   ├── __init__.py
│   ├── main.py          # 应用入口
│   ├── models.py        # Pydantic 模型
│   ├── schemas.py       # 数据库模型
│   ├── crud.py          # 数据库操作
│   ├── dependencies.py  # 依赖注入
│   └── routers/         # 路由模块
│       ├── __init__.py
│       ├── users.py
│       └── items.py
├── tests/
├── requirements.txt
└── .env

2. 使用路由器组织代码

# app/routers/users.py
from fastapi import APIRouter

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

@router.get("/")
async def read_users():
    return [{"username": "user1"}]

# app/main.py
from fastapi import FastAPI
from app.routers import users

app = FastAPI()
app.include_router(users.router)

3. 环境变量配置

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "FastAPI App"
    database_url: str
    secret_key: str

    class Config:
        env_file = ".env"

settings = Settings()

下一步学习

• 数据库集成：学习使用 SQLAlchemy 或 Tortoise ORM
• 认证授权：实现 JWT 认证、OAuth2
• 中间件：添加 CORS、日志、性能监控
• 测试：使用 pytest 编写单元测试和集成测试
• 部署：使用 Docker、Kubernetes 部署到生产环境

总结

FastAPI 是一个功能强大、易于学习的现代 Python Web 框架。通过本教程，你已经掌握了：

• ✅ FastAPI 的核心概念和优势
• ✅ 路由、请求参数、请求体的使用
• ✅ 数据验证和响应模型
• ✅ 依赖注入系统
• ✅ 异步编程支持
• ✅ 构建完整的 RESTful API

继续实践和探索，你将能够构建出高性能、可维护的 Python API 应用！

参考资源

• FastAPI 官方文档
• FastAPI Setup Guide 2025
• FastAPI 完整教程
• FastAPI 快速上手指南