Skip to content

application.commands.handlers.connect_provider_handler

src.application.commands.handlers.connect_provider_handler

ConnectProvider command handler.

Handles provider connection requests. Creates new connection record and transitions to ACTIVE status with provided credentials.

Architecture: - Application layer handler (orchestrates business logic) - Imports only from domain layer (entities, protocols, events) - Uses Result types for error handling - Emits 3-state domain events (Attempted → Succeeded/Failed)

Reference
  • docs/architecture/cqrs-pattern.md
  • docs/architecture/provider-domain-model.md

Classes

ConnectProviderError

ConnectProvider-specific errors.

Source code in src/application/commands/handlers/connect_provider_handler.py 
38
39
40
41
42
43
class ConnectProviderError:
    """ConnectProvider-specific errors."""

    INVALID_CREDENTIALS = "Invalid or missing credentials"
    INVALID_PROVIDER_SLUG = "Invalid provider slug"
    DATABASE_ERROR = "Database error occurred"

ConnectProviderHandler

Handler for ConnectProvider command.

Creates a new provider connection with ACTIVE status. The actual OAuth/API authentication happens in infrastructure layer before this handler is called.

Dependencies (injected via constructor): - ProviderConnectionRepository: For persistence - EventBusProtocol: For domain events

Source code in src/application/commands/handlers/connect_provider_handler.py 
 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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
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
class ConnectProviderHandler:
    """Handler for ConnectProvider command.

    Creates a new provider connection with ACTIVE status.
    The actual OAuth/API authentication happens in infrastructure layer
    before this handler is called.

    Dependencies (injected via constructor):
        - ProviderConnectionRepository: For persistence
        - EventBusProtocol: For domain events
    """

    def __init__(
        self,
        connection_repo: ProviderConnectionRepository,
        event_bus: EventBusProtocol,
    ) -> None:
        """Initialize handler with dependencies.

        Args:
            connection_repo: Provider connection repository.
            event_bus: Event bus for publishing domain events.
        """
        self._connection_repo = connection_repo
        self._event_bus = event_bus

    async def handle(self, cmd: ConnectProvider) -> Result[UUID, str]:
        """Handle ConnectProvider command.

        Creates a new provider connection record with ACTIVE status.
        Emits domain events for audit trail and side effects.

        Args:
            cmd: ConnectProvider command with user, provider, and credentials.

        Returns:
            Success(connection_id): Connection created successfully.
            Failure(error): Validation or database error.

        Side Effects:
            - Publishes ProviderConnectionAttempted event (always)
            - Publishes ProviderConnectionSucceeded event (on success)
            - Publishes ProviderConnectionFailed event (on failure)
            - Creates ProviderConnection in database (on success)
        """
        # Step 1: Emit ATTEMPTED event
        await self._event_bus.publish(
            ProviderConnectionAttempted(
                event_id=uuid7(),
                occurred_at=datetime.now(UTC),
                user_id=cmd.user_id,
                provider_id=cmd.provider_id,
                provider_slug=cmd.provider_slug,
            )
        )

        try:
            # Step 2: Validate credentials
            if cmd.credentials is None:
                await self._emit_failed(cmd, ConnectProviderError.INVALID_CREDENTIALS)
                return cast(
                    Result[UUID, str],
                    Failure(error=ConnectProviderError.INVALID_CREDENTIALS),
                )

            # Step 3: Validate provider_slug
            if not cmd.provider_slug or len(cmd.provider_slug) > 50:
                await self._emit_failed(cmd, ConnectProviderError.INVALID_PROVIDER_SLUG)
                return cast(
                    Result[UUID, str],
                    Failure(error=ConnectProviderError.INVALID_PROVIDER_SLUG),
                )

            # Step 4: Create connection entity
            connection_id = uuid7()
            now = datetime.now(UTC)

            connection = ProviderConnection(
                id=connection_id,
                user_id=cmd.user_id,
                provider_id=cmd.provider_id,
                provider_slug=cmd.provider_slug,
                status=ConnectionStatus.ACTIVE,
                alias=cmd.alias,
                credentials=cmd.credentials,
                connected_at=now,
                created_at=now,
                updated_at=now,
            )

            # Step 5: Save to database
            await self._connection_repo.save(connection)

            # Step 6: Emit SUCCEEDED event
            await self._event_bus.publish(
                ProviderConnectionSucceeded(
                    event_id=uuid7(),
                    occurred_at=datetime.now(UTC),
                    user_id=cmd.user_id,
                    connection_id=connection_id,
                    provider_id=cmd.provider_id,
                    provider_slug=cmd.provider_slug,
                )
            )

            return Success(value=connection_id)

        except Exception as e:
            # Catch-all for database errors
            error_msg = f"{ConnectProviderError.DATABASE_ERROR}: {str(e)}"
            await self._emit_failed(cmd, error_msg)
            return cast(Result[UUID, str], Failure(error=error_msg))

    async def _emit_failed(self, cmd: ConnectProvider, reason: str) -> None:
        """Emit ProviderConnectionFailed event.

        Args:
            cmd: Original command.
            reason: Failure reason.
        """
        await self._event_bus.publish(
            ProviderConnectionFailed(
                event_id=uuid7(),
                occurred_at=datetime.now(UTC),
                user_id=cmd.user_id,
                provider_id=cmd.provider_id,
                provider_slug=cmd.provider_slug,
                reason=reason,
            )
        )
