---

FastAPI 核心概念与实战指南：构建现代化、高性能的 Python API

FastAPI 凭借其卓越的性能、易用性以及对现代 Python 特性（如类型提示）的极致运用，已迅速成为 Python Web 框架领域的一颗璀璨新星。它由 Sebastián Ramírez (tiangolo) 开发，旨在帮助开发者快速构建健壮、高效且文档齐全的 API。本文将深入探讨 FastAPI 的核心概念，并通过实战代码引导您入门，助您掌握这一强大的工具。

FastAPI 简介：为何选择它？

在深入细节之前，我们先来看看 FastAPI 为何备受青睐：

1. 高性能: FastAPI 基于 Starlette (ASGI 框架) 和 Pydantic (数据验证库)，性能与 NodeJS 和 Go 的一些框架相当，远超 Flask 和 Django (在某些基准测试中)。
2. 快速开发: 得益于其直观的 API 设计和强大的类型提示支持，开发速度可以提升 200% 到 300%。
3. 更少 Bug: 类型提示使得编辑器和静态分析工具能在编码阶段就捕获大量错误，减少了约 40% 的人为错误。
4. 智能编辑器支持: 完善的类型提示为 IDE (如 VS Code, PyCharm) 提供了卓越的自动补全和类型检查功能。
5. 易学易用: FastAPI 的设计哲学是简单直观，同时拥有完善的文档。
6. 代码简短: 最小化代码重复，每个参数声明都有多重作用。
7. 健壮性: 自动生成生产可用的代码，并提供交互式文档。
8. 标准驱动: 完全兼容并基于 OpenAPI (前身为 Swagger) 和 JSON Schema 标准。

环境准备与安装

在开始之前，确保您已安装 Python 3.7+。然后，通过 pip 安装 FastAPI 和一个 ASGI 服务器，如 Uvicorn：

bash
pip install fastapi uvicorn[standard]
uvicorn[standard] 会安装 Uvicorn 以及一些推荐的依赖，如 httptools (用于更快的 HTTP 解析) 和 websockets (如果需要 WebSocket 支持)。

核心概念详解

1. 第一个 FastAPI 应用

让我们从一个简单的 “Hello World” 应用开始：

“`python

main.py

from fastapi import FastAPI

app = FastAPI()

@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}
“`

• FastAPI(): 创建一个 FastAPI 应用实例。
• @app.get("/"): 这是一个路径操作装饰器。get 指的是 HTTP GET 方法。/ 是路径。
• async def: FastAPI 支持异步函数 (async def) 和普通函数 (def)。异步函数允许应用处理并发请求而不会阻塞，特别适合 I/O 密集型操作。
• read_root(): 当访问根路径 / 时，此函数将被调用。
• read_item(item_id: int, q: str | None = None):
  • item_id: int: 这是一个路径参数。FastAPI 会自动将路径中的 {item_id} 部分转换为整数。类型提示 int 不仅用于文档，还用于数据验证和转换。
  • q: str | None = None: 这是一个查询参数。如果 URL 是 /items/5?q=somequery，q 的值就是 "somequery"。类型提示 str | None (或 Python 3.10+ 的 str | None) 表示 q 是一个可选的字符串。 = None 提供了默认值。

运行应用：
在终端中，进入 main.py 所在的目录，然后运行：
bash
uvicorn main:app --reload
* main: 指的是 main.py 文件。
* app: 指的是在 main.py 中创建的 FastAPI 实例 app。
* --reload: 使服务器在代码更改后自动重启，非常适合开发。

现在，打开浏览器访问 http://127.0.0.1:8000/ 和 http://127.0.0.1:8000/items/5?q=test_query。

2. 自动交互式 API 文档

FastAPI 的一大亮点是自动生成的 API 文档。启动应用后，访问：
* http://127.0.0.1:8000/docs: 查看 Swagger UI 提供的交互式文档。
* http://127.0.0.1:8000/redoc: 查看 ReDoc 提供的替代文档。

这些文档是根据您的代码（包括路径、参数、类型提示和 Pydantic 模型）自动生成的。您甚至可以直接在 Swagger UI 中测试您的 API 端点。

3. 路径参数 (Path Parameters)

路径参数是 URL 路径的一部分，用花括号 {} 包裹。

“`python
from fastapi import FastAPI

app = FastAPI()

@app.get(“/users/{user_id}”)
async def get_user(user_id: str):
return {“user_id”: user_id}

@app.get(“/files/{file_path:path}”)
async def read_file(file_path: str):
return {“file_path”: file_path}
``
*user_id: str: 声明user_id为字符串类型。
*file_path:path: 特殊的路径参数类型:path允许参数值包含/。例如，/files/home/user/docs/report.txt中的home/user/docs/report.txt会被捕获为file_path` 的值。

