更大的应用 - 多个文件
假设文件结构如下:
.
├── app # 「app」是一个 Python 包
│ ├── __init__.py # 这个文件使「app」成为一个 Python 包
│ ├── main.py # 「main」模块,例如 import app.main
│ ├── dependencies.py # 「dependencies」模块,例如 import app.dependencies
│ └── routers # 「routers」是一个「Python 子包」
│ │ ├── __init__.py # 使「routers」成为一个「Python 子包」
│ │ ├── items.py # 「items」子模块,例如 import app.routers.items
│ │ └── users.py # 「users」子模块,例如 import app.routers.users
│ └── internal # 「internal」是一个「Python 子包」
│ ├── __init__.py # 使「internal」成为一个「Python 子包」
│ └── admin.py # 「admin」子模块,例如 import app.internal.admin
APIRouter
假设专门用于处理用户逻辑的文件是位于 /app/routers/users.py 的子模块
你可以使用 APIRouter 为该模块创建路径操作。
然后你可以使用它来声明路径操作。
from fastapi import APIRouterrouter = APIRouter()@router.get("/users")
async def read_users():return [{"username": "Rick"}, {"username": "Morty"}]@router.get("/users/me", tags=["users"])
async def read_user_me():return {"username": "fakecurrentuser"}@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):return {"username": username}
依赖项
我们将需要一些在应用程序的好几个地方所使用的依赖项。因此,我们将它们放在它们自己的 dependencies 模块(app/dependencies.py)中
from fastapi import Header, HTTPExceptionasync def get_token_header(x_token: str = Header()):if x_token != "fake-super-secret-token":raise HTTPException(status_code=400, detail="X-Token header invalid")async def get_query_token(token: str):if token != "jessica":raise HTTPException(status_code=400, detail="No Jessica token provided")
其他使用 APIRouter 的模块
假设你在位于 app/routers/items.py 的模块中还有专门用于处理应用程序中「item」的端点。
我们知道此模块中的所有路径操作都有相同的:
- 路径 prefix:/items。
- tags:(仅有一个 items 标签)。
- 额外的 responses。
- dependencies:它们都需要我们创建的 X-Token 依赖项。
因此,我们可以将其添加到 APIRouter 中,而不是将其添加到每个路径操作中
我们还可以添加一个 tags 列表和额外的 responses 列表,这些参数将应用于此路由器中包含的所有路径操作。
我们可以添加一个 dependencies 列表,这些依赖项将被添加到路由器中的所有路径操作中,并将针对向它们发起的每个请求执行/解决。
我们仍然可以添加更多将会应用于特定的路径操作的 tags,以及一些特定于该路径操作的额外 responses
from fastapi import APIRouter, Depends, HTTPExceptionfrom ..dependencies import get_token_headerrouter = APIRouter(prefix="/items",tags=["items"],dependencies=[Depends(get_token_header)],responses={404: {"description": "Not found"}}, # 当 API 端点可能返回 404 错误时,在文档中显示 "Not found" 的描述
)fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}@router.get("/")
async def read_items():return fake_items_db@router.get("/{item_id}")
async def read_item(item_id: str):if item_id not in fake_items_db:raise HTTPException(status_code=404, detail="Item not found")return {"name": fake_items_db[item_id]["name"], "item_id": item_id}@router.put("/{item_id}", tags=["custom"], responses={403: {"description": "Operation forbidden"}})
async def update_item(item_id: str):if item_id != "plumbus":raise HTTPException(status_code=403, detail="You can only update the plumbus")return {"item_id": item_id, "name": "The Great Plumbus"}
FastAPI 主体
from fastapi import Depends, FastAPIfrom .dependencies import get_query_token, get_token_header
from .routers import items, users
from .internal import adminapp = FastAPI(dependencies=[Depends(get_query_token)]) # 声明全局依赖# 加入路由
app.include_router(users.router)
app.include_router(items.router)
app.include_router(admin.router,prefix="/admin",tags=["admin"],dependencies=[Depends(get_token_header)],responses={418: {"description": "I'm a teapot"}}, # 示例响应
)@app.get("/")
async def root():return {"message": "Hello World"}
假设你的组织为你提供了 app/internal/admin.py 文件。
它包含一个带有一些由你的组织在多个项目之间共享的管理员路径操作的 APIRouter。
对于此示例,它将非常简单。但是假设由于它是与组织中的其他项目所共享的,因此我们无法对其进行修改,以及直接在 APIRouter 中添加 prefix、dependencies、tags 等
可是我们仍然希望在包含 APIRouter 时设置一个自定义的 prefix,以便其所有路径操作以 /admin 开头,我们希望使用本项目已经有的 dependencies 保护它,并且我们希望它包含自定义的 tags 和 responses。
我们可以通过将这些参数传递给 app.include_router() 来完成所有的声明,而不必修改原始的 APIRouter
查看自动化的 API 文档
现在,使用 app.main 模块和 app 变量运行 uvicorn
后台任务
你可以定义在返回响应后运行的后台任务。
这对需要在请求之后执行的操作很有用,但客户端不必在接收响应之前等待操作完成。
使用 BackgroundTasks
创建要作为后台任务运行的函数。
它只是一个可以接收参数的标准函数。
它可以是 async def 或普通的 def 函数,FastAPI 知道如何正确处理。
然后在你的 路径操作函数 里,用 .add_task() 方法将任务函数传到 后台任务 对象中
from email import message
from fastapi import BackgroundTasks, FastAPIapp = FastAPI()def write_notification(email: str, message = ""):with open("log.txt", mode="w") as email_file:content = f"notification for {email} with message: {message}\n" email_file.write(content)@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):background_tasks.add_task(write_notification,email, message="some notification")return {"message": "Notification sent in the background"}
依赖注入
使用 BackgroundTasks 也适用于依赖注入系统,你可以在多个级别声明 BackgroundTasks 类型的参数:在 路径操作函数 里,在依赖中(可依赖),在子依赖中,等等。
from tkinter import NO
from typing import Annotatedfrom fastapi import BackgroundTasks, Depends, FastAPIapp = FastAPI()def write_log(message: str):with open("log.txt", mode="a") as log:log.write(message)def get_query(background_tasks: BackgroundTasks, q: str | None = None):if q:message = f"found query: {q}\n"background_tasks.add_task(write_log, message)return q@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):message = f"message to {email}\n"background_tasks.add_task(write_log, message)return {"message": "Message sent"}
运行程序,此时设置q为3
元数据和文档 URL
API 元数据
你可以在设置 OpenAPI 规范和自动 API 文档 UI 中使用的以下字段:
from fastapi import FastAPIdescription = """
ChimichangApp API helps you do awesome stuff. 🚀## ItemsYou can **read items**.## UsersYou will be able to:* **Create users** (_not implemented_).
* **Read users** (_not implemented_).
"""app = FastAPI(title="ChimichangApp API",description=description,summary="Deadpool's favorite app. Nuff said.",version="0.0.1",terms_of_service="http://example.com/terms/",contact={"name": "Deadpoolio the Amazing","url": "http://x-force.example.com/contact/","email": "dp@x-force.example.com",},license_info={"name": "Apache 2.0","url": "https://www.apache.org/licenses/LICENSE-2.0.html",},
)@app.get("/items/")
async def read_items():return [{"name": "Katana"}]
运行程序,查看API文档
标签元数据
创建标签元数据并把它传递给 openapi_tags 参数,将 tags 参数和路径操作(以及 APIRouter)一起使用,将其分配给不同的标签:
from fastapi import FastAPItags_metadata = [{"name": "users","description": "Operations with users. The **login** logic is also here.",},{"name": "items","description": "Manage items. So _fancy_ they have their own docs.","externalDocs": {"description": "Items external docs","url": "https://fastapi.tiangolo.com/",},},
]app = FastAPI(openapi_tags=tags_metadata)@app.get("/users/", tags=["users"])
async def get_users():return [{"name": "Harry"}, {"name": "Ron"}]@app.get("/items/", tags=["items"])
async def get_items():return [{"name": "wand"}, {"name": "flying broom"}]
运行程序,打开docs文档
每个标签元数据字典的顺序也定义了在文档用户界面显示的顺序。
例如按照字母顺序,即使 users 排在 items 之后,它也会显示在前面,因为我们将它的元数据添加为列表内的第一个字典。
文档 URLs
你可以配置两个文档用户界面,包括:
- Swagger UI:服务于 /docs。
- 可以使用参数 docs_url 设置它的 URL。
- 可以通过设置 docs_url=None 禁用它。
- ReDoc:服务于 /redoc。
- 可以使用参数 redoc_url 设置它的 URL。
- 可以通过设置 redoc_url=None 禁用它。
from fastapi import FastAPIapp = FastAPI(docs_url="/documentation", redoc_url=None)@app.get("/items/")
async def read_items():return [{"name": "Foo"}]