发布于 由admin

Python的`RESTful API`设计：如何使用`FastAPI`和`Pydantic`构建健壮、类型安全的API。

使用 FastAPI 和 Pydantic 构建健壮、类型安全的 RESTful API

大家好！今天我们来探讨如何使用 FastAPI 和 Pydantic 构建健壮、类型安全的 RESTful API。FastAPI 是一个现代、高性能的 Python Web 框架，用于构建 API。它基于标准 Python 类型提示，并提供了自动的数据验证、序列化和 API 文档生成等功能。Pydantic 则是一个数据验证和设置管理库，它使用 Python 类型提示来定义数据模型，并提供强大的验证和转换功能，与 FastAPI 配合使用，可以极大地提高 API 的开发效率和可靠性。

1. 搭建开发环境

首先，我们需要安装 FastAPI 和 Pydantic。可以使用 pip 来安装：

pip install fastapi uvicorn pydantic
  • fastapi: FastAPI 框架本身。
  • uvicorn: 一个 ASGI (Asynchronous Server Gateway Interface) 服务器，用于运行 FastAPI 应用。
  • pydantic: 数据验证和设置管理库。

安装完成后，我们还需要一个 ASGI 服务器来运行 FastAPI 应用。Uvicorn 是一个常用的选择。

2. 创建第一个 FastAPI 应用

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

from fastapi import FastAPI

app = FastAPI()

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

这段代码创建了一个 FastAPI 应用实例 app，并定义了一个根路由 / 的 GET 请求处理函数 read_root。该函数返回一个 JSON 响应 {"message": "Hello World"}。

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

uvicorn main:app --reload
  • main: 指定 FastAPI 应用所在的文件 (main.py)。
  • app: 指定 FastAPI 应用实例的名称 (app)。
  • --reload: 启用自动重载，当代码发生更改时，服务器会自动重启。

在浏览器中访问 http://localhost:8000，你将会看到 {"message": "Hello World"}。

3. 使用 Pydantic 定义数据模型

Pydantic 的核心是数据模型。我们可以使用 Pydantic 定义数据模型，并使用类型提示来指定字段的类型。例如，我们可以定义一个 Item 模型：

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None  # 可以为 None
    price: float
    tax: float | None = None

在这个例子中，Item 模型有四个字段：

  • name: 字符串类型，必填。
  • description: 字符串类型，可选（默认为 None）。
  • price: 浮点数类型，必填。
  • tax: 浮点数类型，可选（默认为 None）。

str | None 表示该字段可以是字符串类型或 None。这相当于 Optional[str]，但在 Python 3.10+ 中更简洁。

Pydantic 会自动验证输入的数据是否符合模型定义。如果数据不符合要求，Pydantic 会抛出 ValidationError 异常。

4. 在 API 中使用 Pydantic 模型

现在，我们可以将 Item 模型应用到 API 中。修改 main.py 文件：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

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

items = {}  # 模拟数据库

@app.post("/items/{item_id}")
async def create_item(item_id: int, item: Item):
    if item_id in items:
        raise HTTPException(status_code=400, detail="Item already exists")
    items[item_id] = item
    return {"item_id": item_id, **item.dict()}

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

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    items[item_id] = item
    return {"item_id": item_id, **item.dict()}

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    del items[item_id]
    return {"message": f"Item {item_id} deleted"}

在这个例子中，我们定义了四个 API 接口：

  • POST /items/{item_id}: 创建一个新的 Item。item_id 是路径参数，item 是请求体，使用 Item 模型进行验证。如果 item_id 已经存在，则返回 HTTP 400 错误。
  • GET /items/{item_id}: 读取一个 Item。item_id 是路径参数。如果 item_id 不存在，则返回 HTTP 404 错误。
  • PUT /items/{item_id}: 更新一个 Item。item_id 是路径参数，item 是请求体，使用 Item 模型进行验证。如果 item_id 不存在，则返回 HTTP 404 错误。
  • DELETE /items/{item_id}: 删除一个 Item。item_id 是路径参数。如果 item_id 不存在，则返回 HTTP 404 错误。

item.dict() 方法将 Pydantic 模型转换为 Python 字典。**item.dict() 使用字典解包将字典的键值对添加到返回的 JSON 响应中。

可以使用 curl 或 Postman 等工具来测试这些 API 接口。例如，使用 curl 创建一个 Item：

curl -X POST -H "Content-Type: application/json" -d '{"name": "Foo", "description": "A very nice Item", "price": 50.2, "tax": 3.2}' http://localhost:8000/items/1

5. 参数验证和类型转换

FastAPI 和 Pydantic 提供了强大的参数验证和类型转换功能。我们可以使用类型提示来指定参数的类型，FastAPI 会自动验证参数是否符合要求，并将其转换为正确的类型。

例如，我们可以定义一个查询参数 q：

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items(q: str | None = None):
    if q:
        return {"items": [{"name": "Foo"}, {"name": "Bar"}], "q": q}
    return {"items": [{"name": "Foo"}, {"name": "Bar"}]}

