Etc Programming
FastAPI
Dishka Generic DI

Dishka를 사용한 다중 구현체 의존성 주입 패턴

개요

동일한 인터페이스(추상 클래스)에 대해 여러 구현체를 가진 경우, Dishka를 사용하여 적절한 구현체를 주입하는 방법을 정리합니다. 이 문서는 LLM 클라이언트(OpenAI, Gemini)를 예시로 사용합니다.

현재 프로젝트 구조

src/thirdParties/ml/llm/
├── llm_client.py          # LLMClient ABC (인터페이스)
├── impl/
│   ├── openai_client_impl.py   # OpenAI 구현체
│   └── gemini_client_impl.py   # Gemini 구현체

방법 1: Alias를 사용한 구분

타입 별칭(NewType)과 alias를 조합하여 각 구현체를 구분합니다.

from typing import NewType
from dishka import Provider, Scope, provide, alias
from src.thirdParties.ml.llm.llm_client import LLMClient
from src.thirdParties.ml.llm.impl.openai_client_impl import OpenAIClientImpl
from src.thirdParties.ml.llm.impl.gemini_client_impl import GeminiClientImpl
 
# 타입 별칭 생성
OpenAILLMClient = NewType('OpenAILLMClient', LLMClient)
GeminiLLMClient = NewType('GeminiLLMClient', LLMClient)
 
class ThirdPartyProvider(Provider):
    def __init__(self):
        super().__init__(scope=Scope.APP)
    
    @provide(scope=Scope.APP)
    def provide_openai_llm_client(self, logger: Logger) -> OpenAIClientImpl:
        config = LLMClientConfig(
            api_key="your-openai-key",
            default_model=ChatModel.GPT_4O,
            default_embedding_model=EmbeddingModel.TEXT_EMBEDDING_3_LARGE,
        )
        return OpenAIClientImpl(config, logger)
    
    @provide(scope=Scope.APP)
    def provide_gemini_llm_client(self, logger: Logger) -> GeminiClientImpl:
        config = LLMClientConfig(
            api_key="your-gemini-key",
            default_model=ChatModel.GEMINI_2_0_FLASH,
            default_embedding_model=EmbeddingModel.GEMINI_EMBEDDING_001,
        )
        return GeminiClientImpl(config, logger)
    
    # alias로 타입 매핑
    openai_client_alias = alias(source=OpenAIClientImpl, provides=OpenAILLMClient)
    gemini_client_alias = alias(source=GeminiClientImpl, provides=GeminiLLMClient)
    default_client_alias = alias(source=GeminiClientImpl, provides=LLMClient)  # 기본값

사용 예시

# 특정 구현체 주입
def __init__(self, openai_client: OpenAILLMClient, gemini_client: GeminiLLMClient):
    self._openai = openai_client
    self._gemini = gemini_client
 
# 기본 구현체 주입
def __init__(self, llm_client: LLMClient):
    self._client = llm_client  # Gemini 구현체가 주입됨

방법 2: Annotated를 사용한 구분 (추천)

Python의 Annotated 타입과 리터럴 값을 사용하여 구현체를 구분합니다.

from typing import Annotated
from dishka import Provider, Scope, provide
 
class LLMProvider(str):
    """LLM 제공자 식별자"""
    OPENAI = "openai"
    GEMINI = "gemini"
 
class ThirdPartyProvider(Provider):
    def __init__(self):
        super().__init__(scope=Scope.APP)
    
    @provide(scope=Scope.APP)
    def provide_openai_client(self, logger: Logger) -> Annotated[LLMClient, LLMProvider.OPENAI]:
        config = LLMClientConfig(
            api_key="your-openai-key",
            default_model=ChatModel.GPT_4O,
            default_embedding_model=EmbeddingModel.TEXT_EMBEDDING_3_LARGE,
        )
        return OpenAIClientImpl(config, logger)
    
    @provide(scope=Scope.APP)
    def provide_gemini_client(self, logger: Logger) -> Annotated[LLMClient, LLMProvider.GEMINI]:
        config = LLMClientConfig(
            api_key="your-gemini-key", 
            default_model=ChatModel.GEMINI_2_0_FLASH,
            default_embedding_model=EmbeddingModel.GEMINI_EMBEDDING_001,
        )
        return GeminiClientImpl(config, logger)
    
    @provide(scope=Scope.APP)
    def provide_default_llm_client(
        self, 
        gemini: Annotated[LLMClient, LLMProvider.GEMINI]
    ) -> LLMClient:
        return gemini  # Gemini를 기본으로 사용

