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

Question

我开始开发的 FastAPI 应用程序使用了多个服务，我只想在应用程序启动时初始化一次这些服务，然后在不同的地方使用该对象的方法。
它可以是云服务或任何其他重型服务。

可能的方法是使用Lazy loading和 with来实现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到每个非路由函数中。

我描述的所有方法要么是丑陋的、不方便的、非Pythonic的，要么是更糟糕的实践，我们这里也没有像flask中那样的线程局部变量和代理对象，那么对于我描述的此类问题的最佳方法是什么多于？

谢谢！

Answer 1

在启动 FastAPI 应用程序之前初始化重对象通常是一个好主意。这样，当应用程序开始侦听连接（并由负载均衡器提供）时，您就完成了初始化。

您可以设置这些依赖项，并在设置应用程序和主路由器的同一位置进行任何初始化，因为它们也是应用程序的一部分。我通常通过轻量级服务公开重型对象，该服务向控制器本身公开有用的端点，然后通过Depends.

您想要如何执行初始化取决于您在应用程序中的其他要求 - 例如，如果您计划在 cli 工具中重用基础设施或也在 cron 中使用它们。

这就是我在几个项目中所做的方式，到目前为止，它运行良好，并将代码更改保留在它们自己的附近。

模拟重载类 inheavylifting/heavy.py和from .heavy import HeavyLifterin __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

应用程序中的服务层；服务是应用程序向控制器公开的操作和服务，用于处理业务逻辑和其他相关活动。如果某项服务需要访问繁重的服务，则会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

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

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()}

主要.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 会在一切正常时开始服务请求。根据重客户端的实现方式，如果套接字可能因不活动而断开连接（正如大多数数据库库提供的那样），则可能需要池化并重新创建套接字。

我不确定这个例子是否足够容易理解，或者它是否满足您的需要，但希望它至少能让您走得更远。