domain.protocols.provider_factory_protocol

src.domain.protocols.provider_factory_protocol

Provider Factory Protocol - Abstract factory for provider adapters.

This protocol defines the interface for creating provider adapters at runtime based on provider slug. Used by handlers that need to resolve providers dynamically (e.g., sync handlers that discover provider from connection).

Architecture: - Domain layer protocol (no infrastructure imports) - Implemented by ProviderFactory in infrastructure/providers - Enables auto-wiring of handlers that need provider resolution

Usage

class SyncAccountsHandler: def init(self, ..., provider_factory: ProviderFactoryProtocol): self._provider_factory = provider_factory

async def handle(self, cmd: SyncAccounts):
    connection = await self._connection_repo.find_by_id(cmd.connection_id)
    provider = self._provider_factory.get_provider(connection.provider_slug)
    # ... use provider

Reference

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

Classes

ProviderFactoryProtocol

Bases: Protocol

Protocol for provider adapter factory.

Provides runtime resolution of provider adapters by slug. This enables handlers to be auto-wired without knowing the specific provider at construction time.

Example

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

Source code in src/domain/protocols/provider_factory_protocol.py

class ProviderFactoryProtocol(Protocol):
    """Protocol for provider adapter factory.

    Provides runtime resolution of provider adapters by slug.
    This enables handlers to be auto-wired without knowing the
    specific provider at construction time.

    Example:
        >>> factory: ProviderFactoryProtocol = ...
        >>> 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.
        """
        ...

    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.
        """
        ...

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

        Returns:
            List of supported provider slugs.
        """
        ...

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/domain/protocols/provider_factory_protocol.py

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.
    """
    ...

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/domain/protocols/provider_factory_protocol.py

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.
    """
    ...

list_supported

list_supported() -> list[str]

List all supported provider slugs.

Returns:

Type	Description
list[str]	List of supported provider slugs.

Source code in src/domain/protocols/provider_factory_protocol.py

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

    Returns:
        List of supported provider slugs.
    """
    ...