사용 예시

from typing import Annotated
from dishka import inject
 
class ExplainService:
    @inject
    async def __init__(
        self, 
        openai_client: Annotated[LLMClient, LLMProvider.OPENAI],
        gemini_client: Annotated[LLMClient, LLMProvider.GEMINI],
        default_client: LLMClient  # 기본 클라이언트
    ):
        self._openai = openai_client
        self._gemini = gemini_client
        self._default = default_client

방법 3: Factory 패턴

모든 구현체를 관리하는 Factory 클래스를 사용합니다.

from enum import Enum
from dishka import Provider, Scope, provide
 
class LLMProviderType(Enum):
    OPENAI = "openai"
    GEMINI = "gemini"
 
class LLMClientFactory:
    def __init__(self, openai_client: OpenAIClientImpl, gemini_client: GeminiClientImpl):
        self._clients = {
            LLMProviderType.OPENAI: openai_client,
            LLMProviderType.GEMINI: gemini_client,
        }
        self._default = LLMProviderType.GEMINI
    
    def get_client(self, provider_type: LLMProviderType) -> LLMClient:
        return self._clients[provider_type]
    
    def get_default_client(self) -> LLMClient:
        return self._clients[self._default]
    
    def get_all_clients(self) -> dict[LLMProviderType, LLMClient]:
        return self._clients.copy()
 
class ThirdPartyProvider(Provider):
    def __init__(self):
        super().__init__(scope=Scope.APP)
    
    @provide(scope=Scope.APP)
    def provide_openai_impl(self, logger: Logger) -> OpenAIClientImpl:
        config = LLMClientConfig(
            api_key="your-openai-key",
            default_model=ChatModel.GPT_4O,
        )
        return OpenAIClientImpl(config, logger)
    
    @provide(scope=Scope.APP)
    def provide_gemini_impl(self, logger: Logger) -> GeminiClientImpl:
        config = LLMClientConfig(
            api_key="your-gemini-key",
            default_model=ChatModel.GEMINI_2_0_FLASH,
        )
        return GeminiClientImpl(config, logger)
    
    @provide(scope=Scope.APP)
    def provide_llm_factory(
        self, 
        openai: OpenAIClientImpl, 
        gemini: GeminiClientImpl
    ) -> LLMClientFactory:
        return LLMClientFactory(
            openai_client=openai,
            gemini_client=gemini,
        )
    
    @provide(scope=Scope.APP)
    def provide_default_llm_client(self, factory: LLMClientFactory) -> LLMClient:
        return factory.get_default_client()

사용 예시

class ExplainService:
    @inject
    async def __init__(self, llm_factory: LLMClientFactory):
        self._factory = llm_factory
    
    async def explain_with_openai(self, text: str):
        client = self._factory.get_client(LLMProviderType.OPENAI)
        return await client.chat_async(...)
    
    async def explain_with_gemini(self, text: str):
        client = self._factory.get_client(LLMProviderType.GEMINI)
        return await client.chat_async(...)

각 방법의 장단점

방법 1: Alias

  • ✅ 타입 레벨에서 구분 가능
  • ✅ IDE 자동완성 지원 우수
  • ❌ NewType 정의가 추가로 필요
  • ❌ 동적 선택이 어려움

방법 2: Annotated (추천)

  • ✅ 명확한 타입 힌트
  • ✅ 기존 프로젝트 패턴과 일관성
  • ✅ 기본값 지원이 자연스러움
  • ✅ 확장성이 좋음
  • ❌ Annotated 타입 작성이 다소 길어짐