FastAPI 会使用类型提示进行数据验证。如果期望 int 却传入了非数字字符串，会自动返回 422 Unprocessable Entity 错误。

4. 查询参数 (Query Parameters)

查询参数是 URL 中 ? 之后的部分，以 key=value 形式出现，多个参数用 & 分隔。

“`python
from fastapi import FastAPI

app = FastAPI()

http://127.0.0.1:8000/search?query=python&limit=10

@app.get(“/search”)
async def search_items(query: str, limit: int = 10, skip: int = 0):
# 假设 items 是一个包含数据的列表
# results = items[skip : skip + limit]
return {“query”: query, “limit”: limit, “skip”: skip, “results”: []}

可选参数

@app.get(“/items_optional/”)
async def read_items_optional(q: str | None = None):
if q:
return {“message”: f”Querying for: {q}”}
return {“message”: “No query provided”}

布尔类型转换

@app.get(“/process/”)
async def process_data(short: bool = False):
if short:
return {“message”: “Processing short version”}
return {“message”: “Processing full version”}

访问 /process/?short=1, /process/?short=True, /process/?short=true, /process/?short=on 都会被视为 True

``
* 函数参数如果不是路径参数，并且没有从请求体中获取 (下面会讲)，那么 FastAPI 会将其视为查询参数。
*limit: int = 10:limit是一个整数查询参数，默认值为10。
*q: str | None = None:q` 是一个可选的字符串查询参数。
* 类型转换会自动进行。

5. 请求体 (Request Body) 与 Pydantic 模型

当需要客户端向 API 发送数据时（例如创建或更新资源），通常使用请求体。FastAPI 使用 Pydantic 模型来定义、验证和文档化请求体。

“`python
from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import List, Optional

app = FastAPI()

class Item(BaseModel):
name: str
description: str | None = Field(
default=None, title=”The description of the item”, max_length=300
)
price: float = Field(gt=0, description=”The price must be greater than zero”)
tax: float | None = None
tags: List[str] = []

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

@app.post(“/items/”)
async def create_item(item: Item):
item_dict = item.model_dump() # Pydantic v2, or item.dict() in v1
if item.tax:
price_with_tax = item.price + item.tax
item_dict.update({“price_with_tax”: price_with_tax})
return item_dict

@app.put(“/items/{item_id}”)
async def update_item(item_id: int, item: Item, user: User, importance: int | None = None):
# item_id 是路径参数
# item 是来自请求体的 Item 模型
# user 也是来自请求体的 User 模型 (如果API允许多个body参数, FastAPI会处理)
# importance 是可选的查询参数
results = {“item_id”: item_id, “item”: item, “user”: user}
if importance:
results.update({“importance”: importance})
return results
``
*BaseModel: 从pydantic导入，用于创建数据模型。
*Item类定义了期望的请求体结构及其字段类型。
*Field: 可以为模型字段提供额外的元数据和验证规则，如default,title,description,gt(greater than),max_length等。
*create_item(item: Item): FastAPI 会自动将请求的 JSON body 解析、验证并转换为Item类的实例。如果数据无效，会自动返回详细的 422 错误。
*item.model_dump()`: 将 Pydantic 模型实例转换为字典。

FastAPI 的强大之处在于，Pydantic 模型不仅用于输入验证，还用于输出序列化（响应模型）和自动 API 文档生成。

6. 数据校验 (Validation)

Pydantic 和 FastAPI 提供了强大的数据校验能力。
* 路径参数和查询参数校验:
“`python
from fastapi import FastAPI, Query, Path
from typing import Annotated

app = FastAPI()

@app.get("/products/")
async def read_products(
    q: Annotated[str | None, Query(min_length=3, max_length=50, pattern="^fixedquery$")] = None,
    page: Annotated[int, Query(ge=1)] = 1, # ge: greater than or equal
    size: Annotated[int, Query(gt=0, le=100)] = 10 # gt: greater than, le: less than or equal
):
    return {"q": q, "page": page, "size": size}

@app.get("/widgets/{widget_id}")
async def read_widget(
    widget_id: Annotated[int, Path(title="The ID of the widget to get", ge=1, le=1000)],
    q: Annotated[str | None, Query(alias="item-query")] = None # alias 用于查询参数的别名
):
    results = {"widget_id": widget_id}
    if q:
        results.update({"q": q})
    return results
```
*   `Query` 和 `Path` 用于为查询参数和路径参数添加额外的校验和元数据。
*   `Annotated[type, Meta(...)]` 是 Python 3.9+ 的标准方式，用于附加元数据。FastAPI 推荐使用这种方式。
*   `min_length`, `max_length`, `pattern` (正则表达式), `ge`, `gt`, `le`, `lt` 等都是可用的校验。
*   `alias`: 允许在 API 中使用一个名称，而在代码中使用另一个（通常是 Pythonic 的）名称。例如，URL 中的 `item-query` 会映射到函数参数 `q`。

