Python FastAPI 快速入门与开发实践 – wiki基地

作者： /

Python FastAPI 快速入门与开发实践：构建现代化、高性能的 Web API

随着互联网应用的飞速发展，构建高性能、易于维护的 Web API 成为了现代软件开发的核心任务之一。在众多 Python Web 框架中，FastAPI 凭借其卓越的性能、出色的开发者体验以及自动生成的交互式文档，迅速崛起并赢得了广泛的关注和青睐。

本文将带你深入了解 FastAPI，从零开始快速入门，并通过丰富的示例和实践技巧，帮助你掌握使用 FastAPI 构建现代化 Web API 的核心能力。

1. 为什么选择 FastAPI？

在开始学习之前，我们先来了解一下 FastAPI 的魅力所在，它为何能在众多成熟框架中脱颖而出：

  • 极高的性能 (Performance)： FastAPI 基于 Starlette（用于 Web 部分）和 Pydantic（用于数据部分），是目前 Python 框架中性能最优异的之一。它原生支持异步（async/await），能够充分利用现代 CPU 多核优势，处理高并发请求。根据官方和第三方测试，其性能可媲美 NodeJS 和 Go 语言框架。
  • 出色的开发者体验 (Developer Experience)：
    • 代码提示与自动补全： 得益于 Pydantic 的数据模型和 Python 的类型提示，FastAPI 提供了强大的编辑器支持，无论是参数、请求体还是响应，都能获得精准的代码提示，极大提高了开发效率，减少了错误。
    • 自动数据校验 (Data Validation)： 利用 Pydantic 模型，FastAPI 能自动对请求数据（路径参数、查询参数、请求体）进行强大的校验，并在数据无效时自动返回清晰的错误信息。这使得手动编写校验代码成为历史。
    • 自动生成文档 (Automatic Docs)： FastAPI 基于 OpenAPI (以前称为 Swagger) 标准，自动生成交互式的 API 文档（Swagger UI）和备用文档（ReDoc）。开发者无需额外工作即可获得精确、实时的接口说明，方便前后端协作和接口测试。
    • 依赖注入系统 (Dependency Injection)： FastAPI 内建了强大易用的依赖注入系统，可以方便地管理数据库连接、认证、权限检查等依赖关系，使得代码更加模块化、可测试、可维护。
  • 现代化特性： 原生支持异步编程 (async def)、类型提示、OpenAPI 标准、JSON Schema 等现代 Python 和 Web 开发特性。
  • 强大的生态系统： FastAPI 可以轻松地与其他 Python 库集成，如 ORM (SQLAlchemy, Ormar, Tortoise ORM)、数据库、Celery (任务队列)、测试库 (pytest)、安全库等。

2. 环境准备与安装

开始之前，请确保你已经安装了 Python 3.7 或更高版本。FastAPI 对类型提示和异步语法的依赖要求较高的 Python 版本。

接下来，使用 pip 安装 FastAPI 和一个 ASGI 服务器。常用的 ASGI 服务器是 Uvicorn，它具有高性能。

bash
pip install fastapi "uvicorn[standard]"

这里 uvicorn[standard] 会安装 Uvicorn 的标准版本，包含了一些常用的可选依赖。

3. 你的第一个 FastAPI 应用：Hello, World!

让我们从最简单的应用开始。创建一个名为 main.py 的文件，并写入以下代码：

“`python

main.py

from fastapi import FastAPI

创建一个 FastAPI 应用实例

app = FastAPI()

定义一个根路径(‘/’)的GET请求处理器

@app.get(“/”)
async def read_root():
# 返回一个字典，FastAPI 会自动将其转换为 JSON 响应
return {“Hello”: “World”}

定义另一个路径(‘/items/{item_id}’)的GET请求处理器

{item_id} 是一个路径参数

@app.get(“/items/{item_id}”)
async def read_item(item_id: int, q: str | None = None):
# item_id: int 表示 item_id 是一个整数类型的路径参数
# q: str | None = None 表示 q 是一个可选的字符串类型的查询参数，默认为 None
# FastAPI 会自动将路径参数转换为指定的类型 (这里是 int)，并校验查询参数

# 返回一个包含路径参数和查询参数的字典
return {"item_id": item_id, "q": q}

“`

