如何初始化全局对象或变量并在每个 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.pyunder中初始化它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

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

例子

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类似），它允许定义在应用程序启动之前或应用程序关闭时需要运行的代码，因此还可以定义可从. 根据Starlette 的文档：startupshutdownrequest.state

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

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

因此，当在生命周期处理程序中实例化类对象时，您可以将其添加到字典（即 ）中state，并使用 来在端点内访问它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
    # ...