发布于 由admin

Python的`RESTful API`：如何使用`FastAPI`构建可扩展的`RESTful API`。

FastAPI 构建可扩展 RESTful API：一场由浅入深的探索

大家好！今天我们来聊聊如何使用 FastAPI 构建可扩展的 RESTful API。FastAPI 凭借其高性能、易用性以及自动化的数据验证和 API 文档生成，已经成为 Python Web 开发领域的一颗耀眼新星。本次讲座将由浅入深，从基础概念入手，逐步构建一个具有实际意义的 API，并探讨如何使其具备良好的可扩展性。

一、RESTful API 的基石：概念回顾与原则

在深入 FastAPI 之前，我们先来回顾一下 RESTful API 的核心概念和设计原则。REST (Representational State Transfer) 是一种软件架构风格，用于构建分布式系统，尤其是 Web 服务。一个 RESTful API 应该遵循以下关键原则：

  • 客户端-服务器架构： 客户端和服务器之间职责分离，客户端负责用户界面和交互，服务器负责数据存储和处理。
  • 无状态性： 服务器不应存储任何客户端状态信息。每次请求都应包含足够的信息，以便服务器能够理解和处理。
  • 可缓存性： 响应可以被客户端或中间服务器缓存，以提高性能。
  • 分层系统： 客户端不必知道服务器背后的中间层，从而简化了系统架构。
  • 按需代码： (可选) 服务器可以向客户端提供可执行代码，例如 JavaScript，以扩展客户端功能。
  • 统一接口： 这是 REST 的核心原则，它定义了客户端和服务器之间交互的方式。统一接口包括：
    • 资源识别： 使用 URI (Uniform Resource Identifier) 来唯一标识每个资源。
    • 资源的表述： 使用标准数据格式（如 JSON 或 XML）来表示资源的状态。
    • 自描述消息： 消息本身包含足够的信息，以便接收者能够理解。
    • 超媒体作为应用状态引擎 (HATEOAS)： 服务器响应包含指向其他相关资源的链接，从而引导客户端浏览 API。

常用的 HTTP 方法和含义如下表所示：

HTTP 方法 含义 常用场景
GET 从服务器获取资源。不会修改服务器上的任何数据。 获取用户信息、读取文章内容、查询商品列表等。
POST 向服务器提交数据，通常用于创建新资源。 创建用户账户、发布新文章、添加购物车商品等。
PUT 替换服务器上现有资源的所有内容。如果资源不存在，则可能创建新资源（但通常不建议这样做，因为 PUT 的语义是替换）。 更新用户信息（例如，更新用户的姓名、地址等，需要提供用户的所有信息）。
PATCH 修改服务器上现有资源的局部内容。只更新资源的部分字段。 更新用户信息（例如，只更新用户的邮箱地址）。
DELETE 从服务器删除资源。 删除用户账户、删除文章、删除购物车商品等。

二、FastAPI 入门：快速构建你的第一个 API

现在，让我们开始使用 FastAPI 构建一个简单的 API。首先，你需要安装 FastAPI 和 ASGI 服务器（例如 Uvicorn）：

pip install fastapi uvicorn

接下来，创建一个名为 main.py 的文件，并添加以下代码：

from fastapi import FastAPI

app = FastAPI()

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

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

这段代码定义了一个 FastAPI 应用，并创建了两个 API 接口：

  • /: 返回一个简单的 "Hello World" 消息。
  • /items/{item_id}: 接受一个整数类型的 item_id 作为路径参数，以及一个可选的字符串类型的查询参数 q。

要运行这个 API，可以使用 Uvicorn：

uvicorn main:app --reload

这会启动一个本地服务器，默认监听 http://127.0.0.1:8000。你可以通过浏览器或 curl 命令访问这些 API 接口：

  • http://127.0.0.1:8000/: 显示 {"message": "Hello World"}
  • http://127.0.0.1:8000/items/123?q=test: 显示 {"item_id": 123, "q": "test"}

三、数据验证与序列化：Pydantic 的强大之处

FastAPI 集成了 Pydantic，一个强大的数据验证和序列化库。使用 Pydantic 可以方便地定义数据模型，并自动验证请求和响应的数据格式。

修改 main.py 文件，添加以下代码：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: 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

这段代码定义了一个 Item 类，它继承自 BaseModel。这个类定义了 name (字符串类型，必填)、description (字符串类型，可选)、price (浮点数类型，必填) 和 tax (浮点数类型，可选) 四个字段。

@app.post("/items/") 装饰器定义了一个 POST 请求的 API 接口，用于创建新的 Item。 FastAPI 会自动验证请求体中的数据是否符合 Item 类的定义。如果数据无效，FastAPI 会返回一个包含错误信息的 HTTP 422 响应。