代码解释：

  1. from fastapi import FastAPI: 导入 FastAPI 类。
  2. app = FastAPI(): 创建一个 FastAPI 应用实例。所有请求处理都将注册到这个实例上。
  3. @app.get("/"): 这是一个装饰器，用于注册一个处理 HTTP GET 请求的 路径操作 (path operation)，该路径是根路径 /。FastAPI 支持所有标准的 HTTP 方法：@app.post(), @app.put(), @app.delete(), @app.options(), @app.head(), @app.patch(), @app.trace()。
  4. async def read_root():: 定义一个异步函数作为路径操作处理器。使用 async def 允许你在函数内部使用 await 来进行异步操作（例如访问数据库或调用外部服务）。即使你的函数当前没有 await 操作，使用 async def 也是推荐的，因为它使得未来添加异步代码更加方便，并且 FastAPI 本身是异步框架。
  5. return {"Hello": "World"}: 函数返回一个 Python 字典。FastAPI 会自动将其序列化为 JSON 格式，并作为 HTTP 响应返回，响应头会自动设置为 Content-Type: application/json。
  6. @app.get("/items/{item_id}"): 注册另一个 GET 请求处理器，路径是 /items/{item_id}。花括号 {} 中的 item_id 表示这是一个 路径参数。
  7. async def read_item(item_id: int, q: str | None = None):: 定义处理函数。
    • item_id: int: 这里的 item_id 参数名必须与路径中的 {item_id} 匹配。类型提示 : int 告诉 FastAPI 期望 item_id 是一个整数。FastAPI 会自动尝试将从 URL 中获取的路径参数字符串转换为整数，如果转换失败，会自动返回一个 422 Unprocessable Entity 错误。
    • q: str | None = None: 定义一个名为 q 的参数。它没有出现在路径中，因此 FastAPI 会将其识别为 查询参数。str | None 表示 q 可以是字符串或 None。= None 表示这个参数是可选的，如果请求中没有提供 q，它的值默认为 None。

4. 运行你的应用

打开终端，进入到 main.py 文件所在的目录，然后运行 Uvicorn 服务器：

bash
uvicorn main:app --reload

命令解释：

  • uvicorn: 启动 Uvicorn 服务器。
  • main: 指定包含 FastAPI 应用的 Python 模块文件名（不带 .py 后缀）。
  • app: 指定 FastAPI 应用实例的变量名（在 main.py 中是 app = FastAPI()）。
  • --reload: 启用自动重载功能。当你的代码发生变化时，Uvicorn 会自动重启服务器，非常方便开发调试。

你应该会看到类似以下的输出：

INFO: Will watch for changes in these directories: ['/path/to/your/project']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [PID]
INFO: Started server process [PID]
INFO: Waiting for application startup.
INFO: Application startup complete.

这表示你的应用已经在 http://127.0.0.1:8000 上运行了。

5. 访问你的 API 和文档

打开浏览器或使用工具（如 curl、Postman）访问以下地址：

  • http://127.0.0.1:8000/: 你会看到 {"Hello": "World"}。
  • http://127.0.0.1:8000/items/5: 你会看到 {"item_id": 5, "q": null}。
  • http://127.0.0.1:8000/items/10?q=somequery: 你会看到 {"item_id": 10, "q": "somequery"}。
  • http://127.0.0.1:8000/items/abc: 这会触发类型校验失败，你会看到一个 422 错误响应，包含详细的错误信息，告诉你 item_id 应该是整数。
  • http://127.0.0.1:8000/docs: 这将打开自动生成的 Swagger UI 交互式文档。你可以在这里查看所有定义的接口、参数、响应模型，并直接在页面上进行接口测试。
  • http://127.0.0.1:8000/redoc: 这将打开自动生成的 ReDoc 文档，提供了另一种风格的 API 文档展示。

FastAPI 自动生成的文档是其最强大的特性之一，它极大地减少了文档编写和维护的工作量，并确保文档始终与代码保持同步。

