python - 在 FastAPI 中仅初始化一次繁重服务的最佳方法

Question

我开始工作的 FastAPI 应用程序使用了几个服务，我只想在应用程序启动时初始化一次，然后在不同的地方使用这个对象的方法。
它可以是云服务或任何其他重级别。
可能的方法是使用 Lazy loading并与 Singlenton pattern ，但我正在为 FastAPI 寻找更好的方法。
另一种可能的方法是使用 Depends类并缓存它，但它的使用仅对路由方法有意义，而不对从路由方法调用的其他常规方法有意义。
例子:

async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}  
    

async def non_route_function(commons: dict = Depends(common_parameters)):
    print(commons)         # returns `Depends(common_parameters)` 
    

@router.get('/test')
async def test_endpoint(commons: dict = Depends(common_parameters)):
    print(commons)         # returns correct dict
    await non_route_function()
    return {'success': True}

也可以用 @app.on_event("startup")事件在那里初始化重类，但不知道如何在不使用 singleton 的情况下从每个地方访问这个初始化的对象.
另一种丑陋的方法是将初始化的对象保存到 @app( 然后从请求中获取这个应用程序，但是你必须将 request 传递给每个非路由函数。
我描述的所有方法要么是丑陋的、不方便的、非 Python 的，要么是更糟糕的实践，我们这里也没有像在 flask 中那样的线程局部变量和代理对象，那么我描述的这类问题的最佳方法是什么以上？
谢谢!

Answer 1

在启动 FastAPI 应用程序之前初始化重对象通常是个好主意。这样，当应用程序开始监听连接(并且由负载平衡器提供)时，您就完成了初始化。
您可以在设置应用程序和主路由器的同一位置设置这些依赖项并进行任何初始化，因为它们也是应用程序的一部分。我通常通过一个轻量级服务暴露重对象，该服务将有用的端点暴露给 Controller 本身，然后通过 Depends 注入(inject)服务对象。 .
您希望如何执行初始化取决于您在应用程序中的其他要求 - 例如，如果您计划在 cli 工具中重用基础设施或在 cron 中使用它们。
这是我在几个项目中一直采用的方式，到目前为止效果很好，并将代码更改保留在他们自己的附近。
模拟重型类 heavylifting/heavy.py与 from .heavy import HeavyLifter在 __init__.py :

import time

class HeavyLifter:
    def __init__(self, initial):
        self.initial = initial
        time.sleep(self.initial)
    
    def do_stuff(self):
        return 'we did stuff'

在名为 foo 的模块中创建的骨架项目( heavylifting 现在位于 foo/heavylifting 下，以便理解下面的导入):
foo/app.py

from fastapi import FastAPI, APIRouter
from .heavylifting import HeavyLifter

heavy = HeavyLifter(initial=3)

from .views import api_router

app = FastAPI()
app.include_router(api_router)

foo/services.py
应用中的服务层；服务是应用程序向 Controller 公开的操作和服务，用于处理业务逻辑和其他相关事件。如果一个服务需要重访问，它会添加一个 Depends对该服务的要求。

class HeavyService:
    def __init__(self, heavy):
        self.heavy = heavy
        
    def operation_that_requires_heavy(self):
        return self.heavy.do_stuff()
        
class OtherService:
    def __init__(self, heavy_service: HeavyService):
        self.heavy_service = heavy_service
        
    def other_operation(self):
        return self.heavy_service.operation_that_requires_heavy()

foo/app_services.py
这将定义给应用程序的服务公开为依赖项轻量级注入(inject)。由于服务仅附加其依赖项并返回，因此它们会为请求快速创建，然后被丢弃。

from .app import heavy
from .services import HeavyService, OtherService
from fastapi import Depends

async def get_heavy_service():
    return HeavyService(heavy=heavy)
    
async def get_other_service_that_uses_heavy(heavy_service: HeavyService = Depends(get_heavy_service)):
    return OtherService(heavy_service=heavy_service)

foo/views.py
使 FastAPI 实际提供某些服务并测试整个服务 + 重链的公开端点示例:

from fastapi import APIRouter, Depends
from .services import OtherService
from .app_services import get_other_service_that_uses_heavy

api_router = APIRouter()

@api_router.get('/')
async def index(other_service: OtherService = Depends(get_other_service_that_uses_heavy)):
    return {'hello world': other_service.other_operation()}

main.py
应用程序入口点。可以住在 app.py以及。

from fooweb.app import app

if __name__ == '__main__':
    import uvicorn

    uvicorn.run('fooweb.app:app', host='0.0.0.0', port=7272, reload=True)

通过这种方式，重客户端在启动时被初始化，而 uvicorn 在一切正常时开始服务请求。根据重客户端的实现方式，如果套接字因不活动而断开连接(正如大多数数据库库所提供的那样)，它可能需要池化并重新创建套接字。
我不确定这个例子是否容易理解，或者它是否满足您的需要，但希望它至少能让您更进一步。