英文:
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.py
的add_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
属性以及随后的对象(例如,从使用APIRouter
的routers
子模块中),你可以使用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
处理程序以来,它与startup
和shutdown
事件类似,允许定义需要在应用启动之前或应用关闭时运行的代码,因此也可以定义对象以便从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
# ...
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库,让每个人都能够通过互相帮助和分享经验来进步。
评论