你可以使用 curl 命令来测试这个 API：

curl -X POST -H "Content-Type: application/json" -d '{"name": "My Item", "price": 10.0}' http://127.0.0.1:8000/items/

这个命令会向 /items/ 发送一个 POST 请求，请求体包含一个 JSON 对象，其中 name 和 price 字段的值分别是 "My Item" 和 10.0。服务器会返回一个包含 item_dict 信息的 JSON 响应，其中包含了计算得到的price_with_tax。

四、依赖注入：解耦与可测试性

FastAPI 提供了强大的依赖注入系统，可以帮助你解耦代码，提高可测试性。依赖注入允许你将依赖项传递给函数，而不是在函数内部创建它们。

修改 main.py 文件，添加以下代码：

from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()

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

def get_db():
    # 模拟数据库连接
    db = {"items": []}
    try:
        yield db
    finally:
        # 关闭数据库连接 (如果需要)
        pass

@app.post("/items/")
async def create_item(item: Item, db: dict = Depends(get_db)):
    item_dict = item.dict()
    db["items"].append(item_dict)
    return item_dict

这段代码定义了一个 get_db 函数，它模拟了一个数据库连接。Depends(get_db) 告诉 FastAPI 在调用 create_item 函数时，先调用 get_db 函数，并将返回值作为 db 参数传递给 create_item 函数。

这样做的优点是：

  • 解耦： create_item 函数不再直接依赖于具体的数据库实现。
  • 可测试性： 在测试时，你可以轻松地使用 mock 对象来替换真实的数据库连接。
  • 可维护性： 如果你需要更改数据库实现，只需要修改 get_db 函数即可，而不需要修改 create_item 函数。

五、API 版本控制：平滑升级与兼容性

随着 API 的不断发展，可能需要引入新的功能或修改现有功能。为了避免影响现有的客户端，通常需要进行 API 版本控制。

FastAPI 提供了多种 API 版本控制的方法，例如：

  • URL 前缀： 例如 /v1/items/ 和 /v2/items/。
  • 请求头： 例如 Accept: application/vnd.example.v1+json。
  • 查询参数： 例如 /items/?version=1。

这里我们使用 URL 前缀的方式来实现 API 版本控制。

修改 main.py 文件，添加以下代码：

from fastapi import FastAPI, APIRouter
from pydantic import BaseModel

app = FastAPI()

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

router_v1 = APIRouter(prefix="/v1")
router_v2 = APIRouter(prefix="/v2")

@router_v1.post("/items/")
async def create_item_v1(item: Item):
    return {"version": 1, "item": item}

@router_v2.post("/items/")
async def create_item_v2(item: Item):
    return {"version": 2, "item": item, "description_length": len(item.description) if item.description else 0}

app.include_router(router_v1)
app.include_router(router_v2)

这段代码创建了两个 APIRouter 对象，分别对应 API 的 v1 和 v2 版本。每个 APIRouter 对象都定义了自己的 API 接口。然后，使用 app.include_router() 方法将这两个 APIRouter 对象添加到 FastAPI 应用中。

现在，你可以通过 /v1/items/ 和 /v2/items/ 访问不同版本的 API 接口。

六、身份验证与授权：保障 API 安全

身份验证和授权是 API 安全的重要组成部分。身份验证用于验证用户的身份，授权用于确定用户是否有权访问特定的资源。

FastAPI 提供了多种身份验证和授权的方法，例如：

  • HTTP Basic Authentication： 简单易用，但不安全。
  • API Key Authentication： 使用 API 密钥来验证用户身份。
  • OAuth 2.0： 一种授权框架，允许第三方应用安全地访问用户资源。
  • JWT (JSON Web Token)： 一种轻量级的身份验证协议，可以用于在客户端和服务器之间安全地传输用户信息。

这里我们使用 JWT 来实现身份验证和授权。首先，你需要安装 python-jose 和 passlib：

pip install python-jose passlib bcrypt

然后，修改 main.py 文件，添加以下代码：

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext

app = FastAPI()

# 安全配置
SECRET_KEY = "YOUR_SECRET_KEY"  # 实际项目中应该使用更安全的随机字符串
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 密码哈希
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None

class User(BaseModel):
    username: str
    password: str