6. 请求体处理与 Pydantic 模型

对于 POST、PUT、PATCH 等方法，通常需要在请求中发送数据体（Body），通常是 JSON 格式。FastAPI 结合 Pydantic，使得处理请求体变得非常简单高效。

首先，安装 Pydantic（虽然安装 fastapi 时会自动安装，但明确知道它的作用很重要）。

创建一个 Pydantic 模型来定义你的数据结构。修改 main.py：

“`python

main.py

from fastapi import FastAPI
from pydantic import BaseModel # 导入 BaseModel 类

创建一个 FastAPI 应用实例

app = FastAPI()

定义一个 Pydantic 模型来描述请求体的数据结构

class Item(BaseModel):
name: str # 字段 name 必须是字符串类型
price: float # 字段 price 必须是浮点数类型
is_offer: bool | None = None # 字段 is_offer 是可选的布尔类型，默认值为 None

定义一个根路径(‘/’)的GET请求处理器

@app.get(“/”)
async def read_root():
return {“Hello”: “World”}

定义另一个路径(‘/items/{item_id}’)的GET请求处理器

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

定义一个处理 POST 请求的路径操作，用于创建新的 Item

@app.post(“/items/”)
async def create_item(item: Item): # 参数 item 的类型提示是上面定义的 Item 模型
# FastAPI 会自动：
# 1. 读取请求体
# 2. 将其解析为 JSON
# 3. 根据 Item 模型对 JSON 数据进行校验
# 4. 创建一个 Item 类的实例，并将其作为 item 参数传递给函数
# 如果数据不符合模型定义 (缺少字段, 类型错误等)，则返回 422 错误响应

# 你现在可以直接访问 item 实例的属性，这些属性是经过类型转换和校验的
return {"item_name": item.name, "item_price": item.price}

定义一个处理 PUT 请求的路径操作，结合路径参数和请求体

@app.put(“/items/{item_id}”)
async def update_item(item_id: int, item: Item): # 结合使用路径参数和请求体模型
# 可以访问路径参数 item_id 和 请求体模型 item
return {“item_id”: item_id, “item”: item}

“`

代码解释：

  1. from pydantic import BaseModel: 导入 BaseModel，它是所有 Pydantic 模型的基础类。
  2. class Item(BaseModel): ...: 定义一个继承自 BaseModel 的类 Item。类中的属性（如 name, price, is_offer）使用 Python 类型提示来定义数据的期望类型。Pydantic 将根据这些定义自动生成数据模型，并用于数据校验、序列化和文档生成。
  3. @app.post("/items/"): 定义一个 POST 请求处理。
  4. async def create_item(item: Item):: 定义处理函数。注意参数 item: Item。当你在路径操作函数的参数中使用一个 Pydantic 模型作为类型提示时，FastAPI 会知道这个参数应该从请求体中获取数据，并使用 Item 模型进行校验和解析。
  5. 在 /items/ 路径操作中，当你发送一个 JSON 请求体（例如：{"name": "Foo", "price": 10.5}），FastAPI 会自动进行如下处理：
    • 从请求中读取 JSON 数据。
    • 使用 Item 模型验证 JSON 数据。如果 name 不是字符串或 price 不是浮点数，或者 name/price 缺失（因为它们没有默认值），FastAPI 会自动返回一个包含详细错误信息的 422 响应。
    • 如果数据有效，FastAPI 创建一个 Item 类的实例，并将校验和转换后的数据填充到该实例中。
    • 将这个 Item 实例作为 item 参数传递给 create_item 函数。
  6. 在 /items/{item_id} 的 PUT 请求中，我们展示了如何在同一个路径操作函数中同时接收路径参数 (item_id: int) 和请求体 (item: Item)。FastAPI 会根据类型提示区分它们。

重新运行 uvicorn main:app --reload，然后访问 http://127.0.0.1:8000/docs。你会看到新的 POST 和 PUT 接口，并且文档中详细描述了请求体的数据结构和校验规则。你可以在 Swagger UI 中尝试向 /items/ 发送一个 POST 请求，体验自动校验的强大功能。

7. 依赖注入系统 (Dependency Injection)