• 请求体校验: Pydantic 模型本身就执行严格的校验。

7. 响应模型 (Response Model)

虽然 FastAPI 可以自动序列化返回的字典、列表或 Pydantic 模型，但有时您希望确保响应具有特定的结构和数据类型，或者过滤掉一些敏感数据。这时可以使用 response_model 参数。

“`python
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

app = FastAPI()

class UserIn(BaseModel):
username: str
password: str
email: EmailStr
full_name: str | None = None

class UserOut(BaseModel):
username: str
email: EmailStr
full_name: str | None = None

假设这是你的数据库模型 (仅为示例)

class UserInDB(UserIn):
hashed_password: str

def fake_save_user(user_in: UserIn):
hashed_password = user_in.password + “notreallyhashed” # 实际应用中应使用安全的哈希算法
user_in_db = UserInDB(**user_in.model_dump(), hashed_password=hashed_password)
return user_in_db

@app.post(“/users/”, response_model=UserOut)
async def create_user(user: UserIn):
user_saved = fake_save_user(user)
# 即使 user_saved 包含 hashed_password,
# 由于 response_model=UserOut, FastAPI 会自动过滤掉它
return user_saved
``
*response_model=UserOut: 告诉 FastAPI 响应应该符合UserOut模型的结构。
* FastAPI 会使用UserOut来：
* 验证输出数据。
* 序列化数据 (例如，将datetime对象转换为 ISO 格式字符串)。
* 过滤数据，只包括UserOut` 中定义的字段。
* 在 OpenAPI 文档中记录响应结构。

8. 依赖注入 (Dependency Injection)

FastAPI 的依赖注入系统非常强大且易于使用。它允许您声明函数运行所需的依赖项，FastAPI 会负责处理这些依赖项的解析和传递。

“`python
from fastapi import FastAPI, Depends, HTTPException, status
from typing import Annotated

app = FastAPI()

async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {“q”: q, “skip”: skip, “limit”: limit}

类型别名，使依赖注入更清晰

CommonsDep = Annotated[dict, Depends(common_parameters)]

@app.get(“/items_dep/”)
async def read_items_dep(commons: CommonsDep):
# commons 现在是 common_parameters 函数返回的字典
# e.g., {“q”: “search_term”, “skip”: 0, “limit”: 10}
return {“message”: “Items fetched”, “params”: commons}

@app.get(“/users_dep/”)
async def read_users_dep(commons: CommonsDep):
return {“message”: “Users fetched”, “params”: commons}

依赖项也可以是类

class CommonQueryParams:
def init(self, q: str | None = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit

@app.get(“/things/”)
async def read_things(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):

或者更简洁: async def read_things(commons: Annotated[CommonQueryParams, Depends()]):

FastAPI 会将 CommonQueryParams 视为可调用对象并创建实例

return {"message": "Things fetched", "params": commons}

依赖项可以依赖其他依赖项

依赖项可以返回值，也可以不返回值 (例如，仅执行某些操作)

依赖项可以是异步函数或普通函数

示例：模拟数据库连接

async def get_db_session():
print(“Simulating DB session start”)
db = {“data”: “fake_db_connection”}
try:
yield db # yield 使其成为一个上下文管理器式的依赖
finally:
print(“Simulating DB session close”)

DBSessionDep = Annotated[dict, Depends(get_db_session)]

@app.get(“/data/”)
async def get_data_from_db(db: DBSessionDep):
return {“data_from_db”: db[“data”]}

``
*Depends: 标记一个参数为依赖项。
*common_parameters: 一个普通函数，被用作依赖项。FastAPI 会在调用read_items_dep或read_users_dep之前，先调用common_parameters，并将结果传递给commons参数。
* 依赖注入的好处：
* **代码复用**: 共享逻辑（如分页参数、数据库会话、用户认证）。
* **关注点分离**: 路径操作函数专注于业务逻辑，依赖项处理共享任务。
* **易于测试**: 可以轻松地替换或模拟依赖项。
* 使用yield` 的依赖项可以执行一些清理操作（如关闭数据库连接），类似于 Python 的上下文管理器。

9. 安全性 (Security Utilities)

FastAPI 提供了一些工具来处理常见的安全需求，如 OAuth2、API Keys 等。