# 模拟用户数据库
users = {
    "testuser": {"password": pwd_context.hash("testpassword")}
}

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = users.get(token_data.username)
    if user is None:
        raise credentials_exception
    return user

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = users.get(form_data.username)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    if not verify_password(form_data.password, user["password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": form_data.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return {"username": current_user.username}

@app.get("/items/")
async def read_items(current_user: User = Depends(get_current_user)):
    return [{"item": "item1"}, {"item": "item2"}]

这段代码实现了以下功能：

  • 生成密码哈希： 使用 passlib 库来生成密码哈希，以保护用户密码的安全。
  • 创建 JWT： 使用 python-jose 库来创建 JWT，其中包含用户信息和过期时间。
  • 验证 JWT： 使用 python-jose 库来验证 JWT 的有效性。
  • OAuth2 流程： 使用 fastapi.security 模块提供的 OAuth2PasswordBearer 和 OAuth2PasswordRequestForm 类来实现 OAuth2 流程。
  • 获取当前用户： 使用 Depends 函数来获取当前用户的信息。

现在，你可以使用 /token 接口来获取 access token，然后使用 access token 来访问需要身份验证的 API 接口，例如 /users/me 和 /items/。需要在请求头中添加 Authorization: Bearer <access_token>。

七、异步编程：提升 API 性能

FastAPI 基于 ASGI (Asynchronous Server Gateway Interface)，可以充分利用 Python 的异步编程能力。异步编程可以显著提高 API 的性能，尤其是在处理 I/O 密集型任务时。

在前面的例子中，我们已经使用了 async 和 await 关键字来定义异步函数。但是，为了充分利用异步编程的优势，还需要注意以下几点：

  • 使用异步库： 尽量使用异步版本的库，例如 asyncpg (PostgreSQL 异步驱动) 和 aiohttp (异步 HTTP 客户端)。
  • 避免阻塞操作： 避免在异步函数中执行阻塞操作，例如文件 I/O 或网络 I/O。如果必须执行阻塞操作，可以使用 asyncio.to_thread 函数将其放入一个独立的线程中执行。

八、API 文档：自动化生成与维护

FastAPI 自动生成 API 文档，并支持 Swagger UI 和 ReDoc。API 文档可以帮助开发者了解 API 的功能和使用方法。

FastAPI 会根据你定义的 API 接口和数据模型，自动生成 OpenAPI (以前称为 Swagger) 规范。然后，可以使用 Swagger UI 或 ReDoc 来渲染 API 文档。

要访问 API 文档，可以在浏览器中输入以下地址：

  • http://127.0.0.1:8000/docs: Swagger UI
  • http://127.0.0.1:8000/redoc: ReDoc

九、异常处理：优雅地应对错误

良好的异常处理机制可以帮助你优雅地应对错误，并向客户端返回有用的错误信息。

FastAPI 提供了多种异常处理的方法，例如：

  • 使用 try...except 语句： 捕获特定的异常，并进行处理。
  • 使用 HTTPException： 抛出 HTTP 异常，并向客户端返回相应的 HTTP 状态码和错误信息。
  • 自定义异常处理函数： 定义自定义的异常处理函数，用于处理特定的异常。

以下是一个使用 HTTPException 的例子：

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in [1, 2, 3]:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}

如果 item_id 不在允许的列表中，则会抛出一个 HTTPException，并向客户端返回一个 HTTP 404 错误。

十、日志记录：追踪 API 行为

日志记录是追踪 API 行为的重要手段。通过记录 API 的请求和响应信息，可以帮助你诊断问题、监控性能和进行安全审计。

Python 提供了 logging 模块，可以用于进行日志记录。FastAPI 可以与 logging 模块集成，方便地记录 API 的行为。

十一、代码组织和模块化：提升可维护性

随着 API 功能的增加，代码量也会不断增加。为了提高代码的可维护性，需要对代码进行组织和模块化。

可以将 API 接口、数据模型、业务逻辑等分别放入不同的模块中。然后，使用 import 语句将这些模块导入到主文件中。

十二、测试：保证 API 质量

测试是保证 API 质量的重要手段。通过编写单元测试和集成测试，可以确保 API 的功能正确、性能良好和安全可靠。

可以使用 pytest 框架来编写测试。pytest 框架提供了丰富的功能，例如自动发现测试用例、断言、mock 对象等。

十三、部署：让 API 走向世界

最后，我们需要将 API 部署到生产环境中，让用户可以访问。

FastAPI 应用可以部署到多种环境中，例如：

  • 云服务器： 例如 AWS EC2、Google Compute Engine、Azure Virtual Machines。
  • 容器： 例如 Docker、Kubernetes。
  • Serverless： 例如 AWS Lambda、Google Cloud Functions、Azure Functions。

选择哪种部署环境取决于你的具体需求和预算。

快速回顾要点：

本次分享涵盖了 RESTful API 的基本概念和设计原则，并展示了如何使用 FastAPI 构建可扩展的 API。我们学习了数据验证、依赖注入、API 版本控制、身份验证与授权、异步编程、API 文档生成、异常处理、日志记录、代码组织和模块化、测试和部署等关键技术。掌握这些技术，你就可以构建出高质量、可扩展的 RESTful API。

发表回复

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