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

huangapple go评论101阅读模式
英文:

How to initialise a global object or variable and reuse it in every FastAPI endpoint?

问题

我正在编写一个用于发送通知的类。在初始化时,需要连接到通知服务器,这是耗时的。我在FastAPI中使用后台任务来发送通知,因为我不想因通知而延迟响应。以下是示例代码。

文件1.py:
noticlient = NotificationClient()

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

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

这里,通知客户端在全局声明。我可以在文件2.pyadd_some_tasks下初始化它,但这样会使每次请求到达时都初始化,而这需要一些时间。有没有办法使用中间件,在每次请求到达时重用它,这样就不需要每次都初始化。


方法二:在类定义中初始化通知

文件1.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
英文:

I am having a class to send notifications. When being initialised, it involves making a connection to a notification server, which is time-consuming. I use a background task in FastAPI to send notifications, as I don't want to delay the response due to notification. Below is the sample code.

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

Here, notification client is declared globally. I could have it initialised in file2.py under add_some_tasks, but it would get initialised every time a request arrives, and that would require some time. Is there any way to use a middleware to re-use it every time a request arrives, so that it doesn' t need to be initialised every time.

or
Approach two: Initialize notification in 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

得分: 7

选项 1

你可以将自定义类对象存储到应用实例中,这允许你使用通用的app.state属性存储任意额外的状态,如此示例所示2,以及这里这里。要访问app.state属性以及随后的对象(例如,从使用APIRouterrouters子模块中),你可以使用Request对象,如这个答案所示(即使用request.app.state)。你可以使用startup事件(如此处所示)来初始化对象,但由于它现在已被弃用(并可能在将来版本中被移除),你可以改用lifespan函数

示例

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    ''' 在启动时运行
        初始化客户端并将其添加到 app.state 中
    '''
    app.state.n_client = NotificationClient()
    yield
    ''' 在关闭时运行
        关闭连接
        清除变量并释放资源
    '''
    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处理程序以来,它与startupshutdown事件类似,允许定义需要在应用启动之前或应用关闭时运行的代码,因此也可以定义对象以便从request.state中访问。根据Starlette的文档

> lifespan 具有 state 的概念,它是一个字典,可以用于在生命周期和请求之间共享对象。
>
> 请求接收到的 state 是在生命周期处理程序接收到的状态的浅拷贝。

因此,在生命周期处理程序中实例化类对象时,你可以将其添加到字典中(即state),然后在端点中使用request.state访问它。

示例

from fastapi import FastAPI, Request
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    ''' 在启动时运行
        初始化客户端并将其添加到 request.state 中
    '''
    n_client = NotificationClient()
    yield {'n_client': n_client}
    ''' 在关闭时运行
        关闭连接
        清除变量并释放资源
    '''
    n_client.close()


app = FastAPI(lifespan=lifespan)


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

Option 1

You could store the custom class object to the app instance, which allows you to store arbitrary extra state using the generic the app.state attribute, as demonstrated here, as well as here and here. To access the app.state attribute, and subsequently the object, outside the main file (for instance, from a routers submodule that uses APIRouter), you could use the Request object, as demonstrated in this answer (i.e., using request.app.state). You could either use a startup event (as shown here) to initialise the object, but since it is now deprecated (and might be removed in future versions), you could instead use a lifespan function.

Example

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
    # ...

Option 2

Since the introduction of Starlette's lifespan handler, which, similar to startup and shutdown events, allows one to define code that needs to run before the application starts up, or when the application is shutting down, one could also define objects to be accesible from the request.state. As per Starlette's documentation:

> The lifespan has the concept of state, which is a dictionary that
> can be used to share the objects between the lifespan, and the
> requests.
>
> The state received on the requests is a shallow copy of the state
> received on the lifespan handler.

Hence, when instantiating the class object in the lifespan handler, you could then add it to the dictionary (i.e., the state), and access it within an endpoint using request.state.

Example

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
    # ...

huangapple
  • 本文由 发表于 2023年5月24日 18:20:42
  • 转载请务必保留本文链接:https://go.coder-hub.com/76322463.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定