4 FastApi... HCuan

项目的文件结构 ¶
- ApiRouter
- 入口文件 main
后台任务 ¶
- 使用 BackgroundTasks
- Takes 和依赖（ Dependency ）一起使用
元数据和自动文档
静态文件
接口测试

项目的文件结构 ¶

如果要构建应用程序或 Web API ，则很少将所有内容都放在一个文件中。

FastAPI提供了一种方便的工具，可在保持所有灵活性的同时构建应用程序。

如果您使用过 Flask ，那将相当于 Flask 的蓝图。

该app目录包含所有内容。
该app目录还有一个app/main.py文件
有一个子目录app/routers/

其中两个 __inint__.py 为空文件，仅代表此为一个 python 模块

ApiRouter

假设处理用户相关路由的文件的是子模块 /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 蓝图

入口文件 main

在这里导入和使用类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"}},
)

items 的导入

添加了一个前缀的所有路径操作使用参数prefix的值, .前缀不得包含路由最后的 /
添加一个列表，tags该列表将应用于此路由中的所有的路由。
添加预定义responses，该预定义也将包含在所有路由操作中。
- 我们可以传递给路径操作装饰器一个参数responses，他接收一个字典数据，键值是不同的 HTTP 状态码，内容是不同情况下的返回内容(字典格式)
- 如果返回内容包含键值model，那么它的作用与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 可能会很有用。

后台任务¶

您可以定义返回响应后要运行的后台任务。

这对于在请求后需要进行的操作很有用，但是客户端实际上并不需要在接收响应之前等待操作完成。

如：

执行操作后发送的电子邮件通知：
- 由于连接到电子邮件服务器并发送电子邮件的过程通常很慢（几秒钟），因此您可以立即返回响应并在后台发送电子邮件通知。
处理数据：
- 例如，假设您收到的文件必须经过缓慢的处理，您可以返回“已接受”（ HTTP 202 ）响应，并在后台对其进行处理。

使用 `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, ...])

Takes 和依赖（ Dependency ）一起使用

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 官方文档也提供有关于测试的介绍，有需要自己去看呗