使用 FastAPI 创建 RESTful API:设计 RESTful 端点

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于基于标准 Python 类型提示使用 Python 3.6+ 构建 API。在本博客中，我们将探讨如何使用 FastAPI 创建 RESTful 端点，重点介绍此框架的简单性和强大功能。在本文结束时，您将清楚地了解如何使用 FastAPI 设计和实现 RESTful API。

RESTful API 简介

REST（Representational State Transfer的简写）是一种用于设计网络应用程序的架构风格。RESTful API 遵循六个约束：

统一接口：与资源交互的一致方式。

无状态：来自客户端的每个请求都包含理解和处理请求所需的所有信息。

可缓存：响应必须将自身定义为可缓存或不可缓存，以防止客户端重用陈旧数据。

客户端-服务器：客户端和服务器之间的关注点分离。

分层系统：客户端通常无法分辨它是直接连接到终端服务器还是中介。

按需代码（可选）：服务器可以通过传输可执行代码来扩展客户端功能。

FastAPI 可以轻松构建符合这些原则的 API。

设置 FastAPI

首先，让我们安装 FastAPI 和 Uvicorn（ASGI 服务器）：

pip install fastapi uvicorn

接下来，创建一个新的 Python 文件 main.py，我们将在其中定义我们的 API。

设计 RESTful 端点

创建应用程序

首先导入必要的模块并创建 FastAPI 类的实例：

from fastapi import FastAPI


app = FastAPI()

定义模型

我们将使用 Pydantic 定义一个简单的模型，FastAPI 使用它进行数据验证和序列化：

from pydantic import BaseModel


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

CRUD 端点

创建（Create）项目

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    return item

读取（Read）项目

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    return {"name": "Sample Item", "description": "This is a sample item", "price": 10.0, "tax": 1.0}

更新（Update）项目

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

删除（Delete）项目

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"message": "Item deleted successfully"}

运行应用程序

要运行 FastAPI 应用程序，请使用 Uvicorn：

uvicorn main:app --reload

此命令将在 http://127.0.0.1:8000 上启动服务器并启用代码更改的自动重新加载。

完整示例

以下是包含所有 CRUD 操作的完整 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/", response_model=Item)
async def create_item(item: Item):
    return item


@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    return {"name": "Sample Item", "description": "This is a sample item", "price": 10.0, "tax": 1.0}


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


@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"message": "Item deleted successfully"}

以下是使用 FastAPI 创建 RESTful API 的一些其他演示，重点介绍更高级的功能，例如查询参数、路径参数、请求主体和响应模型。

使用 FastAPI 创建 RESTful API 的其他演示

查询参数

查询参数用于过滤请求的结果。让我们创建一个根据可选查询参数检索项目的端点。

from typing import List, Optional
from fastapi import FastAPI


app = FastAPI()


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


items = [
    {"name": "Item 1", "description": "Description 1", "price": 10.0, "tax": 2.0},
    {"name": "Item 2", "description": "Description 2", "price": 20.0, "tax": 3.0},
    {"name": "Item 3", "description": "Description 3", "price": 30.0, "tax": 5.0},
]


@app.get("/items/", response_model=List[Item])
async def read_items(skip: int = 0, limit: int = 10):
    return items[skip: skip + limit]

路径参数

路径参数用于识别特定资源。让我们创建一个通过其 ID 检索项目的端点。

from fastapi import HTTPException


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

请求主体

请求主体用于将数据发送到服务器。让我们创建一个使用请求主体更新项目的端点。

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

响应模型

响应模型用于塑造输出数据。让我们创建一个返回自定义响应模型的端点。

from pydantic import BaseModel


class ItemResponse(BaseModel):
    item: Item
    message: str


@app.post("/items/", response_model=ItemResponse)
async def create_item(item: Item):
    items.append(item.dict())
    return {"item": item, "message": "Item created successfully"}

依赖注入

FastAPI 支持依赖注入，以实现可重用性和关注点分离。让我们创建一个检索数据库连接的简单依赖项。

from typing import Generator
from fastapi import Depends


# Mock database connection
def get_db() -> Generator:
    db = {"items": items}
    try:
        yield db
    finally:
        pass


@app.get("/items/", response_model=List[Item])
async def read_items(skip: int = 0, limit: int = 10, db: dict = Depends(get_db)):
    return db["items"][skip: skip + limit]

身份验证

FastAPI 提供了用于实现身份验证的工具。以下是使用 API 密钥进行身份验证的基本示例。

from fastapi.security.api_key import APIKeyHeader, APIKey


API_KEY_NAME = "access_token"
API_KEY = "secret-key"
api_key_header = APIKeyHeader(name=API_KEY_NAME)


@app.get("/secure-items/", response_model=List[Item])
async def read_secure_items(api_key: str = Depends(api_key_header)):
    if api_key != API_KEY:
        raise HTTPException(status_code=403, detail="Could not validate credentials")
    return items

自定义中间件

您可以添加自定义中间件来处理请求和响应。让我们创建一个记录请求方法和 URL 的中间件。

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
import logging


logging.basicConfig(level=logging.INFO)


class LogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        logging.info(f"Request method: {request.method}, URL: {request.url}")
        response = await call_next(request)
        return response


app.add_middleware(LogMiddleware)


@app.get("/items/", response_model=List[Item])
async def read_items():
    return items

这些额外的演示展示了 FastAPI 在构建 RESTful API 方面的多功能性和强大功能。从处理查询和路径参数到实现身份验证和自定义中间件，FastAPI 为开发现代 Web API 提供了一个强大而高效的框架。

在本博客中，我们介绍了使用 FastAPI 创建 RESTful API 的基础知识。我们定义了模型，创建了 CRUD 端点，并探索了如何运行应用程序以及一些高阶用法。FastAPI 的直观设计和自动文档生成使其成为快速高效构建 API 的绝佳选择。祝您编码愉快！