FastAPI 的依赖注入系统是一个非常强大的特性，它可以让你的代码更易于组织、复用和测试。依赖注入的核心思想是将函数或路径操作所需的资源或功能作为参数传递进去，而不是在函数内部创建或获取。

FastAPI 使用 Depends 函数来实现依赖注入。你可以定义一个函数，它返回一个值或执行一些操作，然后将这个函数作为另一个路径操作函数的参数的默认值，并使用 Depends() 包裹。

“`python

main.py

from fastapi import FastAPI, Depends, HTTPException, status # 导入 Depends, HTTPException, status
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
name: str
price: float
is_offer: bool | None = None

模拟一个用户认证函数

async def get_current_user(token: str | None = None):
# 假设 token 为 “fake-super-secret-token” 时认证成功
if token != “fake-super-secret-token”:
# 如果认证失败，抛出 HTTPException，FastAPI 会自动将其转换为 HTTP 响应
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, # 使用 status 模块获取标准状态码
detail=”Invalid authentication credentials”,
headers={“WWW-Authenticate”: “Bearer”}, # 可以在错误响应中添加额外头信息
)
return {“username”: “假用户”} # 认证成功返回用户信息

模拟一个依赖，用于获取一些公共参数

async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
# 这个函数本身也可以有依赖，比如 Depends(get_current_user)
return {“q”: q, “skip”: skip, “limit”: limit}

@app.get(“/”)
async def read_root():
return {“Hello”: “World”}

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

@app.post(“/items/”)
async def create_item(item: Item):
return {“item_name”: item.name, “item_price”: item.price}

@app.put(“/items/{item_id}”)
async def update_item(item_id: int, item: Item):
return {“item_id”: item_id, “item”: item}

使用依赖注入：在 /users/me 路径操作中需要用户认证

@app.get(“/users/me”)
async def read_users_me(current_user: dict = Depends(get_current_user)):
# current_user 的值将是 get_current_user() 函数的返回值
return current_user

使用依赖注入：在 /items/ 路径操作中引入公共参数依赖

方法一：直接在函数签名中使用 Depends

@app.get(“/items-with-params/”)
async def read_items_with_params(commons: dict = Depends(common_parameters)):
# commons 将是 common_parameters() 函数的返回值
return commons

方法二：在装饰器中使用 dependencies 参数 (这种方式更适用于多个依赖且不直接在函数中使用依赖返回值的情况)

@app.get(“/items-with-params-alt/”, dependencies=[Depends(common_parameters)])
async def read_items_with_params_alt():
# common_parameters 会被执行，但其返回值不会直接作为参数传递给 read_items_with_params_alt
# 这种方式常用于执行前置逻辑，如权限检查等
return {“message”: “This endpoint uses common_parameters dependency”}

“`

代码解释：

  1. Depends(get_current_user): 在 read_users_me 函数的参数 current_user 的默认值位置，我们使用了 Depends(get_current_user)。这告诉 FastAPI 在调用 read_users_me 之前，先执行 get_current_user 函数，并将 get_current_user 的返回值赋给 current_user 参数。
  2. get_current_user 函数本身接收一个 token 参数。FastAPI 会尝试从请求的各种位置（例如 Header、Query 参数等，具体取决于 Depends 更高级的用法，这里为了简化演示，我们假设它是通过某种方式获取的，或者更常见的是结合 OAuth2 等安全模块自动从 Header 获取）解析出 token 的值并传递给 get_current_user。
  3. 如果 get_current_user 抛出了 HTTPException，FastAPI 会捕获它，并立即返回对应的 HTTP 错误响应，而不会继续执行 read_users_me 函数。
  4. Depends(common_parameters) 演示了另一种常见的依赖使用方式，用于提取和处理多个查询参数，使得路径操作函数签名更简洁。
  5. 依赖函数本身也可以有参数，这些参数也会被 FastAPI 解析（例如 common_parameters 中的 q, skip, limit）。依赖函数甚至可以依赖其他依赖！这形成了一个依赖关系链。
  6. 依赖注入的优势：
    • 代码复用： get_current_user 和 common_parameters 等函数可以在多个路径操作中复用。
    • 模块化： 将特定的逻辑（如认证、分页、DB Session）从核心业务逻辑中分离出来。
    • 可测试性： 在单元测试中，你可以轻松地替换真实的依赖实现（例如数据库连接）为 mock 对象，从而独立测试路径操作函数。