Functions
__init__ 
__init__(
    connection_repo: ProviderConnectionRepository,
    event_bus: EventBusProtocol,
) -> None

Parameters:

Name Type Description Default
connection_repo ProviderConnectionRepository

Provider connection repository.

 required
event_bus EventBusProtocol

Event bus for publishing domain events.

 required
Source code in src/application/commands/handlers/connect_provider_handler.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
def __init__(
    self,
    connection_repo: ProviderConnectionRepository,
    event_bus: EventBusProtocol,
) -> None:
    """Initialize handler with dependencies.

    Args:
        connection_repo: Provider connection repository.
        event_bus: Event bus for publishing domain events.
    """
    self._connection_repo = connection_repo
    self._event_bus = event_bus
handle async 
handle(cmd: ConnectProvider) -> Result[UUID, str]

Handle ConnectProvider command.

Creates a new provider connection record with ACTIVE status. Emits domain events for audit trail and side effects.

Parameters:

Name Type Description Default
cmd ConnectProvider

ConnectProvider command with user, provider, and credentials.

 required

Returns:

Name Type Description
Success connection_id

Connection created successfully.
Failure error

Validation or database error.
Side Effects
  • Publishes ProviderConnectionAttempted event (always)
  • Publishes ProviderConnectionSucceeded event (on success)
  • Publishes ProviderConnectionFailed event (on failure)
  • Creates ProviderConnection in database (on success)
Source code in src/application/commands/handlers/connect_provider_handler.py 
 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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
async def handle(self, cmd: ConnectProvider) -> Result[UUID, str]:
    """Handle ConnectProvider command.

    Creates a new provider connection record with ACTIVE status.
    Emits domain events for audit trail and side effects.

    Args:
        cmd: ConnectProvider command with user, provider, and credentials.

    Returns:
        Success(connection_id): Connection created successfully.
        Failure(error): Validation or database error.

    Side Effects:
        - Publishes ProviderConnectionAttempted event (always)
        - Publishes ProviderConnectionSucceeded event (on success)
        - Publishes ProviderConnectionFailed event (on failure)
        - Creates ProviderConnection in database (on success)
    """
    # Step 1: Emit ATTEMPTED event
    await self._event_bus.publish(
        ProviderConnectionAttempted(
            event_id=uuid7(),
            occurred_at=datetime.now(UTC),
            user_id=cmd.user_id,
            provider_id=cmd.provider_id,
            provider_slug=cmd.provider_slug,
        )
    )

    try:
        # Step 2: Validate credentials
        if cmd.credentials is None:
            await self._emit_failed(cmd, ConnectProviderError.INVALID_CREDENTIALS)
            return cast(
                Result[UUID, str],
                Failure(error=ConnectProviderError.INVALID_CREDENTIALS),
            )

        # Step 3: Validate provider_slug
        if not cmd.provider_slug or len(cmd.provider_slug) > 50:
            await self._emit_failed(cmd, ConnectProviderError.INVALID_PROVIDER_SLUG)
            return cast(
                Result[UUID, str],
                Failure(error=ConnectProviderError.INVALID_PROVIDER_SLUG),
            )

        # Step 4: Create connection entity
        connection_id = uuid7()
        now = datetime.now(UTC)

        connection = ProviderConnection(
            id=connection_id,
            user_id=cmd.user_id,
            provider_id=cmd.provider_id,
            provider_slug=cmd.provider_slug,
            status=ConnectionStatus.ACTIVE,
            alias=cmd.alias,
            credentials=cmd.credentials,
            connected_at=now,
            created_at=now,
            updated_at=now,
        )

        # Step 5: Save to database
        await self._connection_repo.save(connection)

        # Step 6: Emit SUCCEEDED event
        await self._event_bus.publish(
            ProviderConnectionSucceeded(
                event_id=uuid7(),
                occurred_at=datetime.now(UTC),
                user_id=cmd.user_id,
                connection_id=connection_id,
                provider_id=cmd.provider_id,
                provider_slug=cmd.provider_slug,
            )
        )

        return Success(value=connection_id)

    except Exception as e:
        # Catch-all for database errors
        error_msg = f"{ConnectProviderError.DATABASE_ERROR}: {str(e)}"
        await self._emit_failed(cmd, error_msg)
        return cast(Result[UUID, str], Failure(error=error_msg))