FastAPI 请求和响应(一文讲透)
💡一则或许对你有用的小广告
欢迎加入小哈的星球 ,你将获得:专属的项目实战 / 1v1 提问 / Java 学习路线 / 学习打卡 / 每月赠书 / 社群讨论
- 新项目:《从零手撸:仿小红书(微服务架构)》 正在持续爆肝中,基于
Spring Cloud Alibaba + Spring Boot 3.x + JDK 17...
,点击查看项目介绍 ;演示链接: http://116.62.199.48:7070 ;- 《从零手撸:前后端分离博客项目(全栈开发)》 2 期已完结,演示链接: http://116.62.199.48/ ;
截止目前, 星球 内专栏累计输出 90w+ 字,讲解图 3441+ 张,还在持续爆肝中.. 后续还会上新更多项目,目标是将 Java 领域典型的项目都整一波,如秒杀系统, 在线商城, IM 即时通讯,权限管理,Spring Cloud Alibaba 微服务等等,已有 3100+ 小伙伴加入学习 ,欢迎点击围观
什么是 FastAPI 请求和响应?
FastAPI 是一个基于 Python 的现代 Web 框架,以高性能、类型注解和开发者友好性著称。在 FastAPI 中,请求和响应是构建 API 的核心流程:请求(Request)是客户端向服务器发送的数据包,而响应(Response)是服务器对请求的处理结果。两者如同快递服务中的“寄件”和“收件”,构成了 API 的完整交互链路。
通过本文,你将掌握如何利用 FastAPI 处理不同类型的请求、构建灵活的响应,并深入理解其底层原理。无论是开发 RESTful API 还是 GraphQL 端点,这些技能都将帮助你高效地实现需求。
FastAPI 请求的类型与处理
1. 请求的基础构成
一个典型的 HTTP 请求包含以下要素:
- 方法(Method):如 GET、POST、PUT 等,定义操作类型。
- 路径(Path):如
/api/users
,指定资源位置。 - 参数(Parameters):附加数据,如查询参数(Query Parameters)或请求体(Body)。
- 头部(Headers):如
Content-Type
,定义数据格式。
2. FastAPI 处理请求的核心工具:Path Operations
FastAPI 通过装饰器(Decorator)将 Python 函数映射到 HTTP 方法和路径上。例如:
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
此代码定义了一个 GET 请求,当客户端访问 /items/123
时,FastAPI 会调用 read_item
函数,并将路径中的 item_id
转换为整数类型。
3. 路径参数与查询参数
路径参数(Path Parameters)
路径参数直接嵌入 URL 中,通过 {}
标记。例如:
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: str):
return {"user_id": user_id, "item_id": item_id}
此接口接收两个参数:user_id
(整数)和 item_id
(字符串)。
查询参数(Query Parameters)
查询参数通过 ?key=value
附加在 URL 后。FastAPI 使用 Query
类型注解来定义:
from fastapi import Query
@app.get("/search")
async def search(q: str = Query(..., min_length=3, max_length=50)):
return {"query": q}
此处,q
是必填参数,且长度需在 3 到 50 之间。
4. 请求体(Body)的处理
对于 POST、PUT 等方法,数据通常通过请求体传递。FastAPI 支持 JSON、表单、文件等多种格式,但最常用的是 JSON。
示例:创建用户
from pydantic import BaseModel
class User(BaseModel):
name: str
email: str
is_active: bool = True
@app.post("/users")
async def create_user(user: User):
return {"user": user.dict()}
当客户端发送以下 JSON 请求时:
{
"name": "Alice",
"email": "alice@example.com"
}
FastAPI 会自动将 JSON 数据映射到 User
模型,并验证字段合法性(如 is_active
默认为 True
)。
FastAPI 响应的构建与优化
1. 响应的基本结构
FastAPI 默认返回 JSON 格式的响应,但开发者可以通过以下方式自定义:
- 响应模型(Response Model):定义返回数据的格式。
- 状态码(Status Code):如
200 OK
、404 Not Found
。 - 响应头(Headers):如
Content-Type
。
示例:指定状态码和响应模型
from fastapi import status
@app.post("/login", response_model=User, status_code=status.HTTP_201_CREATED)
async def login(user: User):
return user
此接口返回 201 Created
状态码,并确保响应符合 User
模型的结构。
2. 自定义响应类
若需返回非 JSON 数据(如 HTML、文件流),可使用 FastAPI 内置的 Response
类:
from fastapi.responses import HTMLResponse
@app.get("/html", response_class=HTMLResponse)
async def get_html():
return """
<html>
<body><h1>Hello from FastAPI!</h1></body>
</html>
"""
此代码将返回 HTML 内容,并自动设置 Content-Type: text/html
。
3. 响应头与元数据
通过 Response
对象或装饰器参数,可以添加自定义响应头:
from fastapi import Response
@app.get("/custom-headers")
async def custom_headers(response: Response):
response.headers["X-Custom-Header"] = "FastAPI-Rocks"
return {"message": "Success"}
客户端收到的响应中将包含 X-Custom-Header: FastAPI-Rocks
。
高级技巧:依赖注入与异步处理
1. 共享逻辑:依赖注入(Dependencies)
通过 Depends
机制,可将公共逻辑(如认证、数据库连接)封装为可复用的函数。例如:
from fastapi import Depends
async def get_db():
db = DatabaseConnection()
try:
yield db
finally:
await db.close()
@app.get("/items")
async def read_items(db: Database = Depends(get_db)):
items = await db.query("SELECT * FROM items")
return items
此代码确保每次请求都会自动获取数据库连接,并在结束后释放资源。
2. 异步请求与高性能
FastAPI 支持异步函数(async
/await
),适合处理 I/O 密集型任务(如网络请求、数据库操作)。例如:
import httpx
@app.get("/external-api")
async def fetch_external_data():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.example.com/data")
return response.json()
通过异步操作,FastAPI 可以在等待外部 API 响应时处理其他请求,显著提升吞吐量。
错误处理与响应控制
1. 返回自定义错误信息
FastAPI 提供 HTTPException
类来抛出标准 HTTP 错误:
from fastapi import HTTPException
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id not in [1, 2, 3]:
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id}
当 item_id
无效时,返回 404 Not Found
和自定义错误信息。
2. 全局异常处理
通过 exception_handler
,可以统一处理特定类型的异常:
@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
return JSONResponse(
status_code=400,
content={"message": f"Invalid input: {str(exc)}"},
)
此代码将所有 ValueError
转换为 400 Bad Request
响应。
实战案例:构建完整的 CRUD API
1. 定义模型与路径操作
假设我们要构建一个管理“书籍”的 API,首先定义数据模型:
class Book(BaseModel):
title: str
author: str
price: float
class BookDB(Book):
id: int
books = [] # 模拟数据库
2. 实现 CRUD 操作
创建书籍(POST)
@app.post("/books", response_model=Book, status_code=201)
async def create_book(book: Book):
new_id = len(books) + 1
new_book = BookDB(id=new_id, **book.dict())
books.append(new_book)
return new_book
查询书籍(GET)
@app.get("/books/{book_id}", response_model=BookDB)
async def read_book(book_id: int):
if book_id <= 0 or book_id > len(books):
raise HTTPException(status_code=404, detail="Book not found")
return books[book_id - 1]
更新书籍(PUT)
@app.put("/books/{book_id}", response_model=BookDB)
async def update_book(book_id: int, updated_book: Book):
if book_id <= 0 or book_id > len(books):
raise HTTPException(status_code=404, detail="Book not found")
book = books[book_id - 1]
updated = BookDB(id=book.id, **updated_book.dict())
books[book_id - 1] = updated
return updated
删除书籍(DELETE)
@app.delete("/books/{book_id}", status_code=204)
async def delete_book(book_id: int):
if book_id <= 0 or book_id > len(books):
raise HTTPException(status_code=404, detail="Book not found")
del books[book_id - 1]
return Response(status_code=204) # 无内容响应
总结与展望
通过本文,我们深入探讨了 FastAPI 请求和响应的核心机制,包括参数处理、响应构建、异步编程和错误管理。从简单到复杂,从基础概念到实战案例,FastAPI 的灵活性和强大功能得以充分体现。
对于初学者,建议从基础路径操作开始,逐步尝试依赖注入和异步编程;中级开发者则可进一步探索自定义响应类、全局异常处理等高级功能。随着 FastAPI 在微服务、API 网关等场景中的广泛应用,掌握其请求与响应的处理逻辑,将成为构建高效、可维护的后端系统的必备技能。
下一步行动:
- 用 FastAPI 搭建一个小型项目(如待办事项列表或天气查询 API)。
- 探索 FastAPI 的 OpenAPI 文档功能(自动生成功能)。
- 结合数据库(如 PostgreSQL 或 MongoDB)实现持久化存储。
通过持续实践,你将更熟练地驾驭 FastAPI,为构建现代化的 Web 应用铺平道路!