在这个例子中，q 是一个可选的查询参数，类型为字符串。如果客户端提供了 q 参数，FastAPI 会将其传递给 read_items 函数。如果没有提供 q 参数，q 的值为 None。

FastAPI 支持各种类型的参数，包括：

  • 路径参数: 使用花括号 {} 定义，例如 /items/{item_id}。
  • 查询参数: 使用 ? 和 & 定义，例如 /items/?q=search_term&limit=10。
  • 请求体: 使用 Pydantic 模型定义，例如 Item 模型。
  • Header 参数: 使用 Header 类定义。
  • Cookie 参数: 使用 Cookie 类定义。

6. 依赖注入

FastAPI 具有强大的依赖注入系统。依赖注入允许你将一些通用的逻辑提取出来，并将其注入到不同的 API 接口中。

例如，我们可以定义一个依赖函数 get_db，用于获取数据库连接：

from fastapi import Depends, FastAPI
from typing import Annotated

app = FastAPI()

def get_db():
    db = "模拟数据库连接" # 替换为真实的数据库连接逻辑
    try:
        yield db
    finally:
        # 在请求完成后关闭数据库连接
        pass  # 替换为真实的数据库连接关闭逻辑

@app.get("/items/")
async def read_items(db: Annotated[str, Depends(get_db)]):
    # 在这里使用数据库连接 db
    return {"db_dependency": "数据库依赖注入成功"}

在这个例子中，get_db 函数使用 yield 关键字返回数据库连接。Depends(get_db) 表示 read_items 函数依赖于 get_db 函数。FastAPI 会自动调用 get_db 函数，并将返回的数据库连接传递给 read_items 函数。

使用 yield 关键字可以确保在请求完成后关闭数据库连接。try...finally 语句块可以确保即使在请求过程中发生异常，数据库连接也会被关闭。

Annotated[str, Depends(get_db)] 注解告诉 FastAPI，db 参数的类型是 str，并且它的值来自于 get_db 函数。 Annotated 是 Python 3.9 引入的类型提示功能，用于添加元数据到类型提示。

7. 异常处理

FastAPI 提供了灵活的异常处理机制。我们可以使用 HTTPException 类来返回 HTTP 错误。

例如，在之前的例子中，我们使用 HTTPException 类来返回 HTTP 404 错误：

from fastapi import FastAPI, HTTPException

app = FastAPI()

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

我们还可以自定义异常处理函数。例如，我们可以定义一个异常处理函数来处理所有未捕获的异常：

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"detail": exc.errors(), "body": exc.body},
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.detail},
    )

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id > 100:
        raise StarletteHTTPException(status_code=400, detail="Item ID too large")
    return {"item_id": item_id}

@app.get("/test_validation")
async def test_validation(value: int):
    return {"value": value}

在这个例子中，@app.exception_handler(RequestValidationError) 装饰器将 validation_exception_handler 函数注册为 RequestValidationError 异常的处理函数。@app.exception_handler(StarletteHTTPException) 装饰器将 http_exception_handler 函数注册为 StarletteHTTPException 异常的处理函数。

当 API 接口抛出 RequestValidationError 异常时，validation_exception_handler 函数会被调用，并将异常信息格式化为 JSON 响应返回。当 API 接口抛出 StarletteHTTPException 异常时，http_exception_handler 函数会被调用，并将异常信息格式化为 JSON 响应返回。

8. 中间件

中间件是位于请求和响应之间的组件。我们可以使用中间件来执行一些通用的操作，例如身份验证、日志记录和请求转换。

例如，我们可以定义一个中间件来记录每个请求的日志：

from fastapi import FastAPI, Request
import time

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)
    print(f"Request processed in {process_time:.4f} seconds")
    return response

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

在这个例子中，@app.middleware("http") 装饰器将 add_process_time_header 函数注册为 HTTP 中间件。add_process_time_header 函数接收两个参数：request 和 call_next。request 是请求对象，call_next 是一个可调用对象，用于调用下一个中间件或 API 接口。

在 add_process_time_header 函数中，我们首先记录请求的开始时间，然后调用 call_next(request) 来处理请求。call_next(request) 会返回一个响应对象。然后，我们计算请求的处理时间，并将处理时间添加到响应头中。最后，我们返回响应对象。

9. API 文档

FastAPI 会自动生成 API 文档。我们可以通过访问 /docs 路由来查看 API 文档。FastAPI 使用 Swagger UI 来展示 API 文档。

FastAPI 还可以生成 OpenAPI 规范。我们可以通过访问 /openapi.json 路由来获取 OpenAPI 规范。

API 文档和 OpenAPI 规范可以帮助开发者更好地了解和使用 API。

10. 总结

FastAPI 和 Pydantic 是构建健壮、类型安全的 RESTful API 的强大工具。它们提供了自动的数据验证、序列化、API 文档生成和依赖注入等功能，可以极大地提高 API 的开发效率和可靠性。

通过合理利用 Pydantic 的数据模型定义、FastAPI 的类型提示和依赖注入系统，可以构建出易于维护、可扩展的 API。 掌握异常处理和中间件的使用，则能进一步提升 API 的健壮性和安全性。

发表回复

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