8. 项目结构与 APIRouter

随着项目变大，所有路径操作都放在一个 main.py 文件中会变得难以管理。FastAPI 提供了 APIRouter 来组织你的 API。你可以将相关的路径操作放在单独的文件中（例如 routers/items.py, routers/users.py），然后在主应用中通过 include_router 将它们包含进来。

创建一个 routers 目录，并在其中创建 items.py 和 users.py。

routers/items.py:

“`python

routers/items.py

from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel

创建一个 APIRouter 实例

prefix=”/items” 表示这个 router 下的所有路径都会以 “/items” 为前缀

tags=[“items”] 用于在文档中对这些路径进行分组

router = APIRouter(
prefix=”/items”,
tags=[“items”],
# dependencies=[Depends(get_token_header)], # 可以在 router 级别添加依赖，应用于所有包含的路径操作
# responses={404: {“description”: “Not found”}}, # 可以在 router 级别添加响应描述
)

定义 Item 模型 (可以放在单独的 models.py 文件中)

class Item(BaseModel):
id: str | None = None # 添加 id 字段
name: str
price: float
is_offer: bool | None = None

模拟数据存储

fake_items_db = {“foo”: {“name”: “Foo”, “price”: 3.2}, “bar”: {“name”: “Bar”, “price”: 0.5}}

定义在这个 router 下的路径操作

@router.get(“/”)
async def read_items():
return fake_items_db

@router.get(“/{item_id}”)
async def read_item(item_id: str):
if item_id not in fake_items_db:
raise HTTPException(status_code=404, detail=”Item not found”)
return fake_items_db[item_id]

@router.put(“/{item_id}”)
async def update_item(item_id: str, item: Item):
if item_id not in fake_items_db:
raise HTTPException(status_code=404, detail=”Item not found”)
# 更新模拟数据
fake_items_db[item_id] = item.model_dump() # Pydantic 模型转字典
return {“item_id”: item_id, **item.model_dump()} # ** 表示字典解包

“`

routers/users.py:

“`python

routers/users.py

from fastapi import APIRouter, Depends

导入上面 main.py 中定义的依赖函数 (或者将其移动到一个公共的 dependencies.py 文件中)

from ..main import get_current_user # 如果 get_current_user 在 main.py 中，这样导入

更好的做法是将依赖函数放在公共文件，例如 dependencies.py

假设我们有一个 common_deps.py 文件，并在其中定义了 get_current_user

from ..common_deps import get_current_user

为了示例简单，我们这里直接定义一个简单的依赖函数

async def get_fake_user(user_id: int):
return {“user_id”: user_id, “username”: f”user{user_id}”}

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

@router.get(“/”)
async def read_users():
return [{“username”: “Rick”}, {“username”: “Morty”}]

使用 get_fake_user 依赖

@router.get(“/{user_id}”)
async def read_user(user: dict = Depends(get_fake_user)):
return user

“`

main.py 更新：

“`python

main.py

from fastapi import FastAPI

从 routers 模块导入创建的 APIRouter 实例

from .routers import items, users # 注意这里的相对导入，实际项目中可能使用绝对导入或 sys.path

创建 FastAPI 应用实例

app = FastAPI()

使用 include_router 将 router 包含到主应用中

prefix 和 tags 在 APIRouter 中定义了，这里可以直接包含

app.include_router(items.router)
app.include_router(users.router)

可以保留根路径等全局性的路径操作

@app.get(“/”)
async def read_root():
return {“Hello”: “World”}

可以定义全局依赖 (例如应用到所有路径操作)

async def global_dependency():

pass

app = FastAPI(dependencies=[Depends(global_dependency)])

全局异常处理等也可以在这里定义

“`

项目结构示例：

