FastAPI 完整教程：学习构建 RESTful API – wiki基地

作者： /

FastAPI 完整教程：学习构建 RESTful API

FastAPI 是一个现代、高性能的 Python Web 框架，用于构建 API。它易于学习、速度快、并且具备开箱即用的类型提示和数据验证。 本教程将带您逐步了解 FastAPI 的核心概念和最佳实践，帮助您构建健壮、高效的 RESTful API。

为什么选择 FastAPI？

在众多 Python Web 框架中，FastAPI 凭借其独特的优势脱颖而出：

  • 高性能: FastAPI 基于 Starlette 和 Pydantic 构建，具有极高的性能，可媲美 Node.js 和 Go。
  • 快速开发: 简洁的语法和强大的类型提示功能，显著缩短了开发周期。
  • 易于学习和使用: 学习曲线平缓，即使是初学者也能快速上手。
  • 自动化的 API 文档: 内置 Swagger UI 和 ReDoc，自动生成美观、交互式的 API 文档。
  • 数据验证: 基于 Pydantic 进行数据验证，确保数据的有效性和一致性。
  • 依赖注入: 强大的依赖注入系统，简化了代码结构和测试。
  • 类型提示: 充分利用 Python 的类型提示，提高代码的可读性和可维护性。
  • 异步支持: 内置对异步编程的支持，可以轻松处理高并发请求。

先决条件

  • Python 3.7+
  • pip (Python 包管理器)

安装 FastAPI

使用 pip 安装 FastAPI：

bash
pip install fastapi uvicorn

  • fastapi: FastAPI 框架本身。
  • uvicorn: 一个 ASGI 服务器，用于运行 FastAPI 应用。

Hello World 示例

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

“`python
from fastapi import FastAPI

app = FastAPI()

@app.get(“/”)
async def root():
return {“message”: “Hello World”}
“`

代码解释:

  • from fastapi import FastAPI: 导入 FastAPI 类。
  • app = FastAPI(): 创建一个 FastAPI 应用实例。
  • @app.get("/"): 定义一个 GET 请求的路由，路径为根路径 /。这是一个装饰器，将 root 函数与该路由关联起来。
  • async def root(): 定义一个异步函数 root，该函数处理请求。 async 关键字表示这是一个协程函数，可以并发执行。
  • return {"message": "Hello World"}: 函数返回一个 JSON 格式的字典，FastAPI 会自动将其序列化为 JSON 响应。

运行应用

保存上面的代码到 main.py 文件，然后使用 uvicorn 命令运行应用：

bash
uvicorn main:app --reload

  • main: 指示要运行的 Python 文件。
  • app: 指示 FastAPI 应用实例的名称。
  • --reload: 启用自动重新加载，当代码发生更改时，服务器会自动重启。这对于开发非常方便。

现在，您可以通过浏览器访问 http://127.0.0.1:8000/ 来查看结果。 浏览器将显示：

json
{"message": "Hello World"}

访问 API 文档

FastAPI 会自动生成 API 文档，您可以通过以下 URL 访问：

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

这些文档包含了 API 的所有端点、参数、响应格式等信息，方便您进行 API 的测试和调试。

路由和 HTTP 方法

FastAPI 支持所有标准的 HTTP 方法，包括：

  • GET: 获取资源
  • POST: 创建新资源
  • PUT: 更新现有资源
  • DELETE: 删除资源
  • PATCH: 部分更新现有资源

可以使用相应的装饰器来定义不同 HTTP 方法的路由：

“`python
from fastapi import FastAPI

app = FastAPI()

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

@app.post(“/items/”)
async def create_item(item: dict):
return item

@app.put(“/items/{item_id}”)
async def update_item(item_id: int, item: dict):
return {“item_id”: item_id, “item”: item}

@app.delete(“/items/{item_id}”)
async def delete_item(item_id: int):
return {“message”: f”Item {item_id} deleted”}
“`

代码解释:

  • @app.get("/items/{item_id}"): 定义一个 GET 请求的路由，路径为 /items/{item_id}。 {item_id} 是一个路径参数，其值将被传递给 read_item 函数。
  • item_id: int: 使用类型提示来声明 item_id 的类型为整数。 FastAPI 会自动进行类型转换和验证。
  • q: str = None: 定义一个查询参数 q，类型为字符串，默认值为 None。 查询参数是 URL 中 ? 后面的参数，例如：/items/1?q=search_term.
  • @app.post("/items/"): 定义一个 POST 请求的路由，路径为 /items/。
  • item: dict: 声明 item 的类型为字典。 客户端需要在请求体中发送 JSON 数据，FastAPI 会自动将其解析为字典。
  • @app.put("/items/{item_id}"): 定义一个 PUT 请求的路由，路径为 /items/{item_id}。
  • @app.delete("/items/{item_id}"): 定义一个 DELETE 请求的路由，路径为 /items/{item_id}。

Pydantic 数据模型