以 OAuth2 密码流（Password Flow）为例：
“`python
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from typing import Annotated

app = FastAPI()

这个对象告诉 FastAPI 从哪个 URL 获取 token

oauth2_scheme = OAuth2PasswordBearer(tokenUrl=”token”)

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

class UserInDB(User):
hashed_password: str

模拟用户数据库

fake_users_db = {
“johndoe”: {
“username”: “johndoe”,
“full_name”: “John Doe”,
“email”: “[email protected]”,
“hashed_password”: “fakehashedpassword”, # 真实应用中应为安全哈希
“disabled”: False,
}
}

def get_user(db, username: str):
if username in db:
user_dict = db[username]
return UserInDB(**user_dict)
return None

模拟验证用户

def fake_decode_token(token: str):
# 在真实应用中，这里会解码并验证 JWT token
user = get_user(fake_users_db, token) # 简单示例，token 就是用户名
return user

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
user = fake_decode_token(token)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=”Invalid authentication credentials”,
headers={“WWW-Authenticate”: “Bearer”},
)
if user.disabled:
raise HTTPException(status_code=400, detail=”Inactive user”)
return user

CurrentUserDep = Annotated[User, Depends(get_current_user)]

Token 端点 (通常是 POST)

@app.post(“/token”)
async def login_for_access_token(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
# 实际应用中:
# 1. 从 form_data.username 和 form_data.password 验证用户
# 2. 如果验证成功，创建 JWT token
user = get_user(fake_users_db, form_data.username)
if not user or user.hashed_password != form_data.password + “fakehashed”: # 简化密码检查
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=”Incorrect username or password”,
headers={“WWW-Authenticate”: “Bearer”},
)
# 简单示例：token 就是用户名
access_token = user.username
return {“access_token”: access_token, “token_type”: “bearer”}

@app.get(“/users/me/”, response_model=User)
async def read_users_me(current_user: CurrentUserDep):
return current_user

@app.get(“/users/me/items/”)
async def read_own_items(current_user: CurrentUserDep):
return [{“item_id”: “Foo”, “owner”: current_user.username}]
``
*OAuth2PasswordBearer(tokenUrl=”token”): 创建一个 OAuth2 "password flow" 的实例。tokenUrl是客户端获取 token 的相对路径。
*Depends(oauth2_scheme): 在需要认证的路径操作函数中，将此依赖注入。FastAPI 会查找请求头中的Authorization: Bearer ，并将其传递给依赖。
*get_current_user: 这是一个依赖项，它接收 token，验证它，并返回用户信息。如果 token 无效或用户不存在/不活跃，则抛出 HTTP 异常。
*OAuth2PasswordRequestForm: 用于/token端点，自动解析username和password(通常来自表单数据)。
* 现在，访问/docs`，你会看到一个 “Authorize” 按钮，可以输入 token 进行认证。

10. 中间件 (Middleware)

中间件是在每个请求被特定路径操作处理之前，以及每个响应被发送回客户端之前运行的函数。

“`python
import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware(“http”)
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() – start_time
response.headers[“X-Process-Time”] = str(process_time)
return response

@app.get(“/test-middleware/”)
async def test_middleware_endpoint():
return {“message”: “Middleware test successful”}
``
*@app.middleware(“http”): 注册一个 HTTP 中间件。
*call_next(request): 调用链中的下一个处理程序（可能是另一个中间件，或最终的路径操作函数）。
* 这个例子中，中间件计算请求处理时间，并将其添加到响应头X-Process-Time` 中。

11. 错误处理 (Error Handling)

FastAPI 自动处理 HTTPException 并返回相应的 HTTP 错误响应。您也可以自定义异常处理器。

“`python
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel

class ItemNotFoundException(Exception):
def init(self, item_id: str):
self.item_id = item_id

app = FastAPI()

items_db = {“foo”: {“name”: “Foo”}, “bar”: {“name”: “Bar”}}

@app.exception_handler(ItemNotFoundException)
async def item_not_found_exception_handler(request: Request, exc: ItemNotFoundException):
return JSONResponse(
status_code=404,
content={“message”: f”Oops! Item with ID ‘{exc.item_id}’ not found.”},
)

@app.get(“/items_error/{item_id}”)
async def read_item_error(item_id: str):
if item_id not in items_db:
raise ItemNotFoundException(item_id=item_id)
# 或者直接使用 FastAPI 的 HTTPException
# raise HTTPException(status_code=404, detail=”Item not found”)
return items_db[item_id]
``
*@app.exception_handler(ItemNotFoundException): 注册一个处理器来捕获ItemNotFoundException类型的异常。
* 当路径操作函数中raise ItemNotFoundException(…)时，item_not_found_exception_handler` 会被调用。

