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