Pydantic 是一个数据验证和设置管理库。 它允许您定义数据模型，并自动进行数据验证和转换。

“`python
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
“`

代码解释:

  • from pydantic import BaseModel: 导入 Pydantic 的 BaseModel 类。
  • class Item(BaseModel): 定义一个名为 Item 的数据模型，继承自 BaseModel。
  • name: str: 声明 name 字段的类型为字符串。
  • description: str = None: 声明 description 字段的类型为字符串，默认值为 None。
  • price: float: 声明 price 字段的类型为浮点数。
  • tax: float = None: 声明 tax 字段的类型为浮点数，默认值为 None。
  • item: Item: 在 create_item 函数中，声明 item 参数的类型为 Item 模型。 FastAPI 会自动将请求体中的 JSON 数据转换为 Item 模型的实例。
  • item.dict(): 将 Item 模型转换为字典。

当客户端发送的请求体不符合 Item 模型的定义时，FastAPI 会自动返回一个包含错误信息的响应。 这大大简化了数据验证的过程。

依赖注入

依赖注入是一种设计模式，允许您将依赖项传递给函数或类，而不是在函数或类内部创建依赖项。

FastAPI 内置了强大的依赖注入系统，可以帮助您编写更清晰、更可测试的代码。

“`python
from fastapi import FastAPI, Depends

app = FastAPI()

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

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

代码解释:

  • from fastapi import Depends: 导入 Depends 类。
  • async def common_parameters(q: str = None, skip: int = 0, limit: int = 100): 定义一个依赖函数 common_parameters，该函数接收查询参数 q、skip 和 limit。
  • commons: dict = Depends(common_parameters): 在 read_items 和 read_users 函数中，使用 Depends 来声明依赖项。 FastAPI 会自动调用 common_parameters 函数，并将返回值传递给 commons 参数。

通过使用依赖注入，我们可以避免在每个函数中重复编写相同的代码，并且可以更容易地测试和修改依赖项。

状态码和响应

您可以自定义 API 的状态码和响应内容。

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

app = FastAPI()

items = {“foo”: “The Foo Fighters”}

@app.get(“/items/{item_id}”)
async def read_item(item_id: str):
if item_id not in items:
raise HTTPException(status_code=404, detail=”Item not found”)
return {“item”: items[item_id]}

@app.get(“/custom-response”)
async def custom_response():
content = {“message”: “This is a custom response.”}
return JSONResponse(content=content, status_code=201)
“`

代码解释:

  • from fastapi import HTTPException: 导入 HTTPException 类，用于抛出 HTTP 异常。
  • raise HTTPException(status_code=404, detail="Item not found"): 如果 item_id 不存在于 items 字典中，则抛出一个 404 错误，并返回相应的错误信息。
  • from fastapi.responses import JSONResponse: 导入 JSONResponse 类，用于创建自定义的 JSON 响应。
  • return JSONResponse(content=content, status_code=201): 创建一个状态码为 201 的 JSON 响应，并返回给客户端。

中间件

中间件是在请求处理前后执行的函数。 它们可以用于执行各种任务，例如：

  • 日志记录
  • 身份验证
  • 授权
  • 请求和响应的修改

“`python
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

CORS (跨域资源共享) middleware

origins = [
“http://localhost”,
“http://localhost:8080”,
“http://example.com”,
]

app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=[““],
allow_headers=[““],
)

@app.middleware(“http”)
async def add_process_time_header(request: Request, call_next):
import time
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
“`

代码解释:

  • from fastapi.middleware.cors import CORSMiddleware: 导入 CORSMiddleware 类，用于处理 CORS 问题。
  • app.add_middleware(CORSMiddleware, ...): 将 CORSMiddleware 添加到应用中，并配置允许的来源、凭据、方法和头部。
  • @app.middleware("http"): 定义一个 HTTP 中间件。
  • async def add_process_time_header(request: Request, call_next): 定义中间件函数，该函数接收一个 Request 对象和一个 call_next 函数。
  • response = await call_next(request): 调用 call_next 函数来处理请求，并获取响应。
  • response.headers["X-Process-Time"] = str(process_time): 在响应头中添加一个 X-Process-Time 头部，用于记录请求的处理时间。

总结

本教程涵盖了 FastAPI 的核心概念，包括：

  • 安装和运行 FastAPI 应用
  • 定义路由和 HTTP 方法
  • 使用 Pydantic 数据模型进行数据验证
  • 使用依赖注入
  • 自定义状态码和响应
  • 使用中间件

通过学习本教程，您应该能够使用 FastAPI 构建简单的 RESTful API。 FastAPI 提供了许多其他功能，例如：

  • 安全性
  • 文件上传
  • WebSocket
  • 异步任务

建议您查阅 FastAPI 的官方文档，以了解更多信息： https://fastapi.tiangolo.com/

希望本教程能帮助您快速入门 FastAPI，并构建出色的 API！

发表评论

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

滚动至顶部