Skip to content

core.container.providers

src.core.container.providers

Provider dependency factory (Registry-driven).

Factory for creating financial provider adapters based on slug. Single factory (get_provider) with capability checking (is_oauth_provider).

Registry-Driven Pattern (F8.1): - Provider metadata stored in domain/providers/registry.py - Container uses registry for lookup and validation - Zero drift: Can't forget to add provider to OAuth list - Self-enforcing: Tests fail if registry/factory mismatch

Usage
  • get_provider(): Returns any provider by slug (auth-agnostic)
  • is_oauth_provider(): TypeGuard for OAuth capability checking
Pattern

For sync handlers (all providers)

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

For OAuth callbacks (OAuth providers only)

provider = get_provider("schwab") if is_oauth_provider(provider): tokens = await provider.exchange_code_for_tokens(code)

Reference

See docs/architecture/provider-registry-architecture.md for registry pattern. See docs/architecture/provider-integration-architecture.md for integration.

Attributes

Classes

Functions

get_provider 

get_provider(slug: str) -> ProviderProtocol

Get financial provider adapter by slug (registry-driven).

Factory function that returns the correct provider implementation based on the provider slug from registry. Follows Composition Root pattern.

Registry-Driven (F8.1): - Looks up provider metadata from PROVIDER_REGISTRY - Validates required settings via metadata.required_settings - Ensures zero drift between registry and factory

This returns providers typed as ProviderProtocol (auth-agnostic). For OAuth-specific operations, use is_oauth_provider() to check capability.

Currently supported providers (from registry): - 'schwab': Charles Schwab (OAuth, brokerage) - 'alpaca': Alpaca Markets (API Key, brokerage) - 'chase_file': Chase Bank (File Import, bank)

Parameters:

Name Type Description Default
slug str

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

 required

Returns:

Type Description
ProviderProtocol

Provider adapter implementing ProviderProtocol.

Raises:

Type Description
ValueError

If provider slug is unknown or provider not configured.
Usage
Sync handlers (auth-agnostic - works for all providers)

from src.core.container import get_provider provider = get_provider("schwab") result = await provider.fetch_accounts(credentials)

OAuth callbacks (check capability first)

provider = get_provider("schwab") if is_oauth_provider(provider): tokens = await provider.exchange_code_for_tokens(code)

Reference
  • docs/architecture/provider-registry-architecture.md
  • docs/architecture/provider-integration-architecture.md
Source code in src/core/container/providers.py 
 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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def get_provider(slug: str) -> "ProviderProtocol":
    """Get financial provider adapter by slug (registry-driven).

    Factory function that returns the correct provider implementation
    based on the provider slug from registry. Follows Composition Root pattern.

    Registry-Driven (F8.1):
        - Looks up provider metadata from PROVIDER_REGISTRY
        - Validates required settings via metadata.required_settings
        - Ensures zero drift between registry and factory

    This returns providers typed as ProviderProtocol (auth-agnostic).
    For OAuth-specific operations, use is_oauth_provider() to check capability.

    Currently supported providers (from registry):
        - 'schwab': Charles Schwab (OAuth, brokerage)
        - 'alpaca': Alpaca Markets (API Key, brokerage)
        - 'chase_file': Chase Bank (File Import, bank)

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

    Returns:
        Provider adapter implementing ProviderProtocol.

    Raises:
        ValueError: If provider slug is unknown or provider not configured.

    Usage:
        # Sync handlers (auth-agnostic - works for all providers)
        from src.core.container import get_provider
        provider = get_provider("schwab")
        result = await provider.fetch_accounts(credentials)

        # OAuth callbacks (check capability first)
        provider = get_provider("schwab")
        if is_oauth_provider(provider):
            tokens = await provider.exchange_code_for_tokens(code)

    Reference:
        - docs/architecture/provider-registry-architecture.md
        - docs/architecture/provider-integration-architecture.md
    """
    # Step 1: 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}")

    # Step 2: 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}"
                )

    # Step 3: 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 _:
            # This should never happen (registry validation in Step 1)
            raise ValueError(
                f"Provider '{slug}' in registry but no factory defined. "
                "This is a bug - please report to maintainers."
            )

is_oauth_provider 

is_oauth_provider(
    provider: ProviderProtocol,
) -> TypeGuard[OAuthProviderProtocol]

Check if provider supports OAuth authentication (registry-driven).

TypeGuard function that performs runtime check and narrows type. After this returns True, the provider is typed as OAuthProviderProtocol.

Registry-Driven (F8.1): - Uses get_oauth_providers() from registry - Zero drift: OAuth list always in sync with registry - Self-enforcing: Tests verify OAuth providers match reality

Use this to check provider capabilities before calling OAuth methods
  • exchange_code_for_tokens()
  • refresh_access_token()

Parameters:

Name Type Description Default
provider ProviderProtocol

Any provider instance.

 required

Returns:

Type Description
TypeGuard[OAuthProviderProtocol]

True if provider supports OAuth (has exchange_code_for_tokens method).
TypeGuard[OAuthProviderProtocol]

Type is narrowed to OAuthProviderProtocol when True.
Usage
In OAuth callback handler:

provider = get_provider(slug) if is_oauth_provider(provider): # Type is now OAuthProviderProtocol tokens = await provider.exchange_code_for_tokens(code) else: raise ValueError(f"Provider {slug} doesn't support OAuth")

In token refresh handler:

provider = get_provider(connection.provider_slug) if is_oauth_provider(provider): result = await provider.refresh_access_token(refresh_token) else: # API Key providers don't need token refresh return Success("No refresh needed")

Reference
  • docs/architecture/provider-registry-architecture.md
  • docs/architecture/provider-integration-architecture.md
Source code in src/core/container/providers.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def is_oauth_provider(
    provider: "ProviderProtocol",
) -> TypeGuard["OAuthProviderProtocol"]:
    """Check if provider supports OAuth authentication (registry-driven).

    TypeGuard function that performs runtime check and narrows type.
    After this returns True, the provider is typed as OAuthProviderProtocol.

    Registry-Driven (F8.1):
        - Uses get_oauth_providers() from registry
        - Zero drift: OAuth list always in sync with registry
        - Self-enforcing: Tests verify OAuth providers match reality

    Use this to check provider capabilities before calling OAuth methods:
        - exchange_code_for_tokens()
        - refresh_access_token()

    Args:
        provider: Any provider instance.

    Returns:
        True if provider supports OAuth (has exchange_code_for_tokens method).
        Type is narrowed to OAuthProviderProtocol when True.

    Usage:
        # In OAuth callback handler:
        provider = get_provider(slug)
        if is_oauth_provider(provider):
            # Type is now OAuthProviderProtocol
            tokens = await provider.exchange_code_for_tokens(code)
        else:
            raise ValueError(f"Provider {slug} doesn't support OAuth")

        # In token refresh handler:
        provider = get_provider(connection.provider_slug)
        if is_oauth_provider(provider):
            result = await provider.refresh_access_token(refresh_token)
        else:
            # API Key providers don't need token refresh
            return Success("No refresh needed")

    Reference:
        - docs/architecture/provider-registry-architecture.md
        - docs/architecture/provider-integration-architecture.md
    """
    return provider.slug in get_oauth_providers()