FastAPI 教程：从零开始构建 RESTful API – wiki基地

作者： /

FastAPI 教程：从零开始构建 RESTful API

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API，使用标准 Python 类型提示。 它具有诸多优点，例如：

  • 速度快: 性能媲美 Node.js 和 Go (基于 Starlette 和 Pydantic)。
  • 易于编码: 开发速度提高约 200% 至 300%。
  • 更少的 Bug: 减少约 40% 的人为错误。
  • 直观: 强大的编辑器支持，可以自动完成代码。
  • 简单: 设计得易于学习和使用。
  • 健壮: 用于生产环境的代码。
  • 基于标准: 基于并完全兼容 API 的开放标准：OpenAPI 和 JSON Schema。

本教程将带你从零开始构建一个简单的 RESTful API，让你了解 FastAPI 的核心概念和用法。我们将创建一个简单的博客文章 API，它具有创建、读取、更新和删除文章的功能。

环境准备

在开始之前，你需要确保已经安装了 Python 3.7+。然后，你需要安装 FastAPI 和 Uvicorn（一个 ASGI 服务器，用于运行 FastAPI 应用）。

bash
pip install fastapi uvicorn

项目结构

首先，创建一个项目文件夹，并在其中创建一个名为 main.py 的文件，用于存放你的 FastAPI 应用代码。

my_blog_api/
├── main.py
└── ...

创建 FastAPI 应用实例

在 main.py 文件中，导入 FastAPI 类并创建一个应用实例。

“`python
from fastapi import FastAPI

app = FastAPI()

@app.get(“/”)
async def read_root():
return {“message”: “Welcome to my blog API!”}
“`

这几行代码做了以下几件事：

  • from fastapi import FastAPI: 从 FastAPI 库导入 FastAPI 类。
  • app = FastAPI(): 创建一个 FastAPI 类的实例，并将其赋值给变量 app。这个 app 对象将是你的 API 的核心。
  • @app.get("/"): 这是一个装饰器，它将下面的函数与 HTTP GET 方法和根路径 / 关联起来。这表示当客户端向 / 发送 GET 请求时，该函数将被执行。
  • async def read_root():: 定义一个异步函数 read_root。异步函数是 FastAPI 的推荐用法，可以提高并发性能。
  • return {"message": "Welcome to my blog API!"}: 函数返回一个 Python 字典。FastAPI 会自动将这个字典转换为 JSON 格式并作为 HTTP 响应返回给客户端。

运行 FastAPI 应用

在终端中，使用 Uvicorn 运行你的 FastAPI 应用。

bash
uvicorn main:app --reload

  • uvicorn main:app: 告诉 Uvicorn 运行 main.py 文件中的 app 对象。
  • --reload: 启用自动重载功能。当你的代码发生更改时，Uvicorn 会自动重启服务器，这对于开发非常方便。

现在，在你的浏览器中访问 http://localhost:8000/，你将会看到 JSON 响应 {"message": "Welcome to my blog API!"}。

添加 API 端点

接下来，我们将添加更多的 API 端点来处理博客文章。

1. 创建博客文章的数据模型

我们需要定义一个数据模型来表示博客文章。我们可以使用 Pydantic 来完成这个任务。 Pydantic 是一个数据验证和设置管理库，它与 FastAPI 集成得很好。

“`python
from typing import Optional

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Post(BaseModel):
id: int
title: str
content: str
published: bool = True
rating: Optional[int] = None

@app.get(“/”)
async def read_root():
return {“message”: “Welcome to my blog API!”}
“`

  • from pydantic import BaseModel: 从 Pydantic 库导入 BaseModel 类。
  • class Post(BaseModel):: 定义一个名为 Post 的类，它继承自 BaseModel。Post 类将用于定义博客文章的数据结构。
  • id: int: 定义 id 字段，类型为整数。
  • title: str: 定义 title 字段，类型为字符串。
  • content: str: 定义 content 字段，类型为字符串。
  • published: bool = True: 定义 published 字段，类型为布尔值，默认值为 True。
  • rating: Optional[int] = None: 定义 rating 字段，类型为可选的整数，默认值为 None。 Optional 来自 typing 模块，表示该字段可以为 None。

2. 创建存储博客文章的内存数据库

为了简单起见，我们将使用一个简单的列表来模拟数据库。在实际应用中，你可能会使用真正的数据库，例如 PostgreSQL 或 MySQL。

“`python
from typing import Optional

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Post(BaseModel):
id: int
title: str
content: str
published: bool = True
rating: Optional[int] = None

posts = [] # 用于存储博客文章的列表

@app.get(“/”)
async def read_root():
return {“message”: “Welcome to my blog API!”}
“`

  • posts = []: 创建一个名为 posts 的空列表。这个列表将用于存储博客文章。

3. 创建 POST 端点来创建博客文章

