FastAPI 实战指南

一、FastAPI简介

1. 什么是 FastAPI？

参考官网原文:
FastAPI is a modern, fast (high-performance), web framework for building APIs with Python based on standard Python type hints.
FastAPI 是一个现代、快速（高性能）的 Web 框架，用于基于标准 Python 类型提示构建 API。

2. FastAPI 与其他框架对比

特性	FastAPI	Flask	Django
性能	高	中等	中等
异步支持	原生支持	需要扩展	需要扩展
类型提示	必需	可选	可选
自动文档	内置	需要扩展	需要扩展
学习曲线	中等	低	高
适合场景	REST API	小型应用	大型全栈应用

3. FastAPI 的核心特性

FastAPI 基于以下核心特性和标准构建：

OpenAPI：FastAPI 使用 OpenAPI 标准为你的所有 API 创建定义，包括定义路径、参数、请求体、安全等
JSON Schema：由于 OpenAPI 基于并使用 JSON Schema 进行模型定义，FastAPI 使用 Pydantic 的模型定义提供 JSON Schema
自动文档：交互式 API 文档（Swagger UI）和备用文档（ReDoc）会根据定义的 OpenAPI 自动生成
现代 Python：完全基于 Python 3.6+ 的类型提示开发，充分利用现代 Python 特性
依赖注入：FastAPI 拥有强大且易用的依赖注入系统，使代码更简洁、更易于测试和复用
安全性与认证：内置支持 OAuth2 和 JWT 认证，以及 API Key 等标准安全机制
异步支持：原生支持 async/await，可以编写异步代码，提高并发性能
数据验证：使用 Pydantic 进行数据验证，自动将请求数据转换为 Python 类型
自动类型转换：自动将请求中的字符串转换为 Python 类型，减少手动转换代码

参考：FastAPI 官方文档 - Project Documentation

二、快速入门

1. 环境搭建

1. 安装 uv

如果尚未安装 uv，可以使用以下命令安装：

# macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

# 或使用 pip
pip install uv

2. 创建项目目录

创建一个新的项目目录并进入：

mkdir my-fastapi-app
cd my-fastapi-app

3. 初始化项目

在项目目录中运行：

# 初始化一个 Python 项目
uv init

# 这会创建一个虚拟环境和 pyproject.toml 文件

4. 安装依赖

# 安装 FastAPI 和 ASGI 服务器 Uvicorn
uv add fastapi uvicorn

提示：安装完成后，依赖会被记录在 pyproject.toml 和 uv.lock 文件中。

2. 创建第一个 API 应用

创建文件 main.py：

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

3. 运行和测试 API

使用 Uvicorn 启动服务器：

uvicorn main:app --reload

参数说明：

main:app：main.py 文件中的 app 对象
--reload：代码修改后自动重启服务器

启动后访问：

API：http://127.0.0.1:8000/
交互式文档：http://127.0.0.1:8000/docs

4. 基础路由

路径参数

路径参数是指在 URL 路径中定义的参数，例如 /items/42 中的 42。

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int):
    return {"item_id": item_id}

查询参数

查询参数是指在 URL 中以键值对的形式传递的参数，例如 /items/?skip=0&limit=10。

from fastapi import FastAPI
from typing import Optional

app = FastAPI()

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

@app.get("/items/")
def read_item(skip: int = 0, limit: int = 10):
    return fake_items_db[skip : skip + limit]

三、请求体与数据验证

1. Pydantic 模型

Pydantic 是一个使用 Python 类型提示进行数据验证和设置管理的库。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.post("/items/")
def create_item(item: Item):
    return item

2. 请求体参数

JSON 参数

JSON 参数对应请求头 Content-Type: application/json。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

class User(BaseModel):
    username: str
    full_name: str | None = None

@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item, user: User):
    results = {"item_id": item_id, "item": item, "user": user}
    return results

Form 表单

Form 表单参数对应请求头 Content-Type: application/x-www-form-urlencoded。

from fastapi import FastAPI
from fastapi import Form

app = FastAPI()

@app.post("/login/")
def login(username: str = Form(...), password: str = Form(...)):
    return {"username": username}

文件上传

文件上传参数对应请求头 Content-Type: multipart/form-data。

from fastapi import FastAPI, UploadFile, File

app = FastAPI()

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

四、响应处理

1. 响应模型

定义响应体的结构。
默认情况下，FastAPI 会自动将响应体转换为 JSON 格式。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float = 10.5
    tags: list[str] = []

items = {
    "foo": {"name": "Foo", "price": 50.2},
    "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
    "baz": {
        "name": "Baz",
        "description": None,
        "price": 50.2,
        "tax": 10.5,
        "tags": [],
    },
}

# 测试: item_id 只能是 foo、bar 或 baz
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: str):
    return items[item_id]

