Feat: Add Cloud deploy functionality out of experimental status

airbyte/_util/text_util.py

@@ -0,0 +1,22 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+"""Utility functions for working with text."""
+
+from __future__ import annotations
+
+import ulid
+
+
+def generate_ulid() -> str:
+    """Generate a new ULID."""
+    return str(ulid.ULID())
+
+
+def generate_random_suffix() -> str:
+    """Generate a random suffix for use in temporary names.
+
+    By default, this function generates a ULID and returns a 9-character string
+    which will be monotonically sortable. It is not guaranteed to be unique but
+    is sufficient for small-scale and medium-scale use cases.
+    """
+    ulid_str = generate_ulid().lower()
+    return ulid_str[:6] + ulid_str[-3:]

airbyte/caches/base.py

@@ -4,7 +4,7 @@
 from __future__ import annotations
 
 from pathlib import Path
-from typing import IO, TYPE_CHECKING, Any, final
+from typing import IO, TYPE_CHECKING, Any, ClassVar, final
 
 import pandas as pd
 import pyarrow as pa
@@ -30,7 +30,6 @@
 
     from airbyte._message_iterators import AirbyteMessageIterator
     from airbyte.caches._state_backend_base import StateBackendBase
-    from airbyte.datasets._base import DatasetBase
     from airbyte.progress import ProgressTracker
     from airbyte.shared.sql_processor import SqlProcessorBase
     from airbyte.shared.state_providers import StateProviderBase
@@ -57,16 +56,23 @@ class CacheBase(SqlConfig, AirbyteWriterInterface):
 
     _name: str = PrivateAttr()
 
-    _deployed_api_root: str | None = PrivateAttr(default=None)
-    _deployed_workspace_id: str | None = PrivateAttr(default=None)
-    _deployed_destination_id: str | None = PrivateAttr(default=None)
-
-    _sql_processor_class: type[SqlProcessorBase] = PrivateAttr()
+    _sql_processor_class: ClassVar[type[SqlProcessorBase]]
     _read_processor: SqlProcessorBase = PrivateAttr()
 
     _catalog_backend: CatalogBackendBase = PrivateAttr()
     _state_backend: StateBackendBase = PrivateAttr()
 
+    paired_destination_name: ClassVar[str | None] = None
+    paired_destination_config_class: ClassVar[type | None] = None
+
+    @property
+    def paired_destination_config(self) -> Any | dict[str, Any]:  # noqa: ANN401  # Allow Any return type
+        """Return a dictionary of destination configuration values."""
+        raise NotImplementedError(
+            f"The type '{type(self).__name__}' does not define an equivalent destination "
+            "configuration."
+        )
+
     def __init__(self, **data: Any) -> None:  # noqa: ANN401
         """Initialize the cache and backends."""
         super().__init__(**data)
@@ -231,6 +237,19 @@ def streams(self) -> dict[str, CachedDataset]:
 
         return result
 