“`python
from typing import Optional

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Post(BaseModel):
id: int
title: str
content: str
published: bool = True
rating: Optional[int] = None

posts = [] # 用于存储博客文章的列表

@app.get(“/”)
async def read_root():
return {“message”: “Welcome to my blog API!”}

@app.post(“/posts”)
async def create_post(post: Post):
posts.append(post)
return post
“`

  • @app.post("/posts"): 这是一个装饰器，它将下面的函数与 HTTP POST 方法和路径 /posts 关联起来。这表示当客户端向 /posts 发送 POST 请求时，该函数将被执行。
  • async def create_post(post: Post):: 定义一个异步函数 create_post。它接受一个 Post 类型的参数 post。FastAPI 会自动将客户端发送的 JSON 请求体解析为 Post 对象。
  • posts.append(post): 将 post 对象添加到 posts 列表中。
  • return post: 返回 post 对象。FastAPI 会自动将这个对象转换为 JSON 格式并作为 HTTP 响应返回给客户端。

4. 创建 GET 端点来获取所有博客文章

“`python
from typing import Optional

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Post(BaseModel):
id: int
title: str
content: str
published: bool = True
rating: Optional[int] = None

posts = [] # 用于存储博客文章的列表

@app.get(“/”)
async def read_root():
return {“message”: “Welcome to my blog API!”}

@app.post(“/posts”)
async def create_post(post: Post):
posts.append(post)
return post

@app.get(“/posts”)
async def get_posts():
return posts
“`

  • @app.get("/posts"): 这是一个装饰器，它将下面的函数与 HTTP GET 方法和路径 /posts 关联起来。这表示当客户端向 /posts 发送 GET 请求时，该函数将被执行。
  • async def get_posts():: 定义一个异步函数 get_posts。
  • return posts: 返回 posts 列表。FastAPI 会自动将这个列表转换为 JSON 格式并作为 HTTP 响应返回给客户端。

5. 创建 GET 端点来获取单个博客文章

“`python
from typing import Optional

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Post(BaseModel):
id: int
title: str
content: str
published: bool = True
rating: Optional[int] = None

posts = [] # 用于存储博客文章的列表

@app.get(“/”)
async def read_root():
return {“message”: “Welcome to my blog API!”}

@app.post(“/posts”)
async def create_post(post: Post):
posts.append(post)
return post

@app.get(“/posts”)
async def get_posts():
return posts

@app.get(“/posts/{post_id}”)
async def get_post(post_id: int):
for post in posts:
if post.id == post_id:
return post
raise HTTPException(status_code=404, detail=”Post not found”)
“`

  • @app.get("/posts/{post_id}"): 这是一个装饰器，它将下面的函数与 HTTP GET 方法和路径 /posts/{post_id} 关联起来。{post_id} 是一个路径参数，它允许我们从 URL 中提取 post_id 的值。
  • async def get_post(post_id: int):: 定义一个异步函数 get_post。它接受一个整数类型的参数 post_id。
  • for post in posts:: 循环遍历 posts 列表。
  • if post.id == post_id:: 检查当前博客文章的 id 是否等于请求的 post_id。
  • return post: 如果找到匹配的博客文章，则返回该文章。
  • raise HTTPException(status_code=404, detail="Post not found"): 如果没有找到匹配的博客文章，则抛出一个 HTTPException 异常。HTTPException 是 FastAPI 提供的异常类，它可以用于返回 HTTP 错误响应。在这里，我们返回一个 404 Not Found 错误，并附带一条消息 “Post not found”。

6. 创建 PUT 端点来更新博客文章

“`python
from typing import Optional

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Post(BaseModel):
id: int
title: str
content: str
published: bool = True
rating: Optional[int] = None

posts = [] # 用于存储博客文章的列表

@app.get(“/”)
async def read_root():
return {“message”: “Welcome to my blog API!”}

@app.post(“/posts”)
async def create_post(post: Post):
posts.append(post)
return post

@app.get(“/posts”)
async def get_posts():
return posts

@app.get(“/posts/{post_id}”)
async def get_post(post_id: int):
for post in posts:
if post.id == post_id:
return post
raise HTTPException(status_code=404, detail=”Post not found”)

@app.put(“/posts/{post_id}”)
async def update_post(post_id: int, updated_post: Post):
for i, post in enumerate(posts):
if post.id == post_id:
posts[i] = updated_post
return updated_post
raise HTTPException(status_code=404, detail=”Post not found”)
“`

  • @app.put("/posts/{post_id}"): 这是一个装饰器，它将下面的函数与 HTTP PUT 方法和路径 /posts/{post_id} 关联起来。
  • async def update_post(post_id: int, updated_post: Post):: 定义一个异步函数 update_post。它接受两个参数：一个整数类型的 post_id 和一个 Post 类型的 updated_post。
  • for i, post in enumerate(posts):: 循环遍历 posts 列表，并使用 enumerate 函数获取索引 i 和博客文章 post。
  • if post.id == post_id:: 检查当前博客文章的 id 是否等于请求的 post_id。
  • posts[i] = updated_post: 如果找到匹配的博客文章，则使用 updated_post 替换该文章。
  • return updated_post: 返回 updated_post 对象。
  • raise HTTPException(status_code=404, detail="Post not found"): 如果没有找到匹配的博客文章，则抛出一个 HTTPException 异常。

