做商业广告有什么网站好推销的,谷歌外贸建站多少钱,做网站需要的企业,商家联盟会员管理系统更大的应用 - 多个文件
假设文件结构如下#xff1a;.
├── app # 「app」是一个 Python 包
│ ├── __init__.py # 这个文件使「app」成为一个 Python 包
│ ├── main.py # 「main」模块#xff0c;例如 import app.main
│ ├…更大的应用 - 多个文件
假设文件结构如下.
├── 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.adminAPIRouter
假设专门用于处理用户逻辑的文件是位于 /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_code400, detailX-Token header invalid)async def get_query_token(token: str):if token ! jessica:raise HTTPException(status_code400, detailNo 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_dbrouter.get(/{item_id})
async def read_item(item_id: str):if item_id not in fake_items_db:raise HTTPException(status_code404, detailItem 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_code403, detailYou 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: Im 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, modew) as email_file:content fnotification 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, messagesome 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, modea) as log:log.write(message)def get_query(background_tasks: BackgroundTasks, q: str | None None):if q:message ffound query: {q}\nbackground_tasks.add_task(write_log, message)return qapp.post(/send-notification/{email})
async def send_notification(email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):message fmessage to {email}\nbackground_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(titleChimichangApp API,descriptiondescription,summaryDeadpools favorite app. Nuff said.,version0.0.1,terms_of_servicehttp://example.com/terms/,contact{name: Deadpoolio the Amazing,url: http://x-force.example.com/contact/,email: dpx-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_tagstags_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_urlNone 禁用它。
ReDoc服务于 /redoc。
可以使用参数 redoc_url 设置它的 URL。可以通过设置 redoc_urlNone 禁用它。from fastapi import FastAPIapp FastAPI(docs_url/documentation, redoc_urlNone)app.get(/items/)
async def read_items():return [{name: Foo}]