your_project/
├── main.py
├── routers/
│ ├── __init__.py # 可以是空文件，或者用于导入 router 实例
│ ├── items.py
│ ├── users.py
└── requirements.txt # 包含 fastapi 和 uvicorn

运行：

确保你在 your_project 目录下，运行 uvicorn main:app --reload。

访问 http://127.0.0.1:8000/docs，你会看到接口文档按照 items 和 users 标签进行了分组，并且路径前缀也正确应用了 (/items/..., /users/...)。这种方式使得大型项目的代码结构清晰、易于管理。

9. 配置管理

在实际应用中，你需要处理数据库连接字符串、密钥、外部服务地址等配置信息。将这些信息硬编码在代码中是不安全的，也不灵活。推荐使用环境变量或配置文件来管理配置。

Pydantic 提供了一个 Settings 类（在 Pydantic v2+ 中需要安装 pydantic-settings，v1 中是 Pydantic 自带的 Settings 类）可以方便地从环境变量中读取配置。

bash
pip install pydantic-settings

创建一个 config.py 文件：

“`python

config.py

from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

class Settings(BaseSettings):
# 定义配置字段，并提供类型提示
# 字段名会自动转换为大写，并从环境变量中读取，例如 SECRET_KEY, DATABASE_URL
secret_key: str = “your-secret-key” # 可以设置默认值
database_url: str = “sqlite:///./sql_app.db”
debug: bool = False # 从 DEBUG 环境变量读取，默认为 False
app_name: str = “Awesome API”

# v2+ 版本使用 model_config
model_config = SettingsConfigDict(
    env_file=".env",       # 允许从 .env 文件加载配置
    env_file_encoding='utf-8', # 指定 .env 文件编码
    case_sensitive=False   # 环境变量名不区分大小写
)

使用 lru_cache 缓存 settings 实例，避免重复加载

@lru_cache
def get_settings():
return Settings()

“`

在项目根目录创建 .env 文件（可选）：

“`dotenv

.env

SECRET_KEY=my-actual-secret-key
DATABASE_URL=postgresql://user:password@host:port/dbname
DEBUG=True
“`

在 main.py 或其他需要配置的地方使用 get_settings() 依赖：

“`python

main.py (或其他文件)

from fastapi import FastAPI, Depends
from .config import Settings, get_settings # 导入配置

创建 FastAPI 应用实例

可以将 app_name 从 settings 中获取

app = FastAPI(title=get_settings().app_name, debug=get_settings().debug)

包含路由等…

from .routers import items, users

app.include_router(items.router)

app.include_router(users.router)

可以在任何路径操作中使用 get_settings() 依赖来获取配置

@app.get(“/info”)
async def read_info(settings: Settings = Depends(get_settings)):
return {
“app_name”: settings.app_name,
“debug_mode”: settings.debug,
# 注意：不要在这里返回敏感信息，如 secret_key 或 database_url
}

“`

运行应用时，Pydantic 会先尝试从 .env 文件加载配置，然后尝试从环境变量加载。环境变量优先级更高。

10. 错误处理

FastAPI 会自动处理一些常见的错误，比如 Pydantic 数据校验失败（返回 422）和 HTTPException。你可以通过捕获 HTTPException 来返回自定义的错误响应。

“`python

main.py

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

… 其他路径操作 …

模拟一个需要 ID 的资源

@app.get(“/items/{item_id}”)
async def read_item(item_id: int):
if item_id != 42:
# 如果 item_id 不是 42，抛出 HTTPException
# detail 会作为 JSON 响应体中的 “detail” 字段
raise HTTPException(status_code=404, detail=”Item not found”)
return {“item_id”: item_id}

你也可以自定义全局或特定的异常处理器

from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationException # Pydantic 校验异常的基类

示例：捕获 RequestValidationException (Pydantic 校验失败) 并返回自定义格式

@app.exception_handler(RequestValidationException)
async def validation_exception_handler(request, exc):
# exc 是 RequestValidationException 实例，包含错误的详细信息
# exc.errors() 返回一个包含错误详情的列表
# exc.body 包含原始请求体数据
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={
“message”: “Validation failed”,
“errors”: exc.errors(), # 包含具体的字段错误信息
“body”: exc.body,
},
)

示例：捕获通用的 HTTPException 并添加额外信息

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
# exc 是 HTTPException 实例
return JSONResponse(
status_code=exc.status_code,
content={“message”: exc.detail, “custom_field”: “some_value”},
headers=exc.headers # 可以保留原始的头信息
)

注意：自定义处理器会覆盖 FastAPI 默认的处理器

如果你想在处理后保留默认行为，需要重新引发异常或手动构建响应

“`

