FastAPI教程：快速构建Python RESTful API – wiki基地

作者： /

FastAPI教程：快速构建Python RESTful API

在现代Web开发中，API（应用程序编程接口）扮演着至关重要的角色。它们是不同软件组件之间通信的桥梁，使得数据交换和功能集成成为可能。Python作为一门功能强大且易于上手的语言，拥有众多Web框架，而FastAPI凭借其卓越的性能、现代化的特性以及对异步编程的天然支持，迅速成为构建RESTful API的热门选择。本教程将带您深入了解FastAPI，从基础概念到实际应用，帮助您快速掌握使用FastAPI构建高效、可靠的Python RESTful API。

什么是FastAPI？

FastAPI是一个现代、快速（高性能）的Web框架，用于基于标准的Python类型提示构建API。它由Sebastián Ramírez (@tiangolo) 创建并维护。

FastAPI的核心特性：

  1. 快速高效：FastAPI的性能与NodeJS和Go相当，这得益于其底层的Starlette（用于Web部分）和Pydantic（用于数据部分）。
  2. 编码快速：得益于Python的简洁性和类型提示，开发速度提升约200%至300%。
  3. 更少Bug：减少约40%的人为（开发者）错误，因为类型提示能在编码阶段就发现很多问题。
  4. 智能提示：强大的编辑器支持，自动补全无处不在，减少调试时间。
  5. 易学易用：设计简洁，文档完善，学习曲线平缓。
  6. 代码健壮：自动生成生产可用的代码，并提供交互式文档。
  7. 标准兼容：基于（并完全兼容）API的开放标准：OpenAPI（以前称为Swagger）和JSON Schema。

为什么选择FastAPI？

与其他Python Web框架（如Flask、Django REST framework）相比，FastAPI具有以下显著优势：

  • 类型提示与数据验证：FastAPI强制使用Python的类型提示。这不仅仅是为了代码可读性，更重要的是，FastAPI利用这些类型提示，通过Pydantic库自动进行数据验证、序列化和反序列化。这意味着你定义了期望的数据结构后，FastAPI会自动处理传入数据的校验，确保其符合你的定义，并能将Python对象轻松转换为JSON响应。
  • 自动交互式API文档：FastAPI会自动根据你的代码和类型提示生成符合OpenAPI规范的API文档。你无需额外编写文档，即可拥有Swagger UI和ReDoc两种交互式文档界面。这对于API的测试、调试和团队协作极为便利。
  • 异步支持：FastAPI从一开始就支持异步编程（async/await）。这意味着你可以轻松编写非阻塞的I/O操作，从而在高并发场景下获得极佳的性能，特别适合处理数据库查询、外部API调用等耗时操作。
  • 依赖注入系统：FastAPI内置了一个简单但功能强大的依赖注入系统。这使得代码的组织、重用和测试变得更加容易。例如，你可以轻松地注入数据库会话、认证凭据等。
  • 基于Starlette：FastAPI构建在ASGI（Asynchronous Server Gateway Interface）框架Starlette之上，继承了其高性能和异步特性。
  • 与Pydantic深度集成：Pydantic不仅用于数据验证，还能帮助你定义清晰的数据模型，并自动处理JSON与Python对象之间的转换。

环境准备与安装

在开始之前，请确保你已经安装了Python 3.7+。建议使用虚拟环境来管理项目依赖。

  1. 创建虚拟环境 (可选但推荐):
    bash
    python -m venv venv
    source venv/bin/activate # Linux/macOS
    # venv\Scripts\activate # Windows

  2. 安装FastAPI和Uvicorn:
    FastAPI是一个Web框架，而Uvicorn是一个ASGI服务器，用于运行FastAPI应用。
    bash
    pip install fastapi uvicorn[standard]
    uvicorn[standard]会安装Uvicorn以及一些推荐的依赖，如uvloop（在Linux/macOS上提供更快的事件循环）和httptools（更快的HTTP解析器）。

你的第一个FastAPI应用

