Зависимости FastAPI с сохранением состояния ⇐ Python

[Anonymous]

Я просматривал документацию Depends, официальный пример
from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()

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

@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
return commons

Однако в моем случае мне нужно обслуживать модель машинного обучения, которая будет обновляться с определенной периодичностью (ежечасно, ежедневно и т. д.). Решение из документации (выше) зависит в вызываемой функции; Я считаю, что он кэшируется, а не генерируется каждый раз. Тем не менее, мой вариант использования — это не какие-то строительные леса, которые нужно поднимать/опускать при каждом вызове. Но мне нужен собственный класс с состоянием. Идея состоит в том, что модель ML (атрибут класса) может обновляться по расписанию и/или асинхронно, а метод ./invocats/ будет обслуживать указанную модель, отражая обновления по мере их возникновения.
В текущем состоянии я использую глобальные переменные. Это хорошо работает, когда все мое приложение умещается в одном скрипте. Однако по мере роста моего приложения мне будет интересно использовать маршрутизатор, но я обеспокоен тем, что глобальное состояние приведет к сбоям.
Есть ли подходящий способ передать экземпляр объекта класса с сохранением состояния в методах?
См. пример класса и метода
class StateManager:
def __init__(self):
self.bucket = os.environ.get("BUCKET_NAME", "artifacts_bucket")
self.s3_model_path = "./model.joblib"
self.local_model_path = './model.joblib'

def get_clients(self):
self.s3 = boto3.client('s3')

def download_model(self):
self.s3.download_file(self.bucket, self.s3_model_path, self.local_model_path)
self.model = joblib.load(self.local_model_path)

...

state = StateManager()
state.download_model()

...

@app.post("/invocations")
def invocations(request: InferenceRequest):
input_data = pd.DataFrame(dict(request), index=[0])
try:
predictions = state.model.predict(input_data)
return JSONResponse({"predictions": predictions.tolist()},
status_code=status.HTTP_200_OK)
except Exception as e:
return JSONResponse({"error": str(e)},
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)



Подробнее здесь: https://stackoverflow.com/questions/790 ... pendencies