7. 创建 DELETE 端点来删除博客文章

“`python
from typing import Optional

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Post(BaseModel):
id: int
title: str
content: str
published: bool = True
rating: Optional[int] = None

posts = [] # 用于存储博客文章的列表

@app.get(“/”)
async def read_root():
return {“message”: “Welcome to my blog API!”}

@app.post(“/posts”)
async def create_post(post: Post):
posts.append(post)
return post

@app.get(“/posts”)
async def get_posts():
return posts

@app.get(“/posts/{post_id}”)
async def get_post(post_id: int):
for post in posts:
if post.id == post_id:
return post
raise HTTPException(status_code=404, detail=”Post not found”)

@app.put(“/posts/{post_id}”)
async def update_post(post_id: int, updated_post: Post):
for i, post in enumerate(posts):
if post.id == post_id:
posts[i] = updated_post
return updated_post
raise HTTPException(status_code=404, detail=”Post not found”)

@app.delete(“/posts/{post_id}”)
async def delete_post(post_id: int):
for i, post in enumerate(posts):
if post.id == post_id:
posts.pop(i)
return {“message”: “Post deleted successfully”}
raise HTTPException(status_code=404, detail=”Post not found”)
“`

  • @app.delete("/posts/{post_id}"): 这是一个装饰器，它将下面的函数与 HTTP DELETE 方法和路径 /posts/{post_id} 关联起来。
  • async def delete_post(post_id: int):: 定义一个异步函数 delete_post。它接受一个整数类型的参数 post_id。
  • for i, post in enumerate(posts):: 循环遍历 posts 列表，并使用 enumerate 函数获取索引 i 和博客文章 post。
  • if post.id == post_id:: 检查当前博客文章的 id 是否等于请求的 post_id。
  • posts.pop(i): 如果找到匹配的博客文章，则从 posts 列表中删除该文章。
  • return {"message": "Post deleted successfully"}: 返回一个包含消息 “Post deleted successfully” 的字典。
  • raise HTTPException(status_code=404, detail="Post not found"): 如果没有找到匹配的博客文章，则抛出一个 HTTPException 异常。

完整的代码

以下是 main.py 文件的完整代码：

“`python
from typing import Optional

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Post(BaseModel):
id: int
title: str
content: str
published: bool = True
rating: Optional[int] = None

posts = [] # 用于存储博客文章的列表

@app.get(“/”)
async def read_root():
return {“message”: “Welcome to my blog API!”}

@app.post(“/posts”)
async def create_post(post: Post):
posts.append(post)
return post

@app.get(“/posts”)
async def get_posts():
return posts

@app.get(“/posts/{post_id}”)
async def get_post(post_id: int):
for post in posts:
if post.id == post_id:
return post
raise HTTPException(status_code=404, detail=”Post not found”)

@app.put(“/posts/{post_id}”)
async def update_post(post_id: int, updated_post: Post):
for i, post in enumerate(posts):
if post.id == post_id:
posts[i] = updated_post
return updated_post
raise HTTPException(status_code=404, detail=”Post not found”)

@app.delete(“/posts/{post_id}”)
async def delete_post(post_id: int):
for i, post in enumerate(posts):
if post.id == post_id:
posts.pop(i)
return {“message”: “Post deleted successfully”}
raise HTTPException(status_code=404, detail=”Post not found”)
“`

使用 API 文档

FastAPI 自动生成交互式的 API 文档。 你可以在 http://localhost:8000/docs 访问它。 你可以使用这个文档来测试你的 API 端点。

总结

本教程介绍了使用 FastAPI 从零开始构建 RESTful API 的基本步骤。 你学习了如何：

  • 创建 FastAPI 应用实例。
  • 定义数据模型。
  • 创建 API 端点来处理不同的 HTTP 方法（GET, POST, PUT, DELETE）。
  • 使用 Pydantic 进行数据验证。
  • 使用 HTTPException 返回错误响应。
  • 利用 FastAPI 自动生成的 API 文档。

这只是 FastAPI 的一个入门教程。 FastAPI 还有许多其他功能，例如依赖注入、安全性、中间件等。 建议你阅读 FastAPI 的官方文档来了解更多信息。

希望本教程对你有所帮助！

发表评论

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

滚动至顶部