让我们从一个经典的 “Hello World” 示例开始。

  1. 创建 main.py 文件:
    “`python
    from fastapi import FastAPI

    创建一个FastAPI应用实例

    app = FastAPI()

    定义一个路径操作装饰器 (Path Operation Decorator)

    @app.get(“/”) # 当GET请求访问根路径”/”时，执行下面的函数
    async def read_root():
    return {“Hello”: “World”}

    @app.get(“/items/{item_id}”) # 路径参数
    async def read_item(item_id: int, q: str | None = None): # item_id是路径参数，q是查询参数
    # item_id: int 表示item_id必须是整数，FastAPI会自动转换和验证
    # q: str | None = None 表示q是一个可选的字符串查询参数，默认值为None
    return {“item_id”: item_id, “q”: q}
    “`

  2. 运行应用:
    在终端中，进入 main.py 所在的目录，然后运行：
    bash
    uvicorn main:app --reload

    • main: 指的是 main.py 文件。
    • app: 指的是在 main.py 中创建的 FastAPI() 实例。
    • --reload: 这个参数让服务器在代码更改后自动重启，非常适合开发阶段。

  3. 访问API:
    打开浏览器，访问以下URL：

    • http://127.0.0.1:8000/
      你会看到JSON响应: {"Hello":"World"}
    • http://127.0.0.1:8000/items/5?q=somequery
      你会看到JSON响应: {"item_id":5,"q":"somequery"}
      尝试将item_id改为非整数，例如http://127.0.0.1:8000/items/foo，FastAPI会返回一个清晰的错误信息，说明路径参数类型不匹配。

  4. 访问自动生成的API文档:
    FastAPI为你自动生成了两种交互式API文档：

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

    在这些界面中，你可以看到API的端点、参数、请求体、响应模型等信息，并且可以直接在浏览器中进行API调用和测试。

核心概念详解

1. 路径参数 (Path Parameters)

路径参数是URL路径的一部分，用于标识特定资源。在FastAPI中，你可以使用与Python格式化字符串相同的语法在路径中声明路径参数，并通过函数参数的类型提示进行类型校验。

“`python
from fastapi import FastAPI

app = FastAPI()

@app.get(“/users/{user_id}”)
async def get_user(user_id: int): # user_id被声明为整数
return {“user_id”: user_id, “message”: f”Details for user {user_id}”}

@app.get(“/files/{file_path:path}”) # 使用:path转换器允许路径包含’/’
async def read_file(file_path: str):
return {“file_path”: file_path}
``
如果访问/users/abc，FastAPI会返回422 Unprocessable Entity错误，因为abc`无法转换为整数。

2. 查询参数 (Query Parameters)

查询参数是URL中 ? 之后的部分，用于过滤或传递额外信息。在FastAPI中，任何未在路径中声明的函数参数，如果具有默认值（或类型提示为Optional或Union[X, None]），则被视为查询参数。

“`python
from fastapi import FastAPI
from typing import Optional, List # Union在Python 3.10+中可以用 | 替代

app = FastAPI()

@app.get(“/search/”)
async def search_items(
keyword: str, # 必需的查询参数
limit: int = 10, # 可选查询参数，默认值为10
skip: Optional[int] = None, # 可选查询参数，默认为None
tags: List[str] = [] # 可选的列表查询参数，可通过多次传递同名参数实现，如 /search/?tags=python&tags=fastapi
):
results = {“keyword”: keyword, “limit”: limit, “skip”: skip, “tags”: tags}
# 实际应用中这里会进行数据库查询等操作
return results
``
访问示例：
*http://127.0.0.1:8000/search/?keyword=fastapi*http://127.0.0.1:8000/search/?keyword=python&limit=5&skip=0*http://127.0.0.1:8000/search/?keyword=web&tags=python&tags=fastapi`

FastAPI同样会对查询参数进行类型转换和校验。如果limit传入非整数，会报错。

3. 请求体 (Request Body) 与 Pydantic模型

当需要客户端向API发送数据时（例如创建或更新资源），通常使用请求体。FastAPI使用Pydantic模型来定义、验证和文档化请求体。

Pydantic是一个数据验证库，它利用Python的类型提示来强制执行数据模式。

“`python
from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional, List

app = FastAPI()

定义一个Pydantic模型

class Item(BaseModel):
name: str
description: Optional[str] = None # 可选字段
price: float = Field(gt=0, description=”The price must be greater than zero”) # price必须大于0
tags: List[str] = []

class User(BaseModel):
username: str
email: str
full_name: Optional[str] = None

@app.post(“/items/”) # 使用POST方法创建资源
async def create_item(item: Item): # 将请求体声明为Item类型
# FastAPI会自动将JSON请求体解析并验证为Item对象
return {“item_name”: item.name, “price_with_tax”: item.price * 1.1}

@app.put(“/items/{item_id}”)
async def update_item(item_id: int, item: Item, owner: User): # 可以同时接收路径参数和多个请求体(FastAPI会将多个Pydantic模型组合)
return {“item_id”: item_id, “updated_item”: item, “owner”: owner}
“`

