使用 FastAPI 构建 API:详细教程与示例
在现代 Web 开发中,构建快速、高效且易于维护的 API 至关重要。FastAPI 是一个现代、高性能的 Web 框架,用于构建 API。它基于标准 Python 类型提示,自动生成 API 文档,并提供开箱即用的数据验证。本教程将深入探讨如何使用 FastAPI 构建 API,涵盖从基本概念到高级特性的各个方面,并提供丰富的示例代码。
一、FastAPI 简介与优势
FastAPI 是一个由 Sebastián Ramírez 开发的 Python Web 框架,旨在简化 API 构建过程,同时保证高性能。 它的设计哲学主要围绕以下几个核心优势:
- 快速开发: FastAPI 显著减少了开发 API 所需的时间。 类型提示和自动文档生成等功能极大地提高了开发效率。
- 高性能: FastAPI 基于 ASGI (Asynchronous Server Gateway Interface), 可以充分利用现代 Web 服务器的异步特性,从而实现卓越的性能。 它超越了传统的 Flask 和 Django 等框架。
- 强大的数据验证: FastAPI 使用 Pydantic 进行数据验证。 Pydantic 允许你使用 Python 类型提示来定义数据模型,并自动验证传入数据是否符合这些模型。
- 自动 API 文档生成: FastAPI 自动生成基于 OpenAPI 和 JSON Schema 的 API 文档。 你可以通过 Swagger UI 或 ReDoc 轻松查看和测试你的 API。
- 易于学习和使用: FastAPI 建立在标准的 Python 类型提示之上, 易于理解和掌握。 官方文档清晰全面, 提供了丰富的示例。
- 与现有技术兼容: FastAPI 可以与多种现有技术集成,包括数据库、身份验证库和测试框架。
二、安装与配置
首先,确保你已经安装了 Python 3.7 或更高版本。 然后,使用 pip 安装 FastAPI:
bash
pip install fastapi
FastAPI 本身只是一个框架,还需要一个 ASGI 服务器来运行你的应用程序。 我们推荐使用 Uvicorn:
bash
pip install uvicorn
三、第一个 FastAPI 应用程序:Hello World
创建一个名为 main.py
的文件,并输入以下代码:
“`python
from fastapi import FastAPI
app = FastAPI()
@app.get(“/”)
async def read_root():
return {“Hello”: “World”}
“`
代码解释:
from fastapi import FastAPI
: 导入 FastAPI 类。app = FastAPI()
: 创建一个 FastAPI 应用程序实例。@app.get("/")
: 一个装饰器,用于定义一个处理 HTTP GET 请求的路径操作函数。 这里的/
是根路径。async def read_root():
: 一个异步函数,用于处理对根路径的 GET 请求。async
关键字表示这是一个异步函数, 允许 FastAPI 并发处理多个请求。return {"Hello": "World"}
: 函数返回一个 JSON 响应。
运行应用程序:
在终端中,导航到包含 main.py
文件的目录,并运行以下命令:
bash
uvicorn main:app --reload
命令解释:
uvicorn
: 启动 Uvicorn ASGI 服务器。main:app
: 指定要运行的 FastAPI 应用程序。main
是包含 FastAPI 应用程序的 Python 文件名,app
是 FastAPI 应用程序实例的名称。--reload
: 启用自动重载, 当代码更改时,服务器会自动重启。
现在,打开你的 Web 浏览器,访问 http://127.0.0.1:8000/
。 你应该看到以下 JSON 响应:
json
{
"Hello": "World"
}
查看 API 文档:
FastAPI 会自动生成 API 文档。 访问 http://127.0.0.1:8000/docs
查看 Swagger UI 文档,或访问 http://127.0.0.1:8000/redoc
查看 ReDoc 文档。
四、路径参数与查询参数
FastAPI 允许你使用路径参数和查询参数来接收来自客户端的数据。
路径参数:
路径参数是 URL 路径的一部分,用于标识特定的资源。
“`python
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("/items/{item_id}")
: 定义一个处理对/items/{item_id}
的 GET 请求的路径操作函数。{item_id}
是一个路径参数, 用于从 URL 中提取 item 的 ID。async def read_item(item_id: int):
: 路径操作函数接收一个名为item_id
的参数。item_id: int
表示item_id
必须是整数类型。 FastAPI 会自动验证传入的参数是否符合类型提示。
访问 http://127.0.0.1:8000/items/5
将返回:
json
{
"item_id": 5
}
查询参数:
查询参数出现在 URL 的问号 (?
) 之后,用于传递可选的参数。
“`python
from fastapi import FastAPI
app = FastAPI()
@app.get(“/items/”)
async def read_item(skip: int = 0, limit: int = 10):
return {“skip”: skip, “limit”: limit}
“`
代码解释:
@app.get("/items/")
: 定义一个处理对/items/
的 GET 请求的路径操作函数。async def read_item(skip: int = 0, limit: int = 10):
: 路径操作函数接收两个查询参数:skip
和limit
。skip: int = 0
和limit: int = 10
表示skip
和limit
必须是整数类型,并且默认值分别为 0 和 10。
访问 http://127.0.0.1:8000/items/?skip=20&limit=50
将返回:
json
{
"skip": 20,
"limit": 50
}
五、请求体与数据验证
FastAPI 使用 Pydantic 进行请求体的数据验证。 你需要定义一个 Pydantic 模型来描述请求体的数据结构。
“`python
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None
app = FastAPI()
@app.post(“/items/”)
async def create_item(item: Item):
item_dict = item.model_dump()
if item.tax:
price_with_tax = item.price + item.tax
item_dict.update({“price_with_tax”: price_with_tax})
return item_dict
“`
代码解释:
from pydantic import BaseModel
: 导入 Pydantic 的BaseModel
类。class Item(BaseModel):
: 定义一个名为Item
的 Pydantic 模型。name
,description
,price
和tax
是模型的属性。name: str
:name
属性必须是字符串类型。description: str | None = None
:description
属性可以是字符串类型或None
,并且默认值为None
。str | None
是 Python 3.10 引入的联合类型注解,等价于typing.Optional[str]
。price: float
:price
属性必须是浮点数类型。tax: float | None = None
:tax
属性可以是浮点数类型或None
,并且默认值为None
。@app.post("/items/")
: 定义一个处理对/items/
的 POST 请求的路径操作函数。async def create_item(item: Item):
: 路径操作函数接收一个名为item
的参数, 其类型为Item
。 FastAPI 会自动将请求体数据转换为Item
模型的实例,并进行数据验证。item_dict = item.model_dump()
: 将 Pydantic 模型转换为字典。if item.tax:
: 如果tax
属性存在,则计算含税价格。item_dict.update({"price_with_tax": price_with_tax})
: 将含税价格添加到字典中。return item_dict
: 返回一个 JSON 响应。
要发送 POST 请求,可以使用 curl
或类似工具:
bash
curl -X POST -H "Content-Type: application/json" -d '{"name": "Foo", "description": "A very nice Item", "price": 50.2, "tax": 5.2}' http://127.0.0.1:8000/items/
这将返回:
json
{
"name": "Foo",
"description": "A very nice Item",
"price": 50.2,
"tax": 5.2,
"price_with_tax": 55.4
}
如果请求体数据不符合 Item
模型的定义,FastAPI 将返回一个错误响应,指示验证失败的原因。
六、依赖注入
FastAPI 具有强大的依赖注入系统,允许你将依赖项注入到路径操作函数中。 这有助于提高代码的可重用性和可测试性。
“`python
from fastapi import FastAPI, Depends
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 10):
return {“q”: q, “skip”: skip, “limit”: limit}
app = FastAPI()
@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
“`
代码解释:
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 10):
: 定义一个名为common_parameters
的依赖项函数。 这个函数接收三个查询参数:q
,skip
和limit
,并返回一个包含这些参数的字典。Depends(common_parameters)
: 使用Depends
函数将common_parameters
函数注入到路径操作函数中。async def read_items(commons: dict = Depends(common_parameters)):
: 路径操作函数接收一个名为commons
的参数, 其类型为dict
。commons
将包含common_parameters
函数返回的字典。@app.get("/users/")
: 另一个使用相同依赖项的路径操作函数。
访问 http://127.0.0.1:8000/items/?skip=50&limit=100
将返回:
json
{
"q": null,
"skip": 50,
"limit": 100
}
七、安全与认证
FastAPI 提供了多种安全和认证机制,包括:
- HTTP Basic Authentication: 最简单的认证方式, 通过用户名和密码进行认证。
- OAuth 2.0 with JWT: 基于 OAuth 2.0 协议和 JSON Web Tokens (JWT) 的安全认证方式。 提供更强的安全性,并支持授权。
- API Keys: 使用 API 密钥进行身份验证。
以下是一个使用 API 密钥进行身份验证的示例:
“`python
from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import APIKeyHeader
from starlette import status
API_KEY = “YOUR_SECRET_API_KEY” # Replace with a secure key management solution
API_KEY_NAME = “X-API-Key”
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
async def get_api_key(api_key_header: str = Security(api_key_header)):
if api_key_header == API_KEY:
return api_key_header
else:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, detail=”Could not validate API Key”
)
app = FastAPI()
@app.get(“/items/”, dependencies=[Depends(get_api_key)])
async def read_items():
return [{“name”: “Item 1”}, {“name”: “Item 2”}]
“`
代码解释:
API_KEY
: 存储 API 密钥,请勿将密钥直接存储在代码中,使用环境变量或更安全的密钥管理方案。API_KEY_NAME
: 定义 API 密钥 HTTP 头部的名称。APIKeyHeader
: 定义一个 API 密钥头部验证器。get_api_key
: 定义一个依赖项函数,用于验证 API 密钥。Security(api_key_header)
: 使用Security
函数将api_key_header
验证器注入到get_api_key
函数中。dependencies=[Depends(get_api_key)]
: 将get_api_key
函数作为依赖项添加到/items/
路径操作函数中。 这意味着只有提供了有效的 API 密钥,才能访问/items/
路径。
要访问 /items/
路径,需要在 HTTP 请求头部中包含 X-API-Key
,其值为 YOUR_SECRET_API_KEY
。
八、测试
FastAPI 提供了强大的测试支持,可以方便地测试你的 API。 可以使用 pytest 等测试框架来编写测试用例。
“`python
from fastapi.testclient import TestClient
from .main import app # 假设你的 FastAPI 应用在 main.py 文件中
client = TestClient(app)
def test_read_main():
response = client.get(“/”)
assert response.status_code == 200
assert response.json() == {“Hello”: “World”}
def test_read_item():
response = client.get(“/items/5”)
assert response.status_code == 200
assert response.json() == {“item_id”: 5}
“`
代码解释:
from fastapi.testclient import TestClient
: 导入TestClient
类。client = TestClient(app)
: 创建一个TestClient
实例,并将其连接到你的 FastAPI 应用程序。response = client.get("/")
: 发送一个 GET 请求到根路径。assert response.status_code == 200
: 断言响应状态码为 200。assert response.json() == {"Hello": "World"}
: 断言响应 JSON 数据与预期值匹配。
九、总结
FastAPI 是一个功能强大且易于使用的 Web 框架,用于构建高性能 API。 它提供了丰富的功能,包括数据验证、自动 API 文档生成、依赖注入和安全认证。 本教程涵盖了 FastAPI 的核心概念和特性,并提供了大量的示例代码,帮助你快速入门 FastAPI 开发。 通过学习本教程,你将能够使用 FastAPI 构建出高效、安全且易于维护的 API。 请务必参考官方文档以获得更详细的信息和高级用法。 祝你编码愉快!