src/retrieval/factory.py

"""Factory for creating retrieval strategies."""

import logging
from typing import Any

from .base import BaseRetrieverStrategy

logger = logging.getLogger(__name__)


class RetrieverFactory:
    """Factory for creating and managing retrieval strategies.

    This factory uses a registry pattern to allow dynamic registration
    of new retrieval strategies. Strategies are registered using the
    @RetrieverFactory.register decorator.

    Example:
        @RetrieverFactory.register("my_strategy")
        class MyStrategy(BaseRetrieverStrategy):
            ...

        # Later:
        strategy = RetrieverFactory.create("my_strategy", config)
    """

    _strategies: dict[str, type[BaseRetrieverStrategy]] = {}

    @classmethod
    def register(cls, name: str):
        """Decorator to register a retrieval strategy.

        Args:
            name: Unique identifier for the strategy

        Returns:
            Decorator function that registers the strategy class

        Example:
            @RetrieverFactory.register("vector")
            class VectorStrategy(BaseRetrieverStrategy):
                ...
        """

        def decorator(strategy_class: type[BaseRetrieverStrategy]):
            if name in cls._strategies:
                logger.warning(f"Overwriting existing strategy: {name}")
            cls._strategies[name] = strategy_class
            logger.debug(f"Registered retrieval strategy: {name}")
            return strategy_class

        return decorator

    @classmethod
    def create(cls, strategy_name: str, config: dict[str, Any]) -> BaseRetrieverStrategy:
        """Create a retrieval strategy instance by name.

        Args:
            strategy_name: Name of the strategy to create
            config: Configuration dictionary for the strategy

        Returns:
            An instance of the requested strategy

        Raises:
            ValueError: If strategy_name is not registered
        """
        if strategy_name not in cls._strategies:
            available = ", ".join(cls._strategies.keys()) or "none"
            msg = f"Unknown retrieval strategy: '{strategy_name}'. Available: {available}"
            raise ValueError(msg)

        strategy_class = cls._strategies[strategy_name]
        logger.info(f"Creating retrieval strategy: {strategy_name}")

        return strategy_class(config)

    @classmethod
    def available_strategies(cls) -> list[str]:
        """List all registered strategy names.

        Returns:
            List of registered strategy names
        """
        return list(cls._strategies.keys())

    @classmethod
    def is_registered(cls, name: str) -> bool:
        """Check if a strategy is registered.

        Args:
            name: Strategy name to check

        Returns:
            True if strategy is registered, False otherwise
        """
        return name in cls._strategies

    @classmethod
    def get_strategy_class(cls, name: str) -> type[BaseRetrieverStrategy] | None:
        """Get the strategy class by name without instantiating.

        Args:
            name: Strategy name

        Returns:
            Strategy class or None if not found
        """
        return cls._strategies.get(name)


def get_retriever_factory() -> type[RetrieverFactory]:
    """Get the RetrieverFactory class.

    Returns:
        The RetrieverFactory class
    """
    return RetrieverFactory