当客户端发送一个POST请求到 /items/ 时，请求体应该是类似这样的JSON：
json
{
"name": "My Item",
"description": "This is a fantastic item.",
"price": 19.99,
"tags": ["new", "featured"]
}
FastAPI会：
1. 读取请求体为JSON。
2. 将其内容转换为Item模型。
3. 验证数据：
* name 是否是字符串且存在。
* description 如果存在，是否是字符串。
* price 是否是浮点数且大于0。
* tags 如果存在，是否是字符串列表。
4. 如果数据无效，返回一个包含详细错误信息的JSON响应。
5. 如果数据有效，将Item对象传递给create_item函数的item参数。

在/docs中，你会看到Item模型的JSON Schema，以及一个可以直接输入JSON进行测试的界面。

4. 响应模型 (Response Model)

有时，你可能不想返回函数直接返回的所有数据，或者想确保返回的数据结构符合特定模式。这时可以使用response_model参数。

“`python
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from typing import Optional, List

app = FastAPI()

class UserIn(BaseModel): # 用于输入的模型
username: str
password: str
email: EmailStr # Pydantic提供的EmailStr会验证邮件格式
full_name: Optional[str] = None

class UserOut(BaseModel): # 用于输出的模型，不包含密码
username: str
email: EmailStr
full_name: Optional[str] = None

伪造一个用户数据库

fake_users_db = {}

@app.post(“/users/”, response_model=UserOut, status_code=201) # 指定响应模型和成功状态码
async def create_user(user: UserIn):
user_data = user.model_dump() # Pydantic v2: model_dump(), v1: dict()
# 在实际应用中，你会在这里存储用户数据（例如，存入数据库）
# 注意：这里为了演示，直接将UserIn对象（包含密码）存入，
# 但返回时，FastAPI会根据UserOut过滤掉password字段。
fake_users_db[user.username] = user_data
return user # 即使返回的是包含密码的UserIn对象，FastAPI也会按UserOut格式化

@app.get(“/users/{username}”, response_model=UserOut)
async def get_user_by_username(username: str):
if username in fake_users_db:
return fake_users_db[username]
# FastAPI会自动处理HTTPException为相应的HTTP错误响应
from fastapi import HTTPException
raise HTTPException(status_code=404, detail=”User not found”)
``
使用response_model=UserOut可以：
* **数据过滤**：只返回UserOut模型中定义的字段。即使create_user函数返回了包含password的对象，响应中也不会有password。
* **数据验证**：确保返回的数据符合UserOut`的结构和类型。
* 文档化：在API文档中清晰地指明响应的结构。

status_code=201 指定了成功创建资源时的HTTP状态码。

5. 表单数据 (Form Data)

除了JSON，API有时也需要处理HTML表单提交的数据（application/x-www-form-urlencoded）。FastAPI使用Form来处理表单字段。

首先，你需要安装python-multipart库，因为它用于解析表单数据：
pip install python-multipart

“`python
from fastapi import FastAPI, Form

app = FastAPI()

@app.post(“/login/”)
async def login(username: str = Form(…), password: str = Form(…)):
# Form(…) 表示这是一个必需的表单字段
# 实际应用中这里会验证用户名和密码
return {“username”: username, “message”: “Login successful”}
``
在/docs`中，你会看到它将请求体识别为表单数据，并提供相应的输入字段。

6. 依赖注入 (Dependencies)

依赖注入是FastAPI一个非常强大的特性。它允许你在路径操作函数运行之前执行一些代码，并共享这些代码的结果。常见的用例包括：数据库会话管理、用户认证和授权、日志记录等。