2. 响应状态码

定义响应状态码。
默认情况下，FastAPI 会根据函数的返回值自动设置状态码。

from fastapi import FastAPI
from fastapi import status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
def create_item(name: str):
    return {"name": name}

3. 自定义响应头

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/headers/")
def get_headers():
    content = {"message": "Hello World"}
    headers = {"X-Cat-Dog": "alone in the world", "Content-Language": "en-US"}
    return JSONResponse(content=content, headers=headers)

五、依赖注入系统

依赖注入系统允许你在路由处理函数中使用依赖项，例如数据库会话、用户认证等。
类似 Spring Boot 中的依赖注入。

1. 基本使用

from fastapi import FastAPI, Depends

app = FastAPI()

def common_parameters(q: str | None = 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

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

2. 类作为依赖

from fastapi import FastAPI, Depends
from typing import Optional

app = FastAPI()

class CommonParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]

@app.get("/items/")
async def read_items(commons: CommonParams = Depends(CommonParams)):
    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

3. 子依赖

from fastapi import FastAPI, Depends

app = FastAPI()

def query_extractor(q: str | None = None):
    return q

def query_or_cookie_extractor(
    q: str = Depends(query_extractor), last_query: str | None = None
):
    if not q:
        return last_query
    return q

@app.get("/items/")
async def read_query(
    query_ref: str = Depends(query_or_cookie_extractor)
):
    return {"query_ref": query_ref}

六、SQLAlchemy 集成

SQLAlchemy 异步模式与 FastAPI 集成完整示例

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.future import select
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase
from pydantic import BaseModel, Field

# ============ 数据库配置 ============

# 使用 SQLite 异步驱动
DATABASE_URL = "sqlite+aiosqlite:///./test_async.db"

# 创建异步引擎
engine = create_async_engine(DATABASE_URL, echo=True)

# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(
    engine, 
    class_=AsyncSession, 
    expire_on_commit=False
)

# 创建基类
class Base(DeclarativeBase):
    pass

# ============ 数据库模型 ============

class User(Base):
    """用户模型（简化版）"""
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String)

# ============ Pydantic 模型（请求/响应） ============

class UserCreate(BaseModel):
    """创建用户模型"""
    name: str = Field(..., min_length=1, max_length=100, description="用户名称")

class UserResponse(BaseModel):
    """用户响应模型"""
    id: int
    name: str
    
    class Config:
        from_attributes = True  # 支持 ORM 模式

# ============ FastAPI 应用初始化 ============

app = FastAPI(
    title="用户管理 API（简化版）",
    description="使用 SQLAlchemy 异步模式的 FastAPI 简化示例",
    version="1.0.0"
)

# ============ 数据库依赖注入 ============

async def get_session() -> AsyncSession:
    """获取数据库会话"""
    async with AsyncSessionLocal() as session:
        yield session

# ============ CRUD 操作函数 ============

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)
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user

# ============ API 路由 ============

@app.post("/users/", response_model=UserResponse, tags=["用户管理"])
async def create_user_endpoint(user: UserCreate, db: AsyncSession = Depends(get_session)):
    """创建新用户"""
    return await create_user(db, user)

@app.get("/users/", response_model=list[UserResponse], tags=["用户管理"])
async def read_users(skip: int = 0, limit: int = 100, db: AsyncSession = Depends(get_session)):
    """获取用户列表"""
    return await get_users(db, skip=skip, limit=limit)

# ============ 启动事件 ============

@app.on_event("startup")
async def startup_event():
    """应用启动时创建数据库表"""
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    print("✓ 数据库表创建成功")

@app.on_event("shutdown")
async def shutdown_event():
    """应用关闭时关闭数据库连接"""
    await engine.dispose()
    print("✓ 数据库连接已关闭")

重要提示：在运行此示例前，请确保已安装所有必需的依赖：

uv add fastapi uvicorn sqlalchemy aiosqlite

运行:

uv run uvicorn main:app --reload

一、FastAPI简介#

1. 什么是 FastAPI？#

2. FastAPI 与其他框架对比#

3. FastAPI 的核心特性#

二、快速入门#

1. 环境搭建#

1. 安装 uv#

2. 创建项目目录#

3. 初始化项目#

4. 安装依赖#

2. 创建第一个 API 应用#

3. 运行和测试 API#

4. 基础路由#

路径参数#

查询参数#

三、请求体与数据验证#

1. Pydantic 模型#

2. 请求体参数#

JSON 参数#

Form 表单#

文件上传#

四、响应处理#

1. 响应模型#

2. 响应状态码#

3. 自定义响应头#

五、依赖注入系统#

1. 基本使用#

2. 类作为依赖#

3. 子依赖#

六、SQLAlchemy 集成#