+    @final
+    def __len__(self) -> int:
+        """Gets the number of streams."""
+        return len(self._catalog_backend.stream_names)
+
+    @final
+    def __bool__(self) -> bool:
+        """Always True.
+
+        This is needed so that caches with zero streams are not falsely (None-like).
+        """
+        return True
+
     def get_state_provider(
         self,
         source_name: str,
@@ -274,7 +293,7 @@ def register_source(
             incoming_stream_names=stream_names,
         )
 
-    def __getitem__(self, stream: str) -> DatasetBase:
+    def __getitem__(self, stream: str) -> CachedDataset:
         """Return a dataset by stream name."""
         return self.streams[stream]
 

airbyte/caches/bigquery.py

@@ -17,21 +17,36 @@
 
 from __future__ import annotations
 
-from typing import NoReturn
+from typing import TYPE_CHECKING, ClassVar, NoReturn
 
-from pydantic import PrivateAttr
+from airbyte_api.models import DestinationBigquery
 
 from airbyte._processors.sql.bigquery import BigQueryConfig, BigQuerySqlProcessor
 from airbyte.caches.base import (
     CacheBase,
 )
 from airbyte.constants import DEFAULT_ARROW_MAX_CHUNK_SIZE
+from airbyte.destinations._translate_cache_to_dest import (
+    bigquery_cache_to_destination_configuration,
+)
+
+
+if TYPE_CHECKING:
+    from airbyte.shared.sql_processor import SqlProcessorBase
 
 
 class BigQueryCache(BigQueryConfig, CacheBase):
     """The BigQuery cache implementation."""
 
-    _sql_processor_class: type[BigQuerySqlProcessor] = PrivateAttr(default=BigQuerySqlProcessor)
+    _sql_processor_class: ClassVar[type[SqlProcessorBase]] = BigQuerySqlProcessor
+
+    paired_destination_name: ClassVar[str | None] = "destination-bigquery"
+    paired_destination_config_class: ClassVar[type | None] = DestinationBigquery
+
+    @property
+    def paired_destination_config(self) -> DestinationBigquery:
+        """Return a dictionary of destination configuration values."""
+        return bigquery_cache_to_destination_configuration(cache=self)
 
     def get_arrow_dataset(
         self,

airbyte/caches/duckdb.py

@@ -17,12 +17,18 @@
 from __future__ import annotations
 
 import warnings
+from typing import TYPE_CHECKING, ClassVar
 
+from airbyte_api.models import DestinationDuckdb
 from duckdb_engine import DuckDBEngineWarning
-from pydantic import PrivateAttr
 
 from airbyte._processors.sql.duckdb import DuckDBConfig, DuckDBSqlProcessor
 from airbyte.caches.base import CacheBase
+from airbyte.destinations._translate_cache_to_dest import duckdb_cache_to_destination_configuration
+
+
+if TYPE_CHECKING:
+    from airbyte.shared.sql_processor import SqlProcessorBase
 
 
 # Suppress warnings from DuckDB about reflection on indices.
@@ -37,7 +43,15 @@
 class DuckDBCache(DuckDBConfig, CacheBase):
     """A DuckDB cache."""
 
-    _sql_processor_class: type[DuckDBSqlProcessor] = PrivateAttr(default=DuckDBSqlProcessor)
+    _sql_processor_class: ClassVar[type[SqlProcessorBase]] = DuckDBSqlProcessor
+
+    paired_destination_name: ClassVar[str | None] = "destination-duckdb"
+    paired_destination_config_class: ClassVar[type | None] = DestinationDuckdb
+
+    @property
+    def paired_destination_config(self) -> DestinationDuckdb:
+        """Return a dictionary of destination configuration values."""
+        return duckdb_cache_to_destination_configuration(cache=self)
 
 
 # Expose the Cache class and also the Config class.

airbyte/caches/motherduck.py

@@ -17,23 +17,33 @@
 from __future__ import annotations
 
 import warnings
+from typing import TYPE_CHECKING, ClassVar
 
+from airbyte_api.models import DestinationDuckdb
 from duckdb_engine import DuckDBEngineWarning
 from overrides import overrides
-from pydantic import Field, PrivateAttr
+from pydantic import Field
 
 from airbyte._processors.sql.duckdb import DuckDBConfig
 from airbyte._processors.sql.motherduck import MotherDuckSqlProcessor
 from airbyte.caches.duckdb import DuckDBCache
+from airbyte.destinations._translate_cache_to_dest import (
+    motherduck_cache_to_destination_configuration,
+)
 from airbyte.secrets import SecretString
 
 
+if TYPE_CHECKING:
+    from airbyte.shared.sql_processor import SqlProcessorBase
+
+
 class MotherDuckConfig(DuckDBConfig):
     """Configuration for the MotherDuck cache."""
 
     database: str = Field()
     api_key: SecretString = Field()
     db_path: str = Field(default="md:")
+    _paired_destination_name: str = "destination-motherduck"
 
     @overrides
     def get_sql_alchemy_url(self) -> SecretString:
@@ -61,7 +71,15 @@ def get_database_name(self) -> str:
 class MotherDuckCache(MotherDuckConfig, DuckDBCache):
     """Cache that uses MotherDuck for external persistent storage."""
 
-    _sql_processor_class: type[MotherDuckSqlProcessor] = PrivateAttr(default=MotherDuckSqlProcessor)
+    _sql_processor_class: ClassVar[type[SqlProcessorBase]] = MotherDuckSqlProcessor
+
+    paired_destination_name: ClassVar[str | None] = "destination-bigquery"
+    paired_destination_config_class: ClassVar[type | None] = DestinationDuckdb
+
+    @property
+    def paired_destination_config(self) -> DestinationDuckdb:
+        """Return a dictionary of destination configuration values."""
+        return motherduck_cache_to_destination_configuration(cache=self)
 
 
 # Expose the Cache class and also the Config class.

airbyte/caches/postgres.py

@@ -19,10 +19,19 @@
 
 from __future__ import annotations
 
-from pydantic import PrivateAttr
+from typing import TYPE_CHECKING, ClassVar
+
+from airbyte_api.models import DestinationPostgres
 
 from airbyte._processors.sql.postgres import PostgresConfig, PostgresSqlProcessor
 from airbyte.caches.base import CacheBase
+from airbyte.destinations._translate_cache_to_dest import (
+    postgres_cache_to_destination_configuration,
+)
+
+
+if TYPE_CHECKING:
+    from airbyte.shared.sql_processor import SqlProcessorBase
 
 
 class PostgresCache(PostgresConfig, CacheBase):
@@ -31,7 +40,25 @@ class PostgresCache(PostgresConfig, CacheBase):
     Also inherits config from the JsonlWriter, which is responsible for writing files to disk.
     """
 
-    _sql_processor_class = PrivateAttr(default=PostgresSqlProcessor)
+    _sql_processor_class: ClassVar[type[SqlProcessorBase]] = PostgresSqlProcessor
+
+    paired_destination_name: ClassVar[str | None] = "destination-bigquery"
+    paired_destination_config_class: ClassVar[type | None] = DestinationPostgres
+
+    @property
+    def paired_destination_config(self) -> DestinationPostgres:
+        """Return a dictionary of destination configuration values."""
+        return postgres_cache_to_destination_configuration(cache=self)
+
+    def clone_as_cloud_destination_config(self) -> DestinationPostgres:
+        """Return a DestinationPostgres instance with the same configuration."""
+        return DestinationPostgres(
+            host=self.host,
+            port=self.port,
+            username=self.username,
+            password=self.password,
+            database=self.database,
+        )
 
 
 # Expose the Cache class and also the Config class.

airbyte/caches/snowflake.py

@@ -21,19 +21,32 @@
 
 from __future__ import annotations
 
-from pydantic import PrivateAttr
+from typing import ClassVar
+
+from airbyte_api.models import DestinationSnowflake
 
 from airbyte._processors.sql.snowflake import SnowflakeConfig, SnowflakeSqlProcessor
 from airbyte.caches.base import CacheBase
-from airbyte.shared.sql_processor import RecordDedupeMode
+from airbyte.destinations._translate_cache_to_dest import (
+    snowflake_cache_to_destination_configuration,
+)
+from airbyte.shared.sql_processor import RecordDedupeMode, SqlProcessorBase
 
 
 class SnowflakeCache(SnowflakeConfig, CacheBase):
     """Configuration for the Snowflake cache."""
 
     dedupe_mode: RecordDedupeMode = RecordDedupeMode.APPEND
 
-    _sql_processor_class = PrivateAttr(default=SnowflakeSqlProcessor)
+    _sql_processor_class: ClassVar[type[SqlProcessorBase]] = SnowflakeSqlProcessor
+
+    paired_destination_name: ClassVar[str | None] = "destination-bigquery"
+    paired_destination_config_class: ClassVar[type | None] = DestinationSnowflake
+
+    @property
+    def paired_destination_config(self) -> DestinationSnowflake:
+        """Return a dictionary of destination configuration values."""
+        return snowflake_cache_to_destination_configuration(cache=self)
 
 
 # Expose the Cache class and also the Config class.