Синглтон по аргументам в dependency_injector python

Содержание

Вопрос или проблема
Ответ или решение
Решение проблемы с кэшированием
Объяснение решения
Проверка работы
Заключение

Вопрос или проблема

Я использую dependency_injector в Python с FastAPI, и я хотел бы кэшировать экземпляры классов по переданным параметрам. Вот мой текущий код:

from fastapi import FastAPI, Depends
from dependency_injector import containers, providers
from dependency_injector.providers import Factory
from dependency_injector.wiring import inject, Provide, Provider

class Service:
    def __init__(self, name: str):
        print(f"Вызван конструктор для {name}")
        self.name = name

    def send_message(self) -> dict[str, str]:
        return {"message": f"Привет, {self.name}"}


class Container(containers.DeclarativeContainer):
    wiring_config = containers.WiringConfiguration(modules=[__name__])
    service_factory = providers.Factory(Service)


def create_app() -> FastAPI:
    app = FastAPI()
    app.container = Container()

    service_factory = app.container.service_factory

    print(service_factory(name="Мир").send_message())
    print(service_factory(name="Вирлд").send_message())
    print(service_factory(name="Мир").send_message())

    return app


app = create_app()


# Определите простой маршрут
@app.get("/hello/{name}")
@inject
async def hello(
    name: str,
    service_factory: Factory[Service] = Depends(Provide[Container.service_factory]),
):
    print(type(service_factory))
    service = service_factory(name=name)
    return service.send_message()

В следующем примере у меня есть две проблемы:

В функции create_app контейнерная фабрика вызывается дважды для “Мир”. (Использование providers.Singleton не работает, сохраненный экземпляр будет первым экземпляром)
По какой-то причине, в маршруте /hello/{name}, если я пытаюсь получить его, service_factory имеет тип “Provide”, и вызов service_factory(name=name) просто вызывает ошибку, говоря, что _Marker не ожидает аргументов. Я попробовал изменить WiringConfig на “.app” (текущее имя файла), но это все равно не сработало)

Поэтому мне бы хотелось, чтобы этот пример работал, не вызывая конструктор с одинаковыми аргументами дважды, и при этом имел подсказки типов.

Ответ или решение

Вопрос, который вы подняли, касается использования библиотеки dependency_injector для кэширования экземпляров классов на основе переданных аргументов в приложении FastAPI. Ваша текущая реализация сталкивается с двумя основными проблемами: повторный вызов конструктора для одинаковых аргументов и проблемы с типами в маршруте.

Решение проблемы с кэшированием

Чтобы кэшировать экземпляры определенных классов в зависимости от аргументов, необходимо реализовать механику, похожую на Singleton, но при этом учитывать параметры. Для этого мы можем использовать providers.Dynamic, который позволяет нам создать интерфейс для получения экземпляра с кэшированием на основании переданных аргументов.

Вот как можно модифицировать ваш код:

from fastapi import FastAPI, Depends
from dependency_injector import containers, providers
from dependency_injector.wiring import inject, Provide

class Service:
    def __init__(self, name: str):
        print(f"Called constructor for {name}")
        self.name = name

    def send_message(self) -> dict[str, str]:
        return {"message": f"Hello, {self.name}"}

class Container(containers.DeclarativeContainer):
    wiring_config = containers.WiringConfiguration(modules=[__name__])
    service_store = providers.Dict()

    # Dynamic provider to cache instances based on parameters
    service_provider = providers.Dynamic(
        factory=lambda name: Container.service_store[name] 
                            if name in Container.service_store 
                            else Container.service_store.setdefault(name, Service(name))
    )

def create_app() -> FastAPI:
    app = FastAPI()
    app.container = Container()
    return app

app = create_app()

# Define a simple route
@app.get("/hello/{name}")
@inject
async def hello(
    name: str,
    service: Service = Depends(Provide[Container.service_provider]),
):
    return service.send_message()

Объяснение решения

Кэширование экземпляров: Мы используем словарь service_store для хранения уже созданных экземпляров класса Service. Если сервис с данным именем уже существует в кэше, мы просто возвращаем его, иначе создаем новый экземпляр и сохраняем его в словаре.
Функция-производитель: Мы используем providers.Dynamic, что позволяет нам реализовать логику получения экземпляра на основе имени, которое передается в маршрут. Это позволяет избежать повторных вызовов конструктора Service с одинаковыми параметрами.
Изменение типа зависимости в маршруте: Теперь в маршруте hello мы используем прямую связь с экземпляром сервиса, а не с service_factory, что предотвращает ошибку, связанную с типом Provide.

Проверка работы

Теперь вы можете протестировать свой API, вызывая ‘/hello/World’ и ‘/hello/Wirld’. Операции должны вернуть ожидаемое сообщение, при этом каждый раз кэшируя экземпляры Service, предотвращая лишние вызовы конструктора.

Заключение

Эта реализация позволяет избежать повторного создания экземпляров и поддерживает типизацию, что повышает читаемость и управляемость кода. Используя данное решение, вы сможете эффективно кэшировать экземпляры классов в зависимости от аргументов, что особенно полезно в контексте приложений, работающих с зависимостями и предоставляющим высокий уровень производительности.