如何初始化全局对象或变量并在每个 FastAPI 端点中重用它？

问题描述：

我有一个用于发送通知的类。初始化时，它涉及与通知服务器建立连接，这很耗时。我使用 FastAPI 中的后台任务来发送通知，因为我不想由于通知而延迟响应。以下是示例代码。

file1.py:
noticlient = NotificationClient()

@app.post("/{data}")
def send_msg(somemsg: str, background_tasks: BackgroundTasks):
    result = add_some_tasks(data, background_tasks, noticlient)
    return result

file2.py:
def add_some_tasks(data, background_tasks: BackgroundTasks, noticlient):
    background_tasks.add_task(noticlient.send, param1, param2)
    result = some_operation
    return result

file2.py这里，通知客户端是全局声明的。我可以在下初始化它add_some_tasks，但每次收到请求时它都会被初始化，这需要一些时间。有没有办法使用中间件在每次收到请求时重新使用它，这样就不需要每次都初始化它了。

或方法二：在 class def 中初始化通知

file1.py:
class childFastApi(FastAPI):
    noticlient = NotificationClient()

app = childFastApi()

@app.post("/{data}")
def send_msg(somemsg: str, background_tasks: BackgroundTasks):
    result = add_some_tasks(data, background_tasks, app.noticlient)
    return result

解决方案 1：

选项 1

您可以将自定义类对象存储到应用程序实例中，这样您就可以使用通用属性存储任意额外状态，如此处以及此处和此处app.state所示。要访问主文件外部的属性，然后访问对象（例如，从使用 的子模块），您可以使用对象，如此答案中所示（即使用）。您可以使用事件（如此处所示）来初始化对象，但由于它现在已被弃用（并且可能会在未来版本中删除），因此您可以改用函数。app.state`routersAPIRouterRequestrequest.app.statestartup`lifespan

例子

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    ''' Run at startup
        Initialize the Client and add it to app.state
    '''
    app.state.n_client = NotificationClient()
    yield
    ''' Run on shutdown
        Close the connection
        Clear variables and release the resources
    '''
    app.state.n_client.close()


app = FastAPI(lifespan=lifespan)


@app.get('/')
async def main(request: Request):
    n_client = request.app.state.n_client
    # ...

选项 2

自从引入了 Starlette 的lifespan处理程序以来，它与startup事件shutdown处理程序类似，允许定义在应用程序启动前或应用程序关闭时需要运行的代码，还可以定义可从 访问的对象request.state。根据Starlette 的文档：

有的lifespan概念state，它是一个字典，可以用来在生命周期和请求之间共享对象。

state请求上接收到的是生命周期处理程序上接收到的状态的浅表副本。

因此，在生命周期处理程序中实例化类对象后，您可以将其添加到字典（即state），并使用在端点内访问它 - 即使是在APIRouter主应用程序文件之外的 s中定义的端点request.state。

例子

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    ''' Run at startup
        Initialize the Client and add it to request.state
    '''
    n_client = NotificationClient()
    yield {'n_client': n_client}
    ''' Run on shutdown
        Close the connection
        Clear variables and release the resources
    '''
    n_client.close()


app = FastAPI(lifespan=lifespan)


@app.get('/')
async def main(request: Request):
    n_client = request.state.n_client
    # ...