FastAPI 中定义路由的方式主要有两种,一种是使用 FastAPI 应用实例的方法(例如 app.get()),一种是使用装饰器(例如 @app.get()),它们的用法和作用略有不同。
方式1:使用 FastAPI 应用实例的方法
app.get()、app.post()等方法用于直接在应用实例上定义路由。
这些方法接受路径字符串和处理函数作为参数,将指定的路径映射到相应的处理函数,用于处理该路径上的请求。
这种方式适用于在全局范围内定义路由,将路由直接添加到应用程序中。
from fastapi import FastAPI def read_root(): return {"message": "Hello, World"} app = FastAPI() app.get("/", read_root)
方式2:使用装饰器
@app.get()、@app.post()等装饰器语法,用于在路由处理函数上方添加装饰器,将处理函数与特定的路径和 HTTP 方法关联起来。
装饰器方式允许直接在路由处理函数上使用装饰器来定义路径、HTTP 方法、响应模型等属性。
这种方式更直观,将路由相关的信息集中在处理函数上,使代码更易读和维护。适用于局部范围内定义路由。
from fastapi import FastAPI app = FastAPI() @app.get("/") def read_root(): return {"message": "Hello, World"}
拓展:匹配所有路径的方式
@app.get("/files/{file_path:path}") # :path 代表任意路径 async def read_file(file_path: str): return {"file_path": file_path}
FastAPI 允许将路由组织成分组,以更好地组织和管理 API 端点。这对于具有多个相关路由的应用程序特别有用。分组可以保持代码的清晰性和可维护性。
路由分组有助于保持代码的整洁和可读性,还可以在文档中更好地组织和呈现 API 端点。
可以根据需求创建多个分组,在分组中将相关的路由放在一起,同时还可以将路由分组放到不同的模块,比如可以创建一个items.py文件,其中包含所有与 items 相关的路由,然后,在主函数模块 main.py 中导入并将分组添加到应用程序。
APIRouter 模块的方法:
FastAPI 模块的方法:
示例
from fastapi import FastAPI, APIRouter app = FastAPI() # 创建一个路由分组 router = APIRouter() # 将路由添加到分组 @router.get("/") def read_root(): return {"message": "Root"} @router.get("/items/") def read_items(): return {"message": "Items"} # 将分组添加到应用程序 app.include_router(router, prefix="/group1", tags=["Group 1"]) # 创建另一个路由分组 router2 = APIRouter() @router2.get("/") def read_another_root(): return {"message": "Another Root"} @router2.get("/things/") def read_things(): return {"message": "Things"} # 将第二个分组添加到应用程序 app.include_router(router2, prefix="/group2", tags=["Group 2"])
路径参数:FastAPI 通过在路径参数两边添加{} 声明动态路径参数
@app.get("/items/{item_id}") def read_item(item_id: int): return {"item_id": item_id}
路径参数 item_id 的值将作为参数 item_id 传递给函数。
包含路径参数的路由必须声明在路径前缀相同的普通路径之后!
举个例子,假设现有两个具有相同路径前缀的路径:
由于路径操作是按顺序依次运行的,所以需要确保路径 /users/me 声明在路径 /users/{user_id} 之前!
否则,/users/{user_id} 的路径还将与 /users/me 相匹配,"认为"自己正在接收一个值为 “me” 的 user_id 参数。
from fastapi import FastAPI app = FastAPI() @app.get("/users/me") async def read_user_me(): return {"user_id": "the current user"} @app.get("/users/{user_id}") async def read_user(user_id: str): return {"user_id": user_id}
预设值
有时只需要给路径参数传递几个常用并且固定的有效值,那么就可以通过枚举来定制预设值。
注:枚举(enums)从 3.4 版本起在 Python 中可用。
from enum import Enum # 创建一个 Enum 类 class ModelName(str, Enum): yinyu = "yinyu_v" s1 = "s1_v" s2 = "s2_v" # 包含枚举的路径参数 ## 路径参数 model_name 的值将传递给函数 get_model 的参数 model_name,并且这个值的取值范围只能是 ModelName 枚举类中类属性的值。 @app.get("/models/{model_name}") async def get_model(model_name: ModelName): # 第 1 种判断方式 if model_name is ModelName.yinyu: return {"model_name": model_name, "message": "yinyu get"} # 第 2 种判断方式,效果一样 if model_name.value == "s1_name": return {"model_name": model_name, "message": "s1 get"} else: return {"model_name": ModelName.s2, "message": "s2 get"}
在 FastAPI 中,fastapi.Path 是一个用于声明路径参数的类,它提供了更多的参数配置选项,允许定义路径参数的类型、默认值、校验规则等。使用 fastapi.Path 可以更精细地控制路径参数的行为。
常用可选传参数:
default:指定路径参数的默认值。
注意:因路径参数总是必需的,若想传参 default,只能设置 default=... ,表示显式的将其标记为必需参数。
title:在自动生成的 API 文档中,用于显示路径参数的标题,提供更好的文档说明。
description:在自动生成的 API 文档中,用于显示路径参数的详细描述,提供更详细的文档说明。
min_length 和 max_length:用于限制路径参数的字符串长度范围。
min 和 max:用于限制路径参数的数值范围,适用于数值类型的路径参数。
regex:使用正则表达式校验路径参数的值,可以使用字符串形式的正则表达式。
deprecated:将路径参数标记为已弃用,用于指示用户不应再使用该参数。
example:在自动生成的 API 文档中,用于为路径参数提供示例值,帮助用户理解如何使用该参数。
gt、ge、lt、le:用于设置数值类型路径参数的大小比较,分别表示大于、大于等于、小于、小于等于。
const:将路径参数设置为一个常量值,这可以用于实现一些特定的路由匹配需求。
查询参数
在 Web 开发中,查询参数(Query Parameters)是通过 URL 中的参数键值对来传递额外的信息给服务器的一种方式。这些参数通常用于过滤、排序、分页等操作。
查询参数通常出现在 URL 的问号(?)后面,每个查询参数由参数名和参数值组成,用等号(=)连接。多个查询参数之间使用与号(&)分隔。
FastAPI 声明查询参数:
在路由函数的参数列表声明不属于路径参数的其他函数参数,将被自动解释为查询参数!
查询参数可以设置默认值:
FastAPI 可以将参数直接添加到路由函数的参数列表来获取查询参数(可以设置默认值,若不设置默认值的话,未传参会报错)
@app.get("/items/{item_id}") def read_item(item_id: int, q: str = None): # q是一个查询参数,可以接受字符串类型的值 return {"item_id": item_id, "q": q} # 调用:localhost:8000/items/2?q=hhh
bool 类型转换
声明为 bool 类型的查询参数,可以将传参 1、True 、true、on、yes 自动类型转换为 True
typing.Union :是 Python 中的一个类型提示工具,用于表示多个类型的联合。它通常与 Union 类型进行配合使用。
在类型注释中,可以使用 typing.Union 来明确指定一个变量的类型可以是多个类型中的一个
fastapi.Query :是 FastAPI 框架中用于处理查询参数的一个工具,并允许指定各种参数来控制查询参数的类型、默认值、验证等。以下是一些常用的可选传参数:
default:指定查询参数的默认值。如果请求中没有提供该查询参数,将使用默认值。可以位置传参。
声明为必传传参数:
方式1:不指定默认值即为必传参数
方式2:指定 default=...
方式3:指定 default=Required
注:需导包 from pydantic import Required
title:用于自动生成的 API 文档中的标题,使文档更加清晰易懂。
description:在自动生成的 API 文档中显示有关查询参数的详细说明。
alias:指定查询参数的别名。当希望在函数中使用不同的名称来引用查询参数时,可以使用这个参数。
regex:使用正则表达式对查询参数进行验证。只有匹配正则表达式的参数值才会被接受。
min_length 和 max_length:限制查询参数值的最小和最大长度。
min 和 max:限制查询参数值的最小和最大值(仅适用于数字类型的参数)。
gt、ge、lt、le:分别用于指定查询参数的大于、大于等于、小于、小于等于的值(仅适用于数字类型的参数)。
multiple:指定查询参数是否可以有多个值,即是否允许使用多个相同名称的查询参数。
一般用于 list 类型的查询参数
deprecated:指定查询参数是否已弃用。
如果设置为 True,在自动生成的 API 文档中将显示该参数已被弃用的信息。
代码示例:
from typing import Union from fastapi import FastAPI,Query @app.get("/items21/") async def read_items(q: Union[str, None] = Query(default=None), a: str = Query(default=None, max_length=50, min_length=3) b: Union[str, None] = Query(default=None, regex="^fixedquery$") ): query_items = {"q": q} return query_items
请求体
使用 Pydantic 库定义数据模型,可以轻松处理请求体。
from pydantic import BaseModel # 定义Item模型 class Item(BaseModel): name: str description: str = None price: float tax: float = None @app.post("/items/") def create_item(item: Item): # 使用Item模型定义了请求体的结构,并将其用作create_item函数的参数 return item
响应模型
通过为路由设置 response_model 参数,可以定义响应数据的结构。
from pydantic import BaseModel class Item(BaseModel): name: str description: str = None price: float tax: float = None @app.post("/items/", response_model=Item) def create_item(item: Item): return item
FastAPI 支持依赖注入,可以方便地管理和复用代码。
在 FastAPI 中,依赖注入是一种强大的功能,它允许将功能(如数据库连接、验证逻辑等)注入到路由处理函数中,以便更好地组织和管理代码,并实现解耦和可测试性。
依赖注入使得能够在需要时将依赖项提供给函数,而不是在函数内部创建这些依赖项。
依赖注入的优势在于它可以将复杂的逻辑从路由处理函数中分离出来,使代码更加清晰和可维护。
from fastapi import FastAPI, Depends app = FastAPI() # 依赖项,模拟数据库连接 def get_db_connection(): db_connection = "fake_db_connection" return db_connection # 路由处理函数,注入依赖项 @app.get("/items/") async def read_items(db: str = Depends(get_db_connection)): return {"message": "Items retrieved", "db_connection": db}
通过自定义异常处理程序来处理应用程序中的错误
from fastapi import HTTPException @app.get("/items/{item_id}") def read_item(item_id: int): if item_id < 1: raise HTTPException(status_code=400, detail="Item not found") return {"item_id": item_id}
使用 fastapi.UploadFile 类型来接收上传的文件数据
from fastapi import FastAPI, UploadFile from fastapi.staticfiles import StaticFiles import os app = FastAPI() # 配置静态文件路径 app.mount("/static", StaticFiles(directory="static"), name="static") @app.post("/file_upload") async def file_upload(file: UploadFile): # 接收文件 res = await file.read() # 写到本地 predict_img_path = os.path.join('static', file.filename) with open(predict_img_path, "wb") as f: f.write(res) return {"code": 1, "msg": "上传成功:{}".format(file.filename)} if __name__ == '__main__': import uvicorn uvicorn.run(app="main:app", host="0.0.0.0", port=2333)
拓展:上传文件
方式1:
import requests import json if __name__ == '__main__': # 上传一张图 file_path = r'E:\工具测试数据\封面上传测试\img\1825513ec0b.png' url = "http://127.0.0.1:2333/file_upload" data = {"file": open(file_path, 'rb')} res = requests.post(url=url, files=data, verify=False) print(json.loads(res.content)) # {'code': 1, 'msg': '上传成功:7f63f6711f5f57d9.jpg'}
方式2:
import requests import json import os from urllib3 import encode_multipart_formdata if __name__ == '__main__': # 上传一张图 file_path = r'E:\工具测试数据\封面上传测试\img\1825513ec0b.png' url = "http://127.0.0.1:2333/file_upload" with open(file_path, 'rb') as f: file_name = os.path.split(file_path)[-1] file = {"file": (file_name, f.read())} encode_data = encode_multipart_formdata(file) data = encode_data[0] content_type = encode_data[1] headers = { 'Content-Type': content_type } res = requests.post(url, headers=headers, data=data) print(json.loads(res.content)) # {'code': 1, 'msg': '上传成功:1825513ec0b.png'}
request 对象
FastAPI 可以在路由处理函数中使用 fastapi.Request 模型来获取请求的详细信息,如头部、查询参数等。
from fastapi import FastAPI, Request @app.get("/user-agent/") def read_user_agent(request: Request): # 获取请求头 headers = request.headers user_agent = headers.get("user-agent") return {"user_agent": user_agent}
请求头部中的信息
FastAPI 可以在路由处理函数中使用 fastapi.Header 模型来获取请求头部中的信息,例如 user_agent 等
from fastapi import FastAPI, Header @app.get("/headers/") def read_headers(user_agent: str = Header(None)): return {"User-Agent": user_agent}
Swagger UI 提供的 api 文档
ip:端口/docs(默认 127.0.0.1:8000/docs)
ReDoc 提供的 api 文档
ip:端口/docs(默认 127.0.0.1:8000/redoc)
my_project/ ├── app/ │ ├── apis/ │ │ ├── __init__.py │ │ ├── auth.py │ │ ├── items.py │ │ └── users.py │ ├── core/ │ │ ├── __init__.py │ │ ├── config.py │ │ ├── security.py │ │ └── sqlalchemy.py │ ├── db/ │ │ ├── __init__.py │ │ └── base.py │ ├── main.py │ └── models/ │ ├── __init__.py │ ├── item.py │ └── user.py ├── docker-compose.yml ├── Dockerfile ├── README.md └── requirements.txt
此外,还可以根据项目需要添加其他目录和文件,例如静态文件目录、测试目录、文档目录等。但是,无论怎样组织FastAPI项目结构,都需要保证代码清晰明了、易于维护。