“`python
from fastapi import FastAPI, Depends, HTTPException, status, Header
from typing import Annotated # Python 3.9+ for Annotated

app = FastAPI()

一个简单的依赖项：获取并验证X-Token头部

async def verify_token(x_token: Annotated[str | None, Header()] = None):
if x_token != “fake-super-secret-token”:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=”X-Token header invalid”
)
return x_token # 返回的值可以被路径操作函数使用

另一个依赖项：可以依赖于其他依赖项

async def verify_key(x_key: Annotated[str | None, Header()] = None):
if x_key != “fake-super-secret-key”:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=”X-Key header invalid”
)
# 此依赖项不返回任何内容，仅用于执行检查
return x_key

路径操作函数使用Depends来声明依赖

@app.get(“/items_secure/”)
async def read_secure_items(token: Annotated[str, Depends(verify_token)], key: Annotated[str, Depends(verify_key)]):
# 如果verify_token或verify_key抛出HTTPException，此函数不会执行
return {“message”: “Access granted with token and key!”, “token”: token, “key”: key}

也可以在路径操作装饰器中为整个路径组应用依赖

@app.get(“/users/”, dependencies=[Depends(verify_token), Depends(verify_key)])

async def read_users():

return [{“username”: “Rick”}, {“username”: “Morty”}]

依赖项也可以是类

class CommonQueryParams:
def init(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit

@app.get(“/query_items/”)
async def read_query_items(commons: Annotated[CommonQueryParams, Depends()]): # FastAPI会自动处理类依赖
# commons = CommonQueryParams(q=q, skip=skip, limit=limit)
response = {}
if commons.q:
response.update({“q”: commons.q})
# 假设我们有一个项目列表
items = [“item1”, “item2”, “item3”, “item4”, “item5”]
response.update({
“items”: items[commons.skip : commons.skip + commons.limit],
“skip”: commons.skip,
“limit”: commons.limit
})
return response
``Annotated[str, Depends(verify_token)]是推荐的现代写法，它结合了类型提示和依赖注入的元数据。如果verify_token或verify_key中的条件不满足并抛出HTTPException，请求会立即中止，并且相应的错误响应会被发送给客户端，路径操作函数read_secure_items` 不会被执行。

7. 错误处理 (Error Handling)

FastAPI会自动处理Pydantic验证错误，并返回包含详细信息的422 Unprocessable Entity响应。对于其他类型的预期错误，你应该使用HTTPException。

“`python
from fastapi import FastAPI, HTTPException

app = FastAPI()

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

@app.get(“/items-error/{item_id}”)
async def read_item_error(item_id: str):
if item_id not in items:
raise HTTPException(
status_code=404,
detail=”Item not found”,
headers={“X-Error”: “There was an error trying to retrieve this item”},
)
return {“item”: items[item_id]}
``
当访问/items-error/bar(假设"bar"不在items中)，客户端会收到一个404 Not Found响应，JSON内容为{“detail”: “Item not found”}，并且响应头中会包含X-Error`。

你还可以使用 @app.exception_handler(CustomException) 来定义自定义异常处理器，以覆盖默认的错误处理行为或处理自定义异常。

8. 异步 (async def) 与同步 (def)

FastAPI支持使用async def定义的异步路径操作函数，也支持使用def定义的同步函数。

  • 对于async def函数：FastAPI会直接await它们。这适用于执行I/O密集型操作（如数据库查询、外部API调用）而不会阻塞事件循环。
  • 对于def函数：FastAPI会将它们在一个外部线程池中运行，然后await其结果。这意味着即使你使用了阻塞的库（如标准的requests库或一些ORM的同步操作），它们也不会阻塞主事件循环。

“`python
import time
from fastapi import FastAPI

app = FastAPI()

@app.get(“/async-path”)
async def async_operation():
# 假设这里有一个异步的I/O操作，例如:
# await some_async_db_query()
# await asyncio.sleep(1) # 模拟异步I/O
return {“message”: “Async operation complete”}

@app.get(“/sync-path”)
def sync_operation():
# 这是一个同步的、可能阻塞的操作
time.sleep(1) # 模拟阻塞I/O
return {“message”: “Sync operation complete (run in thread pool)”}
``
选择async def还是def取决于你的函数内部是否使用了await以及是否需要执行真正的异步I/O。如果你的代码是CPU密集型的，或者没有异步操作，使用普通的def即可。如果你的代码需要调用其他async函数或库，使用async def`。

9. 中间件 (Middleware)

中间件是一种在请求到达路径操作函数之前和响应发送给客户端之前处理请求和响应的函数。它可以用于日志记录、添加自定义头部、GZip压缩、CORS处理等。

“`python
import time
from fastapi import FastAPI, Request

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)
return response

@app.get(“/”)
async def main():
return {“message”: “Hello World”}
``
这个中间件会计算每个请求的处理时间，并将其作为X-Process-Time`头部添加到响应中。

10. 静态文件和模板

FastAPI（通过Starlette）也支持提供静态文件（如CSS、JavaScript、图片）和使用Jinja2等模板引擎渲染HTML页面。

静态文件:
“`python
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

挂载静态文件目录，所有在 “static” 目录下的文件都可以通过 “/static” 路径访问

例如 static/css/styles.css 可以通过 http://localhost:8000/static/css/styles.css 访问

app.mount(“/static”, StaticFiles(directory=”static”), name=”static”)

@app.get(“/”)
async def root():
return {“message”: “Hello, visit /static/somefile.txt (if it exists)”}
``
你需要创建一个名为static` 的目录，并在其中放置你的静态文件。

Jinja2模板:
首先安装Jinja2: pip install jinja2
“`python
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates

app = FastAPI()

假设你的模板文件放在 “templates” 目录下

templates = Jinja2Templates(directory=”templates”)

@app.get(“/view/{item_name}”, response_class=HTMLResponse)
async def read_item_view(request: Request, item_name: str):
# request参数是必需的
return templates.TemplateResponse(
“item_view.html”, # 模板文件名
{“request”: request, “name”: item_name, “id”: 123} # 传递给模板的上下文
)
你需要创建一个`templates`目录，并在其中创建一个`item_view.html`文件，例如：html





Item: {{ name }}

Details for Item: {{ name }}

ID: {{ id }}


“`

构建更大型的应用：APIRouter

当应用变得复杂时，将所有路径操作都放在一个文件中会变得难以管理。FastAPI提供了APIRouter，允许你将API路径组织成模块化的部分。

  1. 创建子路由文件 (例如 routers/users.py):
    “`python
    # routers/users.py
    from fastapi import APIRouter, HTTPException
    from pydantic import BaseModel
    from typing import List

    router = APIRouter(
    prefix=”/users”, # 所有此路由下的路径都会有 /users 前缀
    tags=[“users”], # 在API文档中为这些路径操作分组
    responses={404: {“description”: “Not found”}}, # 为此路由组定义通用响应
    )

    class User(BaseModel):
    id: int
    username: str
    email: str

    fake_db_users = [
    User(id=1, username=”john_doe”, email=”[email protected]”),
    User(id=2, username=”jane_doe”, email=”[email protected]”),
    ]

    @router.get(“/”, response_model=List[User])
    async def get_all_users():
    return fake_db_users

    @router.get(“/{user_id}”, response_model=User)
    async def get_user(user_id: int):
    user = next((user for user in fake_db_users if user.id == user_id), None)
    if user is None:
    raise HTTPException(status_code=404, detail=”User not found”)
    return user
    “`

  2. 在主应用中包含路由器 (例如 main.py):
    “`python
    # main.py
    from fastapi import FastAPI
    from routers import users # 假设routers是一个包 (有__init__.py)

    app = FastAPI()

    app.include_router(users.router) # 包含users路由

    @app.get(“/”)
    async def root():
    return {“message”: “Welcome to the main API”}
    “`

现在，用户相关的API端点（如 /users/ 和 /users/{user_id}）由users.router处理，而主应用app保持简洁。

总结与展望

FastAPI是一个功能强大、现代且高效的Python Web框架，特别适合构建RESTful API。它通过巧妙地结合Python类型提示、Pydantic的数据验证以及Starlette的异步Web能力，为开发者提供了无与伦比的开发体验和运行时性能。

本教程覆盖了FastAPI的核心概念和常用功能，包括路径参数、查询参数、请求体、响应模型、依赖注入、错误处理、异步操作和应用结构化。通过这些知识，你已经具备了使用FastAPI构建复杂API应用的基础。

要进一步深入学习，强烈建议查阅FastAPI的官方文档，它非常详尽且易于理解。此外，社区中也有大量优秀的教程和项目可供参考。随着你对FastAPI的掌握越来越深入，你会发现它不仅能提高开发效率，还能帮助你编写出更健壮、更易于维护的代码。

FastAPI的生态系统也在不断发展，支持更多的数据库、认证方案和部署选项。它无疑是Python Web开发领域一颗冉冉升起的新星，值得每一位Python开发者关注和学习。祝你在FastAPI的旅程中一切顺利！

发表评论

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

滚动至顶部