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

Question

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

可能的方法是使用 Lazy loading 和 Singlenton pattern，但我正在为 FastAPI 寻找更好的方法。

另一种可能的方法是使用 Depends class 并缓存它，但它的用法仅对路由方法有意义，对从路由方法调用的其他常规方法无效。
示例：

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") 事件来初始化 heavy class ，但是不知道如何在不使用 singleton 的情况下使这个初始化的对象可以从任何地方访问。

另一种丑陋的方法也是将初始化的对象保存到@app(，然后从请求中获取这个应用程序，但是你必须将request传递给每个非路由函数。

我描述的所有方法要么丑陋、不方便、非 Pythonic 或更糟糕的做法，我们这里也没有像 Flask 中那样的线程局部变量和代理对象，那么这种方法的最佳方法是什么我上面描述的问题？

谢谢！

Answer 1

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

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

注入服务对象

您想要执行初始化的确切方式将取决于您在应用程序中的其他要求 - 例如，如果您计划在 cli 工具中重新使用基础结构或在 cron 中使用它们。

这是我在几个项目中一直采用的方法，到目前为止效果很好，并且代码更改保持在他们自己的附近。

heavylifting/heavy.py 中的模拟重 class 和 __init__.py 中的 from .heavy import HeavyLifter：

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

应用中的服务层；服务是应用程序向控制器公开的操作和服务，处理业务逻辑和其他相关活动。如果服务需要访问 heavy，它会在该服务上添加 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()}

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

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

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

Optimal way to initialize heavy services only once in FastAPI

python

initialization

singleton

global

fastapi