评论([[comments.sum]])
如果要构建应用程序或 Web API ,则很少将所有内容都放在一个文件中。
FastAPI提供了一种方便的工具,可在保持所有灵活性的同时构建应用程序。
如果您使用过 Flask ,那将相当于 Flask 的蓝图。
app
目录包含所有内容。app
目录还有一个app/main.py
文件app/routers/
其中两个
__inint__.py
为空文件,仅代表此为一个 python 模块
假设处理用户相关路由的文件的是子模块 /app/routers/users.py
但仍然希望他是程序的一部分,而不是单独的一个 程序(???include_router()
是 fastapi 提供的一种向路由中添加子路由的方式。其不同于Mount
直接挂载 App ,而是将子 router 中的路由都拆解出来添加到根 router
导入 ApiRouter 并创建路由
from fastapi import APIRouter
router = APIRouter()
# 这里的 tags 应该是为了给 自动生成的交互式文档分类
@router.get("/users/", tags=["users"])
async def read_users():
return [{"username": "Foo"}, {"username": "Bar"}]
@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}
flask 的蓝图在注册时就已经声明好主路由。我跟喜欢此种方式,至少确保了路由的字符串不会打错( flask 路由:
/, me, /<>
)
您可以将其APIRouter
视为“迷你FastAPI
”类。支持所有相同的选项。所有相同的参数,响应,依赖性,标签等。
另一个路由文件 app/routers/items.py
不需要在每个路径操作中都显式地键入/items/
和键入:tags=["items"]
from fastapi import APIRouter, HTTPException
router = APIRouter()
@router.get("/")
async def read_items():
return [{"name": "Item Foo"}, {"name": "item Bar"}]
@router.get("/{item_id}")
async def read_item(item_id: str):
return {"name": "Fake Specific Item", "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 != "foo":
raise HTTPException(status_code=403, detail="You can only update the item: foo")
return {"item_id": item_id, "name": "The Fighters"}
此处的 主路由不用添加暂且不表(访问应该是:
/item/, /item/item_id
. 但是 tags 不添加会自动生成码?。 并不会 还是需要在外部导入此 路由时添加。各有利弊吧。个人喜欢 flask 蓝图
在这里导入和使用类FastAPI
。
这将是应用程序中将所有内容捆绑在一起的主文件。
导入子路由
from fastapi import Depends, FastAPI, Header, HTTPException
from .routers import items, users
app = FastAPI()
# 拒绝每个 没有 x-token的请求
async 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")
app.include_router(users.router)
app.include_router(
items.router,
prefix="/items",
tags=["items"],
dependencies=[Depends(get_token_header)],
responses={404: {"description": "Not found"}},
)
prefix
的值, .前缀不得包含路由最后的 /
tags
该列表将应用于此路由中的所有的路由。responses
,该预定义也将包含在所有路由操作中。responses
,他接收一个字典数据,键值是不同的 HTTP 状态码,内容是不同情况下的返回内容(字典格式)response_model
相同,指向的内容是 Pydantic 模型。来源dependencies
( 依赖,钩子)该列表将被添加到路由器中的所有*路由中,并将针对对它们的每个请求执行(过滤?。我们将items
直接导入子模块**并使用 items.router
,而不是仅导入其变量router
。
这是因为我们router
在子模块中还有另一个变量users
。
如果我们一个接一个地导入,例如:
from .routers.items import router
from .routers.users import router
则会发生覆盖
prefix
( path 路径您也可以.include_router()
将同一路由文件使用不同的前缀多次使用。
例如,在不同前缀下(例如/api/v1
和/api/latest
)使用同一 API 可能会很有用。
您可以定义返回响应后要运行的后台任务。
这对于在请求后需要进行的操作很有用,但是客户端实际上并不需要在接收响应之前等待操作完成。
如:
执行操作后发送的电子邮件通知:
处理数据:
BackgroundTasks
首先,BackgroundTasks
在路由的操作函数中导入并定义一个参数
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_notification(email: str, message=""):
with open("log.txt", mode="w") as email_file:
content = f"notification for {email}: {message}"
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"}
FastAPI将为您创建该类的实例对象 BackgroundTasks
。并将其作为该函数的参数传递至函数。
添加进后台任务:
background_tasks.add_task(FunctionName, [args, ...])
from typing import Optional
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: Optional[str] = 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: str = Depends(get_query)
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}
您可以在FastAPI应用程序中自定义几种元数据配置:主要为自动生成的 api 文档的各种描述的配置,如 版本,标题,自动文档的位置(网址,禁用文档,....
不做多余追述,参考文档
如果需要从 FastApi 提供静态文件的访问:
首先需要安装
pip install aiofiles
from fastapi import FastAPI, staticfiles
# or
# from fastapi.staticfiles import StaticFiles
app = FastAPI()
app.mount('/static', staticfiles.StaticFiles(directory='template'), name="static")
这样就可以通过 /static/FileNmae
访问服务器目录template
下静态文件了. name="static"
给它FastAPI内部使用的名称?
对象:StaticFiles(directory=None, packages=None, check_dir=True)
directory
-表示目录路径的字符串或os.Pathlike。packages
-python 软件包的字符串列表。html
-以 HTML 模式运行。index.html
如果存在这样的文件,则自动为目录加载。check_dir
-确保目录在实例化时存在。默认为True
。对于不匹配的请求,静态文件将以“ 404 未找到”或“ 405 方法不允许”响应。在 HTML 模式下,如果404.html
文件存在,它将显示为 404 响应
fastapi 官方文档也提供 有关于测试的介绍,有需要自己去看呗
相关推荐:
来自系列:fastapi评论([[comments.sum]])
[[item.name]] [[item.email]] [[item.date_time]]
[[itemSon.name]] [[itemSon.email]] [[itemSon.date_time]]
回复 @[[itemSon.reply_user.name]]:
加载更多([[item.son.length-2]])...