

【个人主页:玄同765】
大语言模型(LLM)开发工程师|中国传媒大学·数字媒体技术(智能交互与游戏设计) 深耕领域:大语言模型开发 / RAG知识库 / AI Agent落地 / 模型微调 技术栈:Python / LangChain/RAG(Dify+Redis+Milvus)| SQL/NumPy | FastAPI+Docker ️ 工程能力:专注模型工程化部署、知识库构建与优化,擅长全流程解决方案 专栏传送门:LLM大模型开发 项目实战指南、Python 从真零基础到纯文本 LLM 全栈实战、从零学 SQL + 大模型应用落地、大模型开发小白专属:从 0 入门 Linux&Shell 「让AI交互更智能,让技术落地更高效」 欢迎技术探讨/项目合作! 关注我,解锁大模型与智能交互的无限可能!
本文用快递柜取件的大白话场景引出 FastAPI 路由的核心概念,并结合专业术语进行深度解析,对比 Flask、Django 路由的特点。之后详细讲解基础路由、路径参数、查询参数、路由组、嵌套路由、依赖注入路由、生命周期路由、WebHook 路由等核心用法,附带 3 个完整实战项目(待办事项 API、博客文章 API、电商产品 API),让你快速掌握 FastAPI 路由的设计与实现。
你有没有用过小区门口的智能快递柜?当快递员把快递放进快递柜后,系统会给你发送一条取件短信,里面包含取件码和对应的快递柜编号(比如 A 区 05 号、B 区 12 号)。你需要找到对应的快递柜编号,输入取件码,才能取出属于你的快递。
其实,Web 应用程序的 API 接口和智能快递柜的工作原理非常相似。当你发送一个 HTTP 请求到 Web 应用程序时,请求中会包含路径信息(比如 /api/todos/1、/api/posts/2),Web 应用程序需要根据路径信息找到对应的处理函数,然后执行处理函数,返回响应结果。这个路径信息与处理函数的映射关系,就是路由。
想象一下,你是一家餐厅的前台服务员,负责根据客人的需求安排不同的厨师处理。客人的需求可能是 “一份宫保鸡丁”、“一份鱼香肉丝”、“一份麻婆豆腐” 等等。你需要根据客人的需求找到对应的厨师(比如宫保鸡丁找张师傅,鱼香肉丝找李师傅,麻婆豆腐找王师傅),然后把客人的需求告诉厨师,让厨师开始制作。
在 Web 应用程序中,客人的需求就是 HTTP 请求,前台服务员就是路由系统,厨师就是处理函数。路由系统根据 HTTP 请求中的路径信息找到对应的处理函数,然后执行处理函数,返回响应结果。
路由(Routing)是指 HTTP 请求路径与处理函数之间的映射关系。它的核心作用是将客户端发送的 HTTP 请求分发到对应的处理函数,处理函数执行完业务逻辑后,返回响应结果给客户端。
在 FastAPI 中,路由系统是基于 Starlette 框架的路由系统实现的,它支持自动解析路径参数、自动验证查询参数、自动生成 API 文档等功能。
为了更好地理解 FastAPI 路由的特点,我们可以将其与 Flask、Django 路由进行对比:
框架 | 路由定义方式 | API 文档支持 | 路径参数解析 | 查询参数验证 | 依赖注入支持 | 异步 / 同步兼容 |
|---|---|---|---|---|---|---|
FastAPI | 装饰器 + APIRouter | 自动生成 Swagger/Redoc | 自动解析 + 类型注解验证 | 自动解析 + 类型注解验证 | 支持(依赖注入系统) | 支持异步 / 同步函数 |
Flask | 装饰器 + Blueprint | 需要安装 Flask-RESTX 等扩展 | 手动解析 + 正则表达式验证 | 手动解析 + 类型转换 | 需要安装 Flask-Injector 等扩展 | 支持同步函数,异步需要安装 Flask-Async 等扩展 |
Django | urls.py 配置 + View 类 | 需要安装 Django REST Framework 等扩展 | 手动解析 + 正则表达式验证 | 手动解析 + 序列化器验证 | 需要安装 Django-Injector 等扩展 | 支持同步函数,异步需要 Django 3.1 + 版本 |
FastAPI 路由系统的技术基础是 Starlette 框架的路由系统。Starlette 是一款轻量级、高性能的 ASGI 框架,它提供了路由、中间件、模板渲染等功能。FastAPI 在 Starlette 的基础上,添加了类型注解支持、自动 API 文档生成等功能。
FastAPI 使用装饰器(如 @app.get、@app.post、@app.put、@app.delete)来定义路由。装饰器的参数是 HTTP 请求的路径,装饰器所装饰的函数是对应的处理函数。
以下是基础路由定义的示例代码:
from fastapi import FastAPI
# 创建FastAPI应用程序实例
app = FastAPI()
# 定义GET请求的路由
@app.get("/")
def read_root():
"""处理根路径的GET请求"""
return {"message": "Hello, World!"}
# 定义GET请求的路由,带路径参数
@app.get("/items/{item_id}")
def read_item(item_id: int):
"""处理带路径参数的GET请求"""
return {"item_id": item_id}
# 定义POST请求的路由
@app.post("/items/")
def create_item(item: dict):
"""处理POST请求"""
return {"item": item}
# 定义PUT请求的路由,带路径参数
@app.put("/items/{item_id}")
def update_item(item_id: int, item: dict):
"""处理带路径参数的PUT请求"""
return {"item_id": item_id, "item": item}
# 定义DELETE请求的路由,带路径参数
@app.delete("/items/{item_id}")
def delete_item(item_id: int):
"""处理带路径参数的DELETE请求"""
return {"item_id": item_id}路径参数是指 HTTP 请求路径中的参数,它们通常用于标识资源的唯一 ID。在 FastAPI 中,路径参数的定义方式是在路径中使用 {} 包裹参数名,然后在处理函数的参数中使用对应的参数名,并添加类型注解。
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
def read_item(item_id: int):
"""处理带路径参数的GET请求"""
return {"item_id": item_id}FastAPI 可以根据类型注解对路径参数进行验证。以下是路径参数验证的示例代码:
from fastapi import FastAPI, Path
app = FastAPI()
@app.get("/items/{item_id}")
def read_item(
item_id: int = Path(
..., # 必需参数
title="Item ID", # 标题
description="The unique identifier of the item", # 描述
ge=1, # 大于等于1
le=1000 # 小于等于1000
)
):
"""处理带路径参数的GET请求,参数有验证"""
return {"item_id": item_id}路径参数的类型可以是枚举类型,这样可以限制路径参数的取值范围。以下是路径参数枚举的示例代码:
from fastapi import FastAPI
from enum import Enum
app = FastAPI()
class ItemType(str, Enum):
"""定义项目类型的枚举"""
BOOK = "book"
ELECTRONIC = "electronic"
CLOTHING = "clothing"
@app.get("/items/{item_type}")
def read_items_by_type(item_type: ItemType):
"""处理带路径参数的GET请求,参数是枚举类型"""
if item_type == ItemType.BOOK:
return {"item_type": item_type, "items": ["Book 1", "Book 2"]}
elif item_type == ItemType.ELECTRONIC:
return {"item_type": item_type, "items": ["Electronic 1", "Electronic 2"]}
elif item_type == ItemType.CLOTHING:
return {"item_type": item_type, "items": ["Clothing 1", "Clothing 2"]}
else:
return {"item_type": item_type, "items": []}路径参数的类型可以是字符串类型,并通过正则表达式来限制路径参数的取值范围。以下是路径参数正则表达式的示例代码:
from fastapi import FastAPI
app = FastAPI()
@app.get("/files/{file_path:path}")
def read_file(file_path: str):
"""处理带路径参数的GET请求,参数是路径类型"""
return {"file_path": file_path}在上面的示例中,file_path:path表示 file_path 参数可以包含斜杠(/),这样可以匹配任意路径。
查询参数是指 HTTP 请求 URL 中的查询字符串参数,它们通常用于过滤、排序、分页等操作。在 FastAPI 中,查询参数的定义方式是在处理函数的参数中添加对应的参数名,并添加类型注解。
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):
"""处理带查询参数的GET请求"""
return {"skip": skip, "limit": limit}FastAPI 可以根据类型注解对查询参数进行验证。以下是查询参数验证的示例代码:
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/")
def read_items(
skip: int = Query(
0, # 默认值
title="Skip", # 标题
description="Number of items to skip", # 描述
ge=0 # 大于等于0
),
limit: int = Query(
10, # 默认值
title="Limit", # 标题
description="Number of items to limit", # 描述
ge=1, # 大于等于1
le=100 # 小于等于100
)
):
"""处理带查询参数的GET请求,参数有验证"""
return {"skip": skip, "limit": limit}查询参数可以是可选的,只需要在类型注解后面添加Optional类型即可。以下是可选查询参数的示例代码:
from fastapi import FastAPI, Query
from typing import Optional
app = FastAPI()
@app.get("/items/")
def read_items(
q: Optional[str] = Query(
None, # 默认值为None,可选参数
title="Query", # 标题
description="Search query" # 描述
)
):
"""处理带可选查询参数的GET请求"""
if q:
return {"q": q}
else:
return {"message": "No query parameter provided"}查询参数可以接受多个值,只需要在类型注解后面添加List类型即可。以下是查询参数多个值的示例代码:
from fastapi import FastAPI, Query
from typing import List
app = FastAPI()
@app.get("/items/")
def read_items(
tags: List[str] = Query(
[], # 默认值为空列表
title="Tags", # 标题
description="Tags to filter items" # 描述
)
):
"""处理带多个查询参数值的GET请求"""
return {"tags": tags}路径参数和查询参数可以混合使用,以下是混合使用的示例代码:
from fastapi import FastAPI, Path, Query
from typing import Optional
app = FastAPI()
@app.get("/items/{item_id}/comments/")
def read_item_comments(
item_id: int = Path(
..., # 必需参数
title="Item ID", # 标题
description="The unique identifier of the item", # 描述
ge=1, # 大于等于1
le=1000 # 小于等于1000
),
skip: int = Query(
0, # 默认值
title="Skip", # 标题
description="Number of comments to skip", # 描述
ge=0 # 大于等于0
),
limit: int = Query(
10, # 默认值
title="Limit", # 标题
description="Number of comments to limit", # 描述
ge=1, # 大于等于1
le=100 # 小于等于100
),
q: Optional[str] = Query(
None, # 默认值为None,可选参数
title="Query", # 标题
description="Search query for comments" # 描述
)
):
"""处理带路径参数和查询参数的GET请求"""
if q:
return {"item_id": item_id, "skip": skip, "limit": limit, "q": q}
else:
return {"item_id": item_id, "skip": skip, "limit": limit}路由组是指将多个相关的路由和处理函数组合在一起,使用相同的前缀、依赖注入、中间件等。FastAPI 使用 APIRouter 类来实现路由分组。
以下是路由组的示例代码:
from fastapi import FastAPI, APIRouter
# 创建FastAPI应用程序实例
app = FastAPI()
# 创建路由组
items_router = APIRouter(
prefix="/items", # 路由前缀
tags=["items"], # 标签,用于API文档分组
responses={404: {"description": "Item not found"}}, # 全局响应
)
# 定义路由组的路由
@items_router.get("/")
def read_items():
"""处理路由组根路径的GET请求"""
return {"items": ["Item 1", "Item 2", "Item 3"]}
@items_router.get("/{item_id}")
def read_item(item_id: int):
"""处理路由组带路径参数的GET请求"""
return {"item_id": item_id}
@items_router.post("/")
def create_item(item: dict):
"""处理路由组的POST请求"""
return {"item": item}
@items_router.put("/{item_id}")
def update_item(item_id: int, item: dict):
"""处理路由组带路径参数的PUT请求"""
return {"item_id": item_id, "item": item}
@items_router.delete("/{item_id}")
def delete_item(item_id: int):
"""处理路由组带路径参数的DELETE请求"""
return {"item_id": item_id}
# 将路由组添加到FastAPI应用程序实例中
app.include_router(items_router)
# 创建另一个路由组
users_router = APIRouter(
prefix="/users", # 路由前缀
tags=["users"], # 标签,用于API文档分组
responses={404: {"description": "User not found"}}, # 全局响应
)
# 定义另一个路由组的路由
@users_router.get("/")
def read_users():
"""处理路由组根路径的GET请求"""
return {"users": ["User 1", "User 2", "User 3"]}
@users_router.get("/{user_id}")
def read_user(user_id: int):
"""处理路由组带路径参数的GET请求"""
return {"user_id": user_id}
@users_router.post("/")
def create_user(user: dict):
"""处理路由组的POST请求"""
return {"user": user}
@users_router.put("/{user_id}")
def update_user(user_id: int, user: dict):
"""处理路由组带路径参数的PUT请求"""
return {"user_id": user_id, "user": user}
@users_router.delete("/{user_id}")
def delete_user(user_id: int):
"""处理路由组带路径参数的DELETE请求"""
return {"user_id": user_id}
# 将另一个路由组添加到FastAPI应用程序实例中
app.include_router(users_router)路由组可以嵌套使用,这样可以实现更复杂的 API 结构。以下是嵌套路由的示例代码:
from fastapi import FastAPI, APIRouter
# 创建FastAPI应用程序实例
app = FastAPI()
# 创建父路由组
items_router = APIRouter(
prefix="/items", # 路由前缀
tags=["items"], # 标签,用于API文档分组
responses={404: {"description": "Item not found"}}, # 全局响应
)
# 创建子路由组
comments_router = APIRouter(
prefix="/{item_id}/comments", # 路由前缀
tags=["comments"], # 标签,用于API文档分组
responses={404: {"description": "Comment not found"}}, # 全局响应
)
# 定义父路由组的路由
@items_router.get("/")
def read_items():
"""处理父路由组根路径的GET请求"""
return {"items": ["Item 1", "Item 2", "Item 3"]}
@items_router.get("/{item_id}")
def read_item(item_id: int):
"""处理父路由组带路径参数的GET请求"""
return {"item_id": item_id}
# 定义子路由组的路由
@comments_router.get("/")
def read_item_comments(item_id: int):
"""处理子路由组根路径的GET请求"""
return {"item_id": item_id, "comments": ["Comment 1", "Comment 2"]}
@comments_router.get("/{comment_id}")
def read_item_comment(item_id: int, comment_id: int):
"""处理子路由组带路径参数的GET请求"""
return {"item_id": item_id, "comment_id": comment_id}
@comments_router.post("/")
def create_item_comment(item_id: int, comment: dict):
"""处理子路由组的POST请求"""
return {"item_id": item_id, "comment": comment}
@comments_router.put("/{comment_id}")
def update_item_comment(item_id: int, comment_id: int, comment: dict):
"""处理子路由组带路径参数的PUT请求"""
return {"item_id": item_id, "comment_id": comment_id, "comment": comment}
@comments_router.delete("/{comment_id}")
def delete_item_comment(item_id: int, comment_id: int):
"""处理子路由组带路径参数的DELETE请求"""
return {"item_id": item_id, "comment_id": comment_id}
# 将子路由组添加到父路由组中
items_router.include_router(comments_router)
# 将父路由组添加到FastAPI应用程序实例中
app.include_router(items_router)FastAPI 提供了强大的依赖注入系统,可以在路由级别和处理函数级别使用依赖注入。在路由级别使用依赖注入时,依赖注入会作用于该路由组的所有路由。
以下是依赖注入路由的示例代码:
from fastapi import FastAPI, APIRouter, Depends, HTTPException, status
from typing import Optional
# 创建FastAPI应用程序实例
app = FastAPI()
# 定义依赖注入函数
def get_query_params(q: Optional[str] = None):
"""获取查询参数"""
if q:
return q
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Query parameter q is required"
)
# 创建路由组,并添加依赖注入
items_router = APIRouter(
prefix="/items", # 路由前缀
tags=["items"], # 标签,用于API文档分组
responses={404: {"description": "Item not found"}}, # 全局响应
dependencies=[Depends(get_query_params)] # 路由级别依赖注入
)
# 定义路由组的路由
@items_router.get("/")
def read_items(q: str = Depends(get_query_params)):
"""处理路由组根路径的GET请求"""
return {"items": ["Item 1", "Item 2", "Item 3"], "q": q}
@items_router.get("/{item_id}")
def read_item(item_id: int, q: str = Depends(get_query_params)):
"""处理路由组带路径参数的GET请求"""
return {"item_id": item_id, "q": q}
# 将路由组添加到FastAPI应用程序实例中
app.include_router(items_router)FastAPI 使用 @app.on_event 装饰器定义生命周期事件。生命周期事件包括:
以下是生命周期路由的示例代码:
from fastapi import FastAPI
# 创建FastAPI应用程序实例
app = FastAPI()
# 应用程序启动时触发
@app.on_event("startup")
def startup_event():
"""应用程序启动事件"""
print("Application startup")
# 应用程序关闭时触发
@app.on_event("shutdown")
def shutdown_event():
"""应用程序关闭事件"""
print("Application shutdown")
# 定义路由
@app.get("/")
def read_root():
"""处理根路径的GET请求"""
return {"message": "Hello, World!"}WebHook 是一种 HTTP 回调机制,它允许服务器在特定事件发生时向客户端发送 HTTP 请求。FastAPI 可以使用 APIRouter 来实现 WebHook 路由。
以下是 WebHook 路由的示例代码:
from fastapi import FastAPI, APIRouter, Request
# 创建FastAPI应用程序实例
app = FastAPI()
# 创建WebHook路由组
webhook_router = APIRouter(
prefix="/webhook", # 路由前缀
tags=["webhook"], # 标签,用于API文档分组
)
# 定义WebHook路由
@webhook_router.post("/event")
async def webhook_event(request: Request):
"""处理WebHook事件"""
data = await request.json()
print(f"WebHook event received: {data}")
return {"status": "ok"}
# 将WebHook路由组添加到FastAPI应用程序实例中
app.include_router(webhook_router)FastAPI 可以使用 StaticFiles 类来提供静态文件服务。静态文件包括 HTML 文件、CSS 文件、JavaScript 文件、图片文件等。
以下是静态文件路由的示例代码:
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
# 创建FastAPI应用程序实例
app = FastAPI()
# 挂载静态文件目录
app.mount("/static", StaticFiles(directory="static"), name="static")
# 定义路由
@app.get("/")
def read_root():
"""处理根路径的GET请求"""
return {"message": "Hello, World!"}在上面的示例中,静态文件目录是 static,访问路径是 /static。
项目描述:创建一个简单的待办事项 API,包含创建待办事项、查询待办事项、更新待办事项、删除待办事项等功能。技术栈:FastAPI、Python 3.7+。代码示例:
from fastapi import FastAPI, HTTPException, status
from typing import List, Optional
from pydantic import BaseModel
# 创建FastAPI应用程序实例
app = FastAPI()
# 定义待办事项的数据模型
class Todo(BaseModel):
"""待办事项的数据模型"""
id: Optional[int] = None
title: str
description: Optional[str] = None
completed: Optional[bool] = False
# 模拟数据库
todos = []
# 定义路由
@app.get("/todos/", response_model=List[Todo])
def read_todos():
"""查询所有待办事项"""
return todos
@app.get("/todos/{todo_id}", response_model=Todo)
def read_todo(todo_id: int):
"""查询单个待办事项"""
for todo in todos:
if todo["id"] == todo_id:
return todo
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Todo with id {todo_id} not found"
)
@app.post("/todos/", response_model=Todo, status_code=status.HTTP_201_CREATED)
def create_todo(todo: Todo):
"""创建待办事项"""
new_todo = todo.dict()
if new_todo["id"] is None:
new_todo["id"] = len(todos) + 1
todos.append(new_todo)
return new_todo
@app.put("/todos/{todo_id}", response_model=Todo)
def update_todo(todo_id: int, todo: Todo):
"""更新待办事项"""
for i, existing_todo in enumerate(todos):
if existing_todo["id"] == todo_id:
updated_todo = todo.dict()
updated_todo["id"] = todo_id
todos[i] = updated_todo
return updated_todo
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Todo with id {todo_id} not found"
)
@app.delete("/todos/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_todo(todo_id: int):
"""删除待办事项"""
for i, existing_todo in enumerate(todos):
if existing_todo["id"] == todo_id:
todos.pop(i)
return
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Todo with id {todo_id} not found"
)项目描述:创建一个博客文章 API,包含创建博客文章、查询博客文章、更新博客文章、删除博客文章等功能。技术栈:FastAPI、Python 3.7+。代码示例:
from fastapi import FastAPI, HTTPException, status
from typing import List, Optional
from pydantic import BaseModel
# 创建FastAPI应用程序实例
app = FastAPI()
# 定义博客文章的数据模型
class Post(BaseModel):
"""博客文章的数据模型"""
id: Optional[int] = None
title: str
content: str
author: str
published: Optional[bool] = False
# 模拟数据库
posts = []
# 定义路由
@app.get("/posts/", response_model=List[Post])
def read_posts():
"""查询所有博客文章"""
return posts
@app.get("/posts/{post_id}", response_model=Post)
def read_post(post_id: int):
"""查询单个博客文章"""
for post in posts:
if post["id"] == post_id:
return post
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Post with id {post_id} not found"
)
@app.post("/posts/", response_model=Post, status_code=status.HTTP_201_CREATED)
def create_post(post: Post):
"""创建博客文章"""
new_post = post.dict()
if new_post["id"] is None:
new_post["id"] = len(posts) + 1
posts.append(new_post)
return new_post
@app.put("/posts/{post_id}", response_model=Post)
def update_post(post_id: int, post: Post):
"""更新博客文章"""
for i, existing_post in enumerate(posts):
if existing_post["id"] == post_id:
updated_post = post.dict()
updated_post["id"] = post_id
posts[i] = updated_post
return updated_post
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Post with id {post_id} not found"
)
@app.delete("/posts/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(post_id: int):
"""删除博客文章"""
for i, existing_post in enumerate(posts):
if existing_post["id"] == post_id:
posts.pop(i)
return
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Post with id {post_id} not found"
)项目描述:创建一个电商产品 API,包含创建电商产品、查询电商产品、更新电商产品、删除电商产品等功能。技术栈:FastAPI、Python 3.7+。代码示例:
from fastapi import FastAPI, HTTPException, status
from typing import List, Optional
from pydantic import BaseModel
# 创建FastAPI应用程序实例
app = FastAPI()
# 定义电商产品的数据模型
class Product(BaseModel):
"""电商产品的数据模型"""
id: Optional[int] = None
name: str
description: Optional[str] = None
price: float
stock: int
category: str
# 模拟数据库
products = []
# 定义路由
@app.get("/products/", response_model=List[Product])
def read_products():
"""查询所有电商产品"""
return products
@app.get("/products/{product_id}", response_model=Product)
def read_product(product_id: int):
"""查询单个电商产品"""
for product in products:
if product["id"] == product_id:
return product
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Product with id {product_id} not found"
)
@app.post("/products/", response_model=Product, status_code=status.HTTP_201_CREATED)
def create_product(product: Product):
"""创建电商产品"""
new_product = product.dict()
if new_product["id"] is None:
new_product["id"] = len(products) + 1
products.append(new_product)
return new_product
@app.put("/products/{product_id}", response_model=Product)
def update_product(product_id: int, product: Product):
"""更新电商产品"""
for i, existing_product in enumerate(products):
if existing_product["id"] == product_id:
updated_product = product.dict()
updated_product["id"] = product_id
products[i] = updated_product
return updated_product
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Product with id {product_id} not found"
)
@app.delete("/products/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_product(product_id: int):
"""删除电商产品"""
for i, existing_product in enumerate(products):
if existing_product["id"] == product_id:
products.pop(i)
return
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Product with id {product_id} not found"
)路由冲突问题是指两个或多个路由的路径模式匹配相同的 HTTP 请求。FastAPI 会根据路由的定义顺序来匹配 HTTP 请求,首先匹配到的路由会处理请求,后面的路由会被忽略。
解决方法:
示例代码:
from fastapi import FastAPI
app = FastAPI()
# 更具体的路由定义在前面
@app.get("/items/{item_id}")
def read_item(item_id: int):
"""处理带路径参数的GET请求"""
return {"item_id": item_id}
# 更抽象的路由定义在后面
@app.get("/items/")
def read_items():
"""处理根路径的GET请求"""
return {"items": ["Item 1", "Item 2"]}路径参数解析错误问题是指路径参数的类型注解与实际值不匹配。FastAPI 会自动根据类型注解对路径参数进行解析和验证,如果解析失败,会返回 422 Unprocessable Entity 响应。
解决方法:
示例代码:
from fastapi import FastAPI, Path
app = FastAPI()
@app.get("/items/{item_id}")
def read_item(
item_id: int = Path(
..., # 必需参数
title="Item ID", # 标题
description="The unique identifier of the item", # 描述
ge=1, # 大于等于1
le=1000 # 小于等于1000
)
):
"""处理带路径参数的GET请求,参数有验证"""
return {"item_id": item_id}查询参数验证失败问题是指查询参数的类型注解与实际值不匹配。FastAPI 会自动根据类型注解对查询参数进行解析和验证,如果解析失败,会返回 422 Unprocessable Entity 响应。
解决方法:
示例代码:
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/")
def read_items(
skip: int = Query(
0, # 默认值
title="Skip", # 标题
description="Number of items to skip", # 描述
ge=0 # 大于等于0
),
limit: int = Query(
10, # 默认值
title="Limit", # 标题
description="Number of items to limit", # 描述
ge=1, # 大于等于1
le=100 # 小于等于100
)
):
"""处理带查询参数的GET请求,参数有验证"""
return {"skip": skip, "limit": limit}路由组嵌套问题是指路由组的前缀和路径参数重叠。FastAPI 会自动处理路由组嵌套的情况,但需要注意路由组的前缀和路径参数的匹配顺序。
解决方法:
示例代码:
from fastapi import FastAPI, APIRouter
app = FastAPI()
items_router = APIRouter(
prefix="/items", # 路由前缀
tags=["items"], # 标签,用于API文档分组
)
comments_router = APIRouter(
prefix="/{item_id}/comments", # 路由前缀
tags=["comments"], # 标签,用于API文档分组
)
@items_router.get("/")
def read_items():
"""处理父路由组根路径的GET请求"""
return {"items": ["Item 1", "Item 2"]}
@items_router.get("/{item_id}")
def read_item(item_id: int):
"""处理父路由组带路径参数的GET请求"""
return {"item_id": item_id}
@comments_router.get("/")
def read_item_comments(item_id: int):
"""处理子路由组根路径的GET请求"""
return {"item_id": item_id, "comments": ["Comment 1", "Comment 2"]}
@comments_router.get("/{comment_id}")
def read_item_comment(item_id: int, comment_id: int):
"""处理子路由组带路径参数的GET请求"""
return {"item_id": item_id, "comment_id": comment_id}
items_router.include_router(comments_router)
app.include_router(items_router)FastAPI 路由系统的发展趋势主要包括: