airbyte.cloud.client_config

  1# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
  2"""Cloud client configuration for Airbyte Cloud API authentication.
  3
  4This module provides the CloudClientConfig class for managing authentication
  5credentials and API configuration when connecting to Airbyte Cloud, OSS, or
  6Enterprise instances.
  7
  8Two authentication methods are supported (mutually exclusive):
  91. OAuth2 client credentials (client_id + client_secret)
 102. Bearer token authentication
 11
 12Example usage with client credentials:
 13    ```python
 14    from airbyte.cloud.client_config import CloudClientConfig
 15
 16    config = CloudClientConfig(
 17        client_id="your-client-id",
 18        client_secret="your-client-secret",
 19    )
 20    ```
 21
 22Example usage with bearer token:
 23    ```python
 24    from airbyte.cloud.client_config import CloudClientConfig
 25
 26    config = CloudClientConfig(
 27        bearer_token="your-bearer-token",
 28    )
 29    ```
 30
 31Example using environment variables:
 32    ```python
 33    from airbyte.cloud.client_config import CloudClientConfig
 34
 35    # Resolves from AIRBYTE_CLOUD_CLIENT_ID, AIRBYTE_CLOUD_CLIENT_SECRET,
 36    # AIRBYTE_CLOUD_BEARER_TOKEN, AIRBYTE_CLOUD_API_URL, and
 37    # AIRBYTE_CLOUD_CONFIG_API_URL environment variables
 38    config = CloudClientConfig.from_env()
 39    ```
 40"""
 41
 42from __future__ import annotations
 43
 44from dataclasses import dataclass
 45
 46from airbyte._util import api_util
 47from airbyte.cloud.auth import (
 48    resolve_cloud_api_url,
 49    resolve_cloud_bearer_token,
 50    resolve_cloud_client_id,
 51    resolve_cloud_client_secret,
 52    resolve_cloud_config_api_url,
 53)
 54from airbyte.exceptions import PyAirbyteInputError
 55from airbyte.secrets.base import SecretString
 56
 57
 58@dataclass
 59class CloudClientConfig:
 60    """Client configuration for Airbyte Cloud API.
 61
 62    This class encapsulates the authentication and API configuration needed to connect
 63    to Airbyte Cloud, OSS, or Enterprise instances. It supports two mutually
 64    exclusive authentication methods:
 65
 66    1. OAuth2 client credentials flow (client_id + client_secret)
 67    2. Bearer token authentication
 68
 69    Exactly one authentication method must be provided. Providing both or neither
 70    will raise a validation error.
 71
 72    Attributes:
 73        client_id: OAuth2 client ID for client credentials flow.
 74        client_secret: OAuth2 client secret for client credentials flow.
 75        bearer_token: Pre-generated bearer token for direct authentication.
 76        api_root: The API root URL. Defaults to Airbyte Cloud API.
 77        config_api_root: The Config API root URL.
 78    """
 79
 80    client_id: SecretString | None = None
 81    """OAuth2 client ID for client credentials authentication."""
 82
 83    client_secret: SecretString | None = None
 84    """OAuth2 client secret for client credentials authentication."""
 85
 86    bearer_token: SecretString | None = None
 87    """Bearer token for direct authentication (alternative to client credentials)."""
 88
 89    api_root: str = api_util.CLOUD_API_ROOT
 90    """The API root URL. Defaults to Airbyte Cloud API."""
 91
 92    config_api_root: str | None = None
 93    """The Config API root URL."""
 94
 95    def __post_init__(self) -> None:
 96        """Validate credentials and ensure secrets are properly wrapped."""
 97        # Wrap secrets in SecretString if they aren't already
 98        if self.client_id is not None:
 99            self.client_id = SecretString(self.client_id)
100        if self.client_secret is not None:
101            self.client_secret = SecretString(self.client_secret)
102        if self.bearer_token is not None:
103            self.bearer_token = SecretString(self.bearer_token)
104
105        # Validate mutual exclusivity
106        has_client_credentials = self.client_id is not None or self.client_secret is not None
107        has_bearer_token = self.bearer_token is not None
108
109        if has_client_credentials and has_bearer_token:
110            raise PyAirbyteInputError(
111                message="Cannot use both client credentials and bearer token authentication.",
112                guidance=(
113                    "Provide either client_id and client_secret together, "
114                    "or bearer_token alone, but not both."
115                ),
116            )
117
118        if has_client_credentials and (self.client_id is None or self.client_secret is None):
119            # If using client credentials, both must be provided
120            raise PyAirbyteInputError(
121                message="Incomplete client credentials.",
122                guidance=(
123                    "When using client credentials authentication, "
124                    "both client_id and client_secret must be provided."
125                ),
126            )
127
128        if not has_client_credentials and not has_bearer_token:
129            raise PyAirbyteInputError(
130                message="No authentication credentials provided.",
131                guidance=(
132                    "Provide either client_id and client_secret together for OAuth2 "
133                    "client credentials flow, or bearer_token for direct authentication."
134                ),
135            )
136
137    @property
138    def uses_bearer_token(self) -> bool:
139        """Return True if using bearer token authentication."""
140        return self.bearer_token is not None
141
142    @property
143    def uses_client_credentials(self) -> bool:
144        """Return True if using client credentials authentication."""
145        return self.client_id is not None and self.client_secret is not None
146
147    @classmethod
148    def from_env(
149        cls,
150        *,
151        api_root: str | None = None,
152        config_api_root: str | None = None,
153    ) -> CloudClientConfig:
154        """Create CloudClientConfig from environment variables.
155
156        This factory method resolves credentials from environment variables,
157        providing a convenient way to create credentials without explicitly
158        passing secrets.
159
160        Environment variables used:
161            - `AIRBYTE_CLOUD_CLIENT_ID`: OAuth client ID (for client credentials flow).
162            - `AIRBYTE_CLOUD_CLIENT_SECRET`: OAuth client secret (for client credentials flow).
163            - `AIRBYTE_CLOUD_BEARER_TOKEN`: Bearer token (alternative to client credentials).
164            - `AIRBYTE_CLOUD_API_URL`: Optional. The API root URL (defaults to Airbyte Cloud).
165            - `AIRBYTE_CLOUD_CONFIG_API_URL`: Optional. The Config API root URL.
166
167        The method will first check for a bearer token. If not found, it will
168        attempt to use client credentials.
169
170        Args:
171            api_root: The API root URL. If not provided, will be resolved from
172                the `AIRBYTE_CLOUD_API_URL` environment variable, or default to
173                the Airbyte Cloud API.
174            config_api_root: The Config API root URL. If not provided, will be resolved
175                from the `AIRBYTE_CLOUD_CONFIG_API_URL` environment variable.
176
177        Returns:
178            A CloudClientConfig instance configured with credentials from the environment.
179
180        Raises:
181            PyAirbyteSecretNotFoundError: If required credentials are not found in
182                the environment.
183        """
184        resolved_api_root = resolve_cloud_api_url(api_root)
185        resolved_config_api_root = resolve_cloud_config_api_url(config_api_root)
186
187        # Try bearer token first
188        bearer_token = resolve_cloud_bearer_token()
189        if bearer_token:
190            return cls(
191                bearer_token=bearer_token,
192                api_root=resolved_api_root,
193                config_api_root=resolved_config_api_root,
194            )
195
196        # Fall back to client credentials
197        return cls(
198            client_id=resolve_cloud_client_id(),
199            client_secret=resolve_cloud_client_secret(),
200            api_root=resolved_api_root,
201            config_api_root=resolved_config_api_root,
202        )