방법 3: Factory

  • ✅ 런타임에 동적으로 구현체 선택 가능
  • ✅ 모든 구현체를 한곳에서 관리
  • ✅ 테스트하기 쉬움
  • ❌ 추가 추상화 계층
  • ❌ 타입 힌트가 덜 명확함

추천 사항

현재 프로젝트 구조와 패턴을 고려할 때 **방법 2 (Annotated)**를 추천합니다:

  1. 이미 프로젝트에서 Annotated와 FromDishka를 사용하고 있음
  2. 타입 힌트가 명확하여 코드 가독성이 높음
  3. 필요에 따라 특정 구현체나 기본 구현체를 유연하게 주입받을 수 있음
  4. Dishka의 철학과 잘 맞음

실제 구현 예시

# backend_main/dependencies/providers/third_party_provider.py
 
from typing import Annotated
from dishka import Provider, Scope, provide
from src.configs.config import Config
from src.infrastructure.logging.logger import Logger
from src.thirdParties.ml.llm.llm_client import LLMClient
from src.thirdParties.ml.llm.impl.openai_client_impl import OpenAIClientImpl
from src.thirdParties.ml.llm.impl.gemini_client_impl import GeminiClientImpl
from src.thirdParties.ml.llm.models.llm_config import LLMClientConfig
from src.thirdParties.ml.llm.models.llm_models import ChatModel, EmbeddingModel
 
class LLMProvider(str):
    """LLM 제공자 식별자"""
    OPENAI = "openai"
    GEMINI = "gemini"
 
class ThirdPartyProvider(Provider):
    def __init__(self):
        super().__init__(scope=Scope.APP)
    
    @provide(scope=Scope.APP)
    def provide_openai_client(
        self, 
        config: Config,
        logger: Logger
    ) -> Annotated[LLMClient, LLMProvider.OPENAI]:
        llm_config = LLMClientConfig(
            api_key=config.openai_api_key,
            default_model=ChatModel.GPT_4O,
            default_embedding_model=EmbeddingModel.TEXT_EMBEDDING_3_LARGE,
        )
        return OpenAIClientImpl(llm_config, logger)
    
    @provide(scope=Scope.APP)
    def provide_gemini_client(
        self, 
        config: Config,
        logger: Logger
    ) -> Annotated[LLMClient, LLMProvider.GEMINI]:
        llm_config = LLMClientConfig(
            api_key=config.gemini_api_key, 
            default_model=ChatModel.GEMINI_2_0_FLASH,
            default_embedding_model=EmbeddingModel.GEMINI_EMBEDDING_001,
        )
        return GeminiClientImpl(llm_config, logger)
    
    @provide(scope=Scope.APP)
    def provide_default_llm_client(
        self, 
        gemini: Annotated[LLMClient, LLMProvider.GEMINI]
    ) -> LLMClient:
        # 기본적으로 Gemini를 사용
        return gemini
# 서비스에서 사용하는 예시
 
from typing import Annotated
from dishka import inject
from src.thirdParties.ml.llm.llm_client import LLMClient
 
class ExplainServiceImpl:
    @inject
    async def __init__(
        self,
        default_llm: LLMClient,  # 기본 LLM (Gemini)
        openai: Annotated[LLMClient, LLMProvider.OPENAI],  # OpenAI 명시적 주입
        logger: Logger
    ):
        self._default_llm = default_llm
        self._openai = openai
        self._logger = logger
    
    async def explain_with_default(self, text: str):
        # 기본 LLM 사용
        return await self._default_llm.chat_async(...)
    
    async def explain_with_gpt4(self, text: str):
        # OpenAI GPT-4 사용
        return await self._openai.chat_async(...)

마무리

Dishka는 다양한 방법으로 다중 구현체를 관리할 수 있는 유연성을 제공합니다. 프로젝트의 요구사항과 기존 패턴을 고려하여 적절한 방법을 선택하는 것이 중요합니다.

Celery vs FastAPI Background TaskGet Started