11. 数据库集成 (简述)

FastAPI 本身不绑定任何特定的 ORM 或数据库。由于其异步特性，与异步 ORM（如 SQLAlchemy 的异步版本、Ormar、Tortoise ORM）结合是常见的做法。

集成数据库时，依赖注入系统再次发挥了关键作用。你可以创建一个依赖函数，负责获取一个数据库会话（Session），并在请求处理完毕后关闭它。

“`python

database.py (示例，需要安装数据库驱动和 SQLAlchemy)

pip install sqlalchemy asyncpg # asyncpg for PostgreSQL

pip install sqlalchemy aiosqlite # aiosqlite for SQLite async

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.declarative import declarative_base # or use SQLAlchemy 2.0 style
from sqlalchemy.orm import sessionmaker

假设你在 config.py 中定义了 DATABASE_URL

from .config import get_settings

settings = get_settings()

创建异步引擎

engine = create_async_engine(
settings.database_url, echo=settings.debug # echo=True 会打印 SQL 语句
)

创建异步会话工厂

autocommit=False 表示手动提交事务

autoflush=False 表示手动刷新会话

bind=engine 绑定到异步引擎

AsyncSessionLocal = sessionmaker(
bind=engine, class_=AsyncSession, expire_on_commit=False, autocommit=False, autoflush=False
)

创建一个用于依赖注入的函数，获取并管理数据库会话

async def get_db():
db = AsyncSessionLocal()
try:
yield db # 使用 yield 关键字，FastAPI 会在请求结束后关闭依赖
finally:
await db.close() # 确保会话被关闭

定义模型 (略)

Base = declarative_base()

class User(Base):

tablename = “users”

id = Column(Integer, primary_key=True, index=True)

email = Column(String, unique=True, index=True)

hashed_password = Column(String)

is_active = Column(Boolean, default=True)

在应用启动时创建所有表 (可选)

async def init_db():
async with engine.begin() as conn:
# await conn.run_sync(Base.metadata.create_all) # 如果使用声明式基类

在 main.py 或其他 router 中使用 get_db 依赖：

from fastapi import FastAPI, Depends

from .database import get_db, AsyncSession

from . import crud # 假设你有一个 crud.py 文件用于数据库操作

@app.post(“/users/”)

async def create_user(user: UserCreate, db: AsyncSession = Depends(get_db)):

db_user = await crud.get_user_by_email(db, email=user.email)

if db_user:

raise HTTPException(status_code=400, detail=”Email already registered”)

return await crud.create_user(db, user=user)

“`

这种模式（依赖注入一个 DB Session）是 FastAPI 中处理数据库连接的标准且推荐的方式，它使得数据库操作的代码与路径操作逻辑解耦，并且易于管理会话生命周期。

12. 单元测试

FastAPI 提供了 TestClient 类，基于 requests 库，可以方便地对你的应用进行同步测试。

bash
pip install httpx pytest

httpx 是一个异步 HTTP 客户端库，FastAPI 的 TestClient 就是基于它实现的。

创建一个 test_main.py 文件：

