Skip to content

infrastructure.providers.provider_factory

src.infrastructure.providers.provider_factory

Provider Factory Implementation.

Concrete implementation of ProviderFactoryProtocol that resolves provider adapters at runtime based on slug. Delegates to existing get_provider() function from container.

Architecture: - Infrastructure layer implementation - Implements ProviderFactoryProtocol from domain - Used as app-scoped singleton via get_provider_factory()

Reference
  • docs/architecture/provider-integration-architecture.md
  • docs/architecture/dependency-injection.md

Attributes

Classes

ProviderFactory

Concrete provider factory implementation.

Provides runtime resolution of provider adapters by slug. Validates configuration before instantiating providers.

Example

factory = ProviderFactory() provider = factory.get_provider("schwab") result = await provider.fetch_accounts(credentials)

Source code in src/infrastructure/providers/provider_factory.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class ProviderFactory:
    """Concrete provider factory implementation.

    Provides runtime resolution of provider adapters by slug.
    Validates configuration before instantiating providers.

    Example:
        >>> factory = ProviderFactory()
        >>> provider = factory.get_provider("schwab")
        >>> result = await provider.fetch_accounts(credentials)
    """

    def get_provider(self, slug: str) -> ProviderProtocol:
        """Get provider adapter by slug.

        Args:
            slug: Provider identifier (e.g., 'schwab', 'alpaca', 'chase_file').

        Returns:
            Provider adapter implementing ProviderProtocol.

        Raises:
            ValueError: If provider slug is unknown or not configured.
        """
        # Lookup metadata from registry
        metadata = get_provider_metadata(slug)
        if not metadata:
            supported = ", ".join([p.slug for p in PROVIDER_REGISTRY])
            raise ValueError(f"Unknown provider: {slug}. Supported: {supported}")

        # Validate required settings (registry-driven)
        if metadata.required_settings:
            for setting_name in metadata.required_settings:
                if not getattr(settings, setting_name, None):
                    required = ", ".join(metadata.required_settings)
                    raise ValueError(
                        f"Provider '{slug}' not configured. Required settings: {required}"
                    )

        # Lazy import and instantiate (avoid circular imports)
        match slug:
            case "schwab":
                from src.infrastructure.providers.schwab import SchwabProvider

                return SchwabProvider(settings=settings)

            case "alpaca":
                from src.infrastructure.providers.alpaca import AlpacaProvider

                return AlpacaProvider(settings=settings)

            case "chase_file":
                from src.infrastructure.providers.chase import ChaseFileProvider

                return ChaseFileProvider()

            case _:
                raise ValueError(
                    f"Provider '{slug}' in registry but no factory defined. "
                    "This is a bug - please report to maintainers."
                )

    def supports(self, slug: str) -> bool:
        """Check if provider slug is supported.

        Args:
            slug: Provider identifier to check.

        Returns:
            True if provider is supported and configured.
        """
        metadata = get_provider_metadata(slug)
        if not metadata:
            return False

        # Check if required settings are present
        if metadata.required_settings:
            for setting_name in metadata.required_settings:
                if not getattr(settings, setting_name, None):
                    return False

        return True

    def list_supported(self) -> list[str]:
        """List all supported provider slugs.

        Returns:
            List of supported provider slugs that are configured.
        """
        return [p.slug for p in PROVIDER_REGISTRY if self.supports(p.slug)]
Functions
get_provider 
get_provider(slug: str) -> ProviderProtocol

Get provider adapter by slug.

Parameters:

Name Type Description Default
slug str

Provider identifier (e.g., 'schwab', 'alpaca', 'chase_file').

 required

Returns:

Type Description
ProviderProtocol

Provider adapter implementing ProviderProtocol.

Raises:

Type Description
ValueError

If provider slug is unknown or not configured.
Source code in src/infrastructure/providers/provider_factory.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def get_provider(self, slug: str) -> ProviderProtocol:
    """Get provider adapter by slug.

    Args:
        slug: Provider identifier (e.g., 'schwab', 'alpaca', 'chase_file').

    Returns:
        Provider adapter implementing ProviderProtocol.

    Raises:
        ValueError: If provider slug is unknown or not configured.
    """
    # Lookup metadata from registry
    metadata = get_provider_metadata(slug)
    if not metadata:
        supported = ", ".join([p.slug for p in PROVIDER_REGISTRY])
        raise ValueError(f"Unknown provider: {slug}. Supported: {supported}")

    # Validate required settings (registry-driven)
    if metadata.required_settings:
        for setting_name in metadata.required_settings:
            if not getattr(settings, setting_name, None):
                required = ", ".join(metadata.required_settings)
                raise ValueError(
                    f"Provider '{slug}' not configured. Required settings: {required}"
                )

    # Lazy import and instantiate (avoid circular imports)
    match slug:
        case "schwab":
            from src.infrastructure.providers.schwab import SchwabProvider

            return SchwabProvider(settings=settings)

        case "alpaca":
            from src.infrastructure.providers.alpaca import AlpacaProvider

            return AlpacaProvider(settings=settings)

        case "chase_file":
            from src.infrastructure.providers.chase import ChaseFileProvider

            return ChaseFileProvider()

        case _:
            raise ValueError(
                f"Provider '{slug}' in registry but no factory defined. "
                "This is a bug - please report to maintainers."
            )
supports 
supports(slug: str) -> bool

Check if provider slug is supported.

Parameters:

Name Type Description Default
slug str

Provider identifier to check.

 required

Returns:

Type Description
bool

True if provider is supported and configured.
Source code in src/infrastructure/providers/provider_factory.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def supports(self, slug: str) -> bool:
    """Check if provider slug is supported.

    Args:
        slug: Provider identifier to check.

    Returns:
        True if provider is supported and configured.
    """
    metadata = get_provider_metadata(slug)
    if not metadata:
        return False

    # Check if required settings are present
    if metadata.required_settings:
        for setting_name in metadata.required_settings:
            if not getattr(settings, setting_name, None):
                return False

    return True
list_supported 
list_supported() -> list[str]

List all supported provider slugs.

Returns:

Type Description
list[str]

List of supported provider slugs that are configured.
Source code in src/infrastructure/providers/provider_factory.py 
108
109
110
111
112
113
114
def list_supported(self) -> list[str]:
    """List all supported provider slugs.

    Returns:
        List of supported provider slugs that are configured.
    """
    return [p.slug for p in PROVIDER_REGISTRY if self.supports(p.slug)]

Functions