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

Question

我正在上一堂发送通知的课。初始化时，需要建立与通知服务器的连接，这非常耗时。我使用 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

中初始化它，但是每次请求到达时它都会被初始化，这需要一些时间。有什么办法可以使用中间件在每次请求到达时重新使用它，这样就不需要每次都初始化。

或 方法二：在 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

Answer 1

选项1

您可以将自定义类对象存储到 app 实例，这允许您使用通用

app.state

属性存储任意额外状态，如 here、here 和 here 所示。要访问

app.state

属性，以及随后在主文件外部的对象（例如，从使用 routers 的 APIRouter 子模块），您可以使用

Request

对象，如这个答案（即使用

request.app.state

）。您可以使用

startup

事件（如此处所示）来初始化对象，但由于它现在已弃用（并且可能在未来版本中删除），因此您可以使用

lifespan

函数 .

示例

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    ''' Run at startup
        Initialise 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

中定义的端点 -使用

request.state

。

示例

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    ''' Run at startup
        Initialise 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
    # ...