“`python

test_main.py

from fastapi.testclient import TestClient
from .main import app # 导入你的 FastAPI 应用实例

创建 TestClient 实例

client = TestClient(app)

编写测试函数

def test_read_main():
# 使用 client 发起请求，就像使用 requests 一样
response = client.get(“/”)
# 验证响应状态码
assert response.status_code == 200
# 验证响应体
assert response.json() == {“Hello”: “World”}

def test_read_item():
item_id = 123
response = client.get(f”/items/{item_id}”)
assert response.status_code == 200
assert response.json() == {“item_id”: item_id, “q”: None}

def test_read_item_with_query():
item_id = 456
query = “testquery”
response = client.get(f”/items/{item_id}?q={query}”)
assert response.status_code == 200
assert response.json() == {“item_id”: item_id, “q”: query}

def test_create_item():
new_item = {“name”: “New Item”, “price”: 25.5}
response = client.post(“/items/”, json=new_item) # 使用 json 参数发送 JSON 请求体
assert response.status_code == 200
# 验证响应体 (可能需要根据你的 create_item 函数返回的内容调整)
assert response.json() == {“item_name”: “New Item”, “item_price”: 25.5}

def test_create_item_invalid_data():
invalid_item = {“name”: “Invalid”} # 缺少 price
response = client.post(“/items/”, json=invalid_item)
assert response.status_code == 422 # 期望 Pydantic 校验失败返回 422
# 可以进一步验证响应体中的错误详情

你可以使用 pytest 运行这些测试

进入项目目录，运行 pytest

“`

使用 pytest 命令运行测试：

bash
pytest

TestClient 使得测试 FastAPI 应用变得非常直观和方便，你可以像发起实际 HTTP 请求一样来测试你的路径操作。

13. 部署 (简述)

部署 FastAPI 应用通常涉及以下步骤：

  1. 安装依赖： 确保生产环境中安装了你的应用所需的所有 Python 包，包括 fastapi 和 ASGI 服务器（如 uvicorn 或 hypercorn）。
  2. 使用生产级 ASGI 服务器： 在生产环境不要使用 uvicorn --reload。使用像 uvicorn main:app 这样的命令。为了充分利用多核 CPU，通常会在 Uvicorn 前面加上 Gunicorn 等进程管理器，或者直接使用 Uvicorn 的 Worker 参数：uvicorn main:app --workers 4。
  3. 反向代理： 在 ASGI 服务器前面放置一个反向代理服务器，如 Nginx 或 Traefik。反向代理可以处理 SSL 终止、负载均衡、静态文件服务、缓存和基本的安全防护。
  4. 容器化： 使用 Docker 是一个非常流行的部署方式。创建一个 Dockerfile，将你的应用和依赖打包到一个容器中。
  5. 平台选择： 可以部署到各种平台，如云虚拟机 (AWS EC2, GCP Compute Engine)、容器服务 (Docker Swarm, Kubernetes)、PaaS 平台 (Heroku, Google App Engine)、无服务器平台 (AWS Lambda + API Gateway, Google Cloud Functions + API Gateway)。

FastAPI 本身轻量且高性能，非常适合部署在各种环境中。

14. 总结

FastAPI 是一个为构建 API 而生的现代、高性能 Python Web 框架。通过本文的学习，你应该已经掌握了 FastAPI 的核心概念和基本用法：

  • 快速入门： 创建并运行了你的第一个 FastAPI 应用。
  • 路由与参数： 学会了如何定义不同 HTTP 方法的路径操作，以及如何处理路径参数和查询参数。
  • 数据校验与请求体： 深入理解了 Pydantic 在请求体处理和数据校验中的强大作用。
  • 依赖注入： 掌握了 FastAPI 依赖注入系统的基本用法，以及它如何帮助组织代码和管理资源。
  • 项目结构： 了解了如何使用 APIRouter 来构建模块化的应用。
  • 实用实践： 探讨了配置管理、错误处理、数据库集成和测试等方面的基础知识。
  • 自动文档： 体验了 FastAPI 自动生成的交互式 API 文档。

FastAPI 提供了构建现代化、高性能 API 所需的一切工具。它的设计哲学强调开发者体验和代码效率，使得编写健壮的 API 变得愉快而高效。

这仅仅是 FastAPI 功能的冰山一角。FastAPI 还提供了包括安全性（OAuth2, JWT）、背景任务、WebSockets、测试客户端等更多高级特性。鼓励你继续深入学习官方文档，并在实际项目中不断实践，探索 FastAPI 的更多可能性。

祝你在使用 FastAPI 的旅程中取得成功！

发表评论

您的邮箱地址不会被公开。 必填项已用 * 标注

滚动至顶部