12. 后台任务 (Background Tasks)

对于不需要立即完成的操作（如发送邮件通知），可以使用后台任务。

“`python
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_notification(email: str, message=””):
with open(“log.txt”, mode=”a”) as email_file:
content = f”Notification for {email}: {message}\n”
email_file.write(content)

@app.post(“/send-notification/{email}”)
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message=”Some notification”)
return {“message”: “Notification will be sent in the background”}
``
*BackgroundTasks: 参数类型，FastAPI 会自动注入一个实例。
*background_tasks.add_task(func, arg1, arg2, …)`: 添加一个任务到后台执行。该任务会在响应发送后运行。

13. 应用结构与 APIRouter

当应用变大时，将所有路径操作放在一个文件中会变得难以管理。APIRouter 允许您将路径操作组织到多个模块中。

“`python

./routers/items.py

from fastapi import APIRouter
from pydantic import BaseModel

router = APIRouter(
prefix=”/items”,
tags=[“items”], # 在 OpenAPI 文档中分组
responses={404: {“description”: “Not found”}}, # 可以为整个 router 设置通用响应
)

class Item(BaseModel):
name: str
price: float

fake_items_db = {“plumbus”: {“name”: “Plumbus”, “price”: 3.50}}

@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:
# 将使用 router 级别定义的 404 响应
return {“error”: “Item not found”} # 或者 raise HTTPException
return {item_id: fake_items_db[item_id]}

./main.py

from fastapi import FastAPI
from .routers import items # 假设 items.py 在同级 routers 目录下

app = FastAPI()

app.include_router(items.router)

@app.get(“/”)
async def root():
return {“message”: “Hello from main app”}
``
* 创建APIRouter实例，可以指定prefix（所有路由都以此开头）、tags（用于文档分组）等。
* 在主应用中使用app.include_router(router_instance)` 来包含这些路由。

实战技巧与最佳实践

1. 充分利用类型提示: 这是 FastAPI 的核心。它不仅能帮助你和你的团队，还能让 FastAPI 为你做更多事（验证、序列化、文档）。
2. Pydantic 模型是你的朋友: 用它们来定义清晰的数据结构，进行输入输出验证。学习 Field 的高级用法。
3. 异步优先: 对于 I/O 密集型操作（数据库查询、外部 API 调用），使用 async 和 await 来提高并发性能。如果使用了会阻塞的库，可以考虑使用 from starlette.concurrency import run_in_threadpool。
4. 依赖注入 DRY (Don’t Repeat Yourself): 将共享逻辑（如数据库会话管理、认证）封装到依赖项中。
5. 合理组织项目结构: 对于大型应用，使用 APIRouter 将功能模块化。

6. 编写测试: FastAPI 与 Pytest 配合良好。使用 TestClient 来测试你的 API。
   “`python
   # tests/test_main.py
   from fastapi.testclient import TestClient
   from ..main import app # 调整导入路径

   client = TestClient(app)

   def test_read_root():
   response = client.get(“/”)
   assert response.status_code == 200
   assert response.json() == {“Hello”: “World”}

   def test_read_item():
   response = client.get(“/items/5?q=somequery”)
   assert response.status_code == 200
   assert response.json() == {“item_id”: 5, “q”: “somequery”}
   ``
7. **配置管理**: 使用环境变量或配置文件 (如 Pydantic 的BaseSettings) 来管理应用配置。
8. **日志记录**: 配置 Python 的logging模块以记录应用活动和错误。
9. **版本控制你的 API**: 例如，使用路径前缀/v1/items或/v2/items`。
   10. 持续学习: FastAPI 社区活跃，文档优秀。关注官方文档和 GitHub 仓库的更新。

总结

FastAPI 提供了一个现代化、高效且愉悦的 Python API 开发体验。它通过巧妙地结合 Python 类型提示、Pydantic 的数据验证能力以及 Starlette 的 ASGI 性能，为开发者带来了前所未有的便利。从简单的路径操作到复杂的依赖注入和安全机制，FastAPI 都提供了简洁而强大的解决方案。

本文涵盖了 FastAPI 的核心概念和一些实战技巧，希望能为您构建高性能 API 提供坚实的基础。随着您实践的深入，您会发现 FastAPI 还有更多值得探索的特性，如 WebSocket 支持、GraphQL (通过 Strawberry 或 Ariadne)、数据库集成 (如 SQLModel 或 SQLAlchemy) 等。现在，开始用 FastAPI 构建您的下一个出色 API 吧！

---