293def get_available_connectors(
294    install_type: InstallType | str | None = InstallType.INSTALLABLE,
295) -> list[str]:
296    """Return a list of all available connectors.
297
298    Connectors will be returned in alphabetical order, with the standard prefix "source-".
299
300    Args:
301        install_type: The type of installation for the connector.
302            Defaults to `InstallType.INSTALLABLE`.
303    """
304    if install_type is None or install_type == InstallType.INSTALLABLE:
305        # Filter for installable connectors (default behavior).
306        if is_docker_installed():
307            logger.info("Docker is detected. Returning all connectors.")
308            return sorted(_get_registry_cache().keys())
309
310        logger.info("Docker was not detected. Returning only Python and Manifest-only connectors.")
311        return sorted(
312            [
313                connector_name
314                for connector_name, conn_info in _get_registry_cache().items()
315                if conn_info.language in {Language.PYTHON, Language.MANIFEST_ONLY}
316            ]
317        )
318
319    if not isinstance(install_type, InstallType):
320        install_type = InstallType(install_type)
321
322    if install_type == InstallType.PYTHON:
323        return sorted(
324            connector_name
325            for connector_name, conn_info in _get_registry_cache().items()
326            if conn_info.pypi_package_name is not None
327        )
328
329    if install_type == InstallType.JAVA:
330        warnings.warn(
331            message="Java connectors are not yet supported.",
332            stacklevel=2,
333        )
334        return sorted(
335            connector_name
336            for connector_name, conn_info in _get_registry_cache().items()
337            if conn_info.language == Language.JAVA
338        )
339
340    if install_type in {InstallType.DOCKER, InstallType.ANY}:
341        return sorted(_get_registry_cache().keys())
342
343    if install_type == InstallType.YAML:
344        return sorted(
345            conn.name
346            for conn in _get_registry_cache().values()
347            if InstallType.YAML in conn.install_types
348        )
349
350    # pragma: no cover  # Should never be reached.
351    raise exc.PyAirbyteInputError(
352        message="Invalid install type.",
353        context={
354            "install_type": install_type,
355        },
356    )

 85def get_colab_cache(
 86    cache_name: str = "default_cache",
 87    sub_dir: str = "Airbyte/cache",
 88    schema_name: str = "main",
 89    table_prefix: str | None = "",
 90    drive_name: str = _MY_DRIVE,
 91    mount_path: str = _GOOGLE_DRIVE_DEFAULT_MOUNT_PATH,
 92) -> DuckDBCache:
 93    """Get a local cache for storing data, using the default database path.
 94
 95    Unlike the default `DuckDBCache`, this implementation will easily persist data across multiple
 96    Colab sessions.
 97
 98    Please note that Google Colab may prompt you to authenticate with your Google account to access
 99    your Google Drive. When prompted, click the link and follow the instructions.
100
101    Colab will require access to read and write files in your Google Drive, so please be sure to
102    grant the necessary permissions when prompted.
103
104    All arguments are optional and have default values that are suitable for most use cases.
105
106    Args:
107        cache_name: The name to use for the cache. Defaults to "colab_cache". Override this if you
108            want to use a different database for different projects.
109        sub_dir: The subdirectory to store the cache in. Defaults to "Airbyte/cache". Override this
110            if you want to store the cache in a different subdirectory than the default.
111        schema_name: The name of the schema to write to. Defaults to "main". Override this if you
112            want to write to a different schema.
113        table_prefix: The prefix to use for all tables in the cache. Defaults to "". Override this
114            if you want to use a different prefix for all tables.
115        drive_name: The name of the Google Drive to use. Defaults to "MyDrive". Override this if you
116            want to store data in a shared drive instead of your personal drive.
117        mount_path: The path to mount Google Drive to. Defaults to "/content/drive". Override this
118            if you want to mount Google Drive to a different path (not recommended).
119
120    ## Usage Examples
121
122    The default `get_colab_cache` arguments are suitable for most use cases:
123
124    ```python
125    from airbyte.caches.colab import get_colab_cache
126
127    colab_cache = get_colab_cache()
128    ```
129
130    Or you can call `get_colab_cache` with custom arguments:
131
132    ```python
133    custom_cache = get_colab_cache(
134        cache_name="my_custom_cache",
135        sub_dir="Airbyte/custom_cache",
136        drive_name="My Company Drive",
137    )
138    ```
139    """
140    try:
141        from google.colab import drive  # noqa: PLC0415 # type: ignore[reportMissingImports]
142    except ImportError:
143        drive = None
144        msg = (
145            "The `google.colab` interface is only available in Google Colab. "
146            "Please run this code in a Google Colab notebook."
147        )
148        raise ImportError(msg) from None
149
150    drive.mount(mount_path)
151    drive_root = (
152        Path(mount_path) / drive_name
153        if drive_name == _MY_DRIVE
154        else Path(mount_path) / "Shareddrives" / drive_name
155    )
156
157    cache_dir = drive_root / sub_dir
158    cache_dir.mkdir(parents=True, exist_ok=True)
159    db_file_path = cache_dir / f"{cache_name}.duckdb"
160
161    print(f"Using persistent PyAirbyte cache in Google Drive: `{db_file_path}`.")
162    return DuckDBCache(
163        db_path=db_file_path,
164        cache_dir=cache_dir,
165        schema_name=schema_name,
166        table_prefix=table_prefix,
167    )

31def get_default_cache() -> DuckDBCache:
32    """Get a local cache for storing data, using the default database path.
33
34    Cache files are stored in the `.cache` directory, relative to the current
35    working directory.
36    """
37    cache_dir = DEFAULT_CACHE_ROOT / "default_cache"
38    return DuckDBCache(
39        db_path=cache_dir / "default_cache.duckdb",
40        cache_dir=cache_dir,
41    )

22def get_destination(  # noqa: PLR0913 # Too many arguments
23    name: str,
24    config: dict[str, Any] | None = None,
25    *,
26    config_change_callback: ConfigChangeCallback | None = None,
27    version: str | None = None,
28    use_python: bool | Path | str | None = None,
29    pip_url: str | None = None,
30    local_executable: Path | str | None = None,
31    docker_image: str | bool | None = None,
32    use_host_network: bool = False,
33    install_if_missing: bool = True,
34    install_root: Path | None = None,
35    no_executor: bool = False,
36) -> Destination:
37    """Get a connector by name and version.
38
39    Args:
40        name: connector name
41        config: connector config - if not provided, you need to set it later via the set_config
42            method.
43        config_change_callback: callback function to be called when the connector config changes.
44        streams: list of stream names to select for reading. If set to "*", all streams will be
45            selected. If not provided, you can set it later via the `select_streams()` or
46            `select_all_streams()` method.
47        version: connector version - if not provided, the currently installed version will be used.
48            If no version is installed, the latest available version will be used. The version can
49            also be set to "latest" to force the use of the latest available version.
50        use_python: (Optional.) Python interpreter specification:
51            - True: Use current Python interpreter. (Inferred if `pip_url` is set.)
52            - False: Use Docker instead.
53            - Path: Use interpreter at this path.
54            - str: Use specific Python version. E.g. "3.11" or "3.11.10". If the version is not yet
55                installed, it will be installed by uv. (This generally adds less than 3 seconds
56                to install times.)
57        pip_url: connector pip URL - if not provided, the pip url will be inferred from the
58            connector name.
59        local_executable: If set, the connector will be assumed to already be installed and will be
60            executed using this path or executable name. Otherwise, the connector will be installed
61            automatically in a virtual environment.
62        docker_image: If set, the connector will be executed using Docker. You can specify `True`
63            to use the default image for the connector, or you can specify a custom image name.
64            If `version` is specified and your image name does not already contain a tag
65            (e.g. `my-image:latest`), the version will be appended as a tag (e.g. `my-image:0.1.0`).
66        use_host_network: If set, along with docker_image, the connector will be executed using
67            the host network. This is useful for connectors that need to access resources on
68            the host machine, such as a local database. This parameter is ignored when
69            `docker_image` is not set.
70        install_if_missing: Whether to install the connector if it is not available locally. This
71            parameter is ignored when local_executable is set.
72        install_root: (Optional.) The root directory where the virtual environment will be
73            created. If not provided, the current working directory will be used.
74        no_executor: If True, use NoOpExecutor which fetches specs from the registry without
75            local installation. This is useful for scenarios where you need to validate
76            configurations but don't need to run the connector locally (e.g., deploying to Cloud).
77    """
78    executor = get_connector_executor(
79        name=name,
80        version=version,
81        use_python=use_python,
82        pip_url=pip_url,
83        local_executable=local_executable,
84        docker_image=docker_image,
85        use_host_network=use_host_network,
86        install_if_missing=install_if_missing,
87        install_root=install_root,
88        no_executor=no_executor,
89    )
90
91    return Destination(
92        name=name,
93        config=config,
94        config_change_callback=config_change_callback,
95        executor=executor,
96    )

 64def get_secret(
 65    secret_name: str,
 66    /,
 67    *,
 68    sources: list[SecretManager | SecretSourceEnum] | None = None,
 69    default: str | SecretString | None = None,
 70    allow_prompt: bool = True,
 71    **kwargs: dict[str, Any],
 72) -> SecretString:
 73    """Get a secret from the environment.
 74
 75    The optional `sources` argument of enum type `SecretSourceEnum` or list of `SecretSourceEnum`
 76    options. If left blank, all available sources will be checked. If a list of `SecretSourceEnum`
 77    entries is passed, then the sources will be checked using the provided ordering.
 78
 79    If `allow_prompt` is `True` or if SecretSourceEnum.PROMPT is declared in the `source` arg, then
 80    the user will be prompted to enter the secret if it is not found in any of the other sources.
 81
 82    Raises:
 83        PyAirbyteSecretNotFoundError: If the secret is not found in any of the configured sources,
 84            and if no default value is provided.
 85        PyAirbyteInputError: If an invalid source name is provided in the `sources` argument.
 86    """
 87    if secret_name.startswith(SECRETS_HYDRATION_PREFIX):
 88        # If the secret name starts with the hydration prefix, we assume it's a secret reference.
 89        # We strip the prefix and get the actual secret name.
 90        secret_name = secret_name.removeprefix(SECRETS_HYDRATION_PREFIX).lstrip()
 91
 92    if "source" in kwargs:
 93        warnings.warn(
 94            message="The `source` argument is deprecated. Use the `sources` argument instead.",
 95            category=DeprecationWarning,
 96            stacklevel=2,
 97        )
 98        sources = kwargs.pop("source")  # type: ignore [assignment]
 99
100    available_sources: dict[str, SecretManager] = {}
101    for available_source in _get_secret_sources():
102        # Add available sources to the dict. Order matters.
103        available_sources[available_source.name] = available_source
104
105    if sources is None:
106        # If ANY is in the list, then we don't need to check any other sources.
107        # This is the default behavior.
108        sources = list(available_sources.values())
109
110    elif not isinstance(sources, list):
111        sources = [sources]  # type: ignore [unreachable]  # This is a 'just in case' catch.
112
113    # Replace any SecretSourceEnum strings with the matching SecretManager object
114    for source in list(sources):
115        if isinstance(source, SecretSourceEnum):
116            if source not in available_sources:
117                raise exc.PyAirbyteInputError(
118                    guidance="Invalid secret source name.",
119                    input_value=source,
120                    context={
121                        "Available Sources": list(available_sources.keys()),
122                    },
123                )
124
125            sources[sources.index(source)] = available_sources[source]
126
127    secret_managers = cast("list[SecretManager]", sources)
128
129    if SecretSourceEnum.PROMPT in secret_managers:
130        prompt_source = secret_managers.pop(
131            # Mis-typed, but okay here since we have equality logic for the enum comparison:
132            secret_managers.index(SecretSourceEnum.PROMPT),  # type: ignore [arg-type]
133        )
134
135        if allow_prompt:
136            # Always check prompt last. Add it to the end of the list.
137            secret_managers.append(prompt_source)
138
139    for secret_mgr in secret_managers:
140        val = secret_mgr.get_secret(secret_name)
141        if val:
142            return SecretString(val)
143
144    if default:
145        return SecretString(default)
146
147    raise exc.PyAirbyteSecretNotFoundError(
148        secret_name=secret_name,
149        sources=[str(s) for s in available_sources],
150    )

 48def get_source(  # noqa: PLR0913 # Too many arguments
 49    name: str,
 50    config: dict[str, Any] | None = None,
 51    *,
 52    config_change_callback: ConfigChangeCallback | None = None,
 53    streams: str | list[str] | None = None,
 54    version: str | None = None,
 55    use_python: bool | Path | str | None = None,
 56    pip_url: str | None = None,
 57    local_executable: Path | str | None = None,
 58    docker_image: bool | str | None = None,
 59    use_host_network: bool = False,
 60    source_manifest: bool | dict | Path | str | None = None,
 61    install_if_missing: bool = True,
 62    install_root: Path | None = None,
 63    no_executor: bool = False,
 64) -> Source:
 65    """Get a connector by name and version.
 66
 67    If an explicit install or execution method is requested (e.g. `local_executable`,
 68    `docker_image`, `pip_url`, `source_manifest`), the connector will be executed using this method.
 69
 70    Otherwise, an appropriate method will be selected based on the available connector metadata:
 71    1. If the connector is registered and has a YAML source manifest is available, the YAML manifest
 72       will be downloaded and used to to execute the connector.
 73    2. Else, if the connector is registered and has a PyPI package, it will be installed via pip.
 74    3. Else, if the connector is registered and has a Docker image, and if Docker is available, it
 75       will be executed using Docker.
 76
 77    Args:
 78        name: connector name
 79        config: connector config - if not provided, you need to set it later via the set_config
 80            method.
 81        config_change_callback: callback function to be called when the connector config changes.
 82        streams: list of stream names to select for reading. If set to "*", all streams will be
 83            selected. If not provided, you can set it later via the `select_streams()` or
 84            `select_all_streams()` method.
 85        version: connector version - if not provided, the currently installed version will be used.
 86            If no version is installed, the latest available version will be used. The version can
 87            also be set to "latest" to force the use of the latest available version.
 88        use_python: (Optional.) Python interpreter specification:
 89            - True: Use current Python interpreter. (Inferred if `pip_url` is set.)
 90            - False: Use Docker instead.
 91            - Path: Use interpreter at this path.
 92            - str: Use specific Python version. E.g. "3.11" or "3.11.10". If the version is not yet
 93                installed, it will be installed by uv. (This generally adds less than 3 seconds
 94                to install times.)
 95        pip_url: connector pip URL - if not provided, the pip url will be inferred from the
 96            connector name.
 97        local_executable: If set, the connector will be assumed to already be installed and will be
 98            executed using this path or executable name. Otherwise, the connector will be installed
 99            automatically in a virtual environment.
100        docker_image: If set, the connector will be executed using Docker. You can specify `True`
101            to use the default image for the connector, or you can specify a custom image name.
102            If `version` is specified and your image name does not already contain a tag
103            (e.g. `my-image:latest`), the version will be appended as a tag (e.g. `my-image:0.1.0`).
104        use_host_network: If set, along with docker_image, the connector will be executed using
105            the host network. This is useful for connectors that need to access resources on
106            the host machine, such as a local database. This parameter is ignored when
107            `docker_image` is not set.
108        source_manifest: If set, the connector will be executed based on a declarative YAML
109            source definition. This input can be `True` to attempt to auto-download a YAML spec,
110            `dict` to accept a Python dictionary as the manifest, `Path` to pull a manifest from
111            the local file system, or `str` to pull the definition from a web URL.
112        install_if_missing: Whether to install the connector if it is not available locally. This
113            parameter is ignored when `local_executable` or `source_manifest` are set.
114        install_root: (Optional.) The root directory where the virtual environment will be
115            created. If not provided, the current working directory will be used.
116        no_executor: If True, use NoOpExecutor which fetches specs from the registry without
117            local installation. This is useful for scenarios where you need to validate
118            configurations but don't need to run the connector locally (e.g., deploying to Cloud).
119    """
120    executor = get_connector_executor(
121        name=name,
122        version=version,
123        use_python=use_python,
124        pip_url=pip_url,
125        local_executable=local_executable,
126        docker_image=docker_image,
127        use_host_network=use_host_network,
128        source_manifest=source_manifest,
129        install_if_missing=install_if_missing,
130        install_root=install_root,
131        no_executor=no_executor,
132    )
133
134    return Source(
135        name=name,
136        config=config,
137        config_change_callback=config_change_callback,
138        streams=streams,
139        executor=executor,
140    )

44def new_local_cache(
45    cache_name: str | None = None,
46    cache_dir: str | Path | None = None,
47    *,
48    cleanup: bool = True,
49) -> DuckDBCache:
50    """Get a local cache for storing data, using a name string to seed the path.
51
52    Args:
53        cache_name: Name to use for the cache. Defaults to None.
54        cache_dir: Root directory to store the cache in. Defaults to None.
55        cleanup: Whether to clean up temporary files. Defaults to True.
56
57    Cache files are stored in the `.cache` directory, relative to the current
58    working directory.
59    """
60    if cache_name:
61        if " " in cache_name:
62            raise exc.PyAirbyteInputError(
63                message="Cache name cannot contain spaces.",
64                input_value=cache_name,
65            )
66
67        if not cache_name.replace("_", "").isalnum():
68            raise exc.PyAirbyteInputError(
69                message="Cache name can only contain alphanumeric characters and underscores.",
70                input_value=cache_name,
71            )
72
73    cache_name = cache_name or str(ulid.ULID())
74    cache_dir = cache_dir or (DEFAULT_CACHE_ROOT / cache_name)
75    if not isinstance(cache_dir, Path):
76        cache_dir = Path(cache_dir)
77
78    return DuckDBCache(
79        db_path=cache_dir / f"db_{cache_name}.duckdb",
80        cache_dir=cache_dir,
81        cleanup=cleanup,
82    )

39class BigQueryCache(BigQueryConfig, CacheBase):
40    """The BigQuery cache implementation."""
41
42    _sql_processor_class: ClassVar[type[SqlProcessorBase]] = BigQuerySqlProcessor
43
44    paired_destination_name: ClassVar[str | None] = "destination-bigquery"
45    paired_destination_config_class: ClassVar[type | None] = DestinationBigquery
46
47    @property
48    def paired_destination_config(self) -> DestinationBigquery:
49        """Return a dictionary of destination configuration values."""
50        return bigquery_cache_to_destination_configuration(cache=self)
51
52    def get_arrow_dataset(
53        self,
54        stream_name: str,
55        *,
56        max_chunk_size: int = DEFAULT_ARROW_MAX_CHUNK_SIZE,
57    ) -> NoReturn:
58        """Raises NotImplementedError; BigQuery doesn't support `pd.read_sql_table`.
59
60        See: https://github.com/airbytehq/PyAirbyte/issues/165
61        """
62        raise NotImplementedError(
63            "BigQuery doesn't currently support to_arrow"
64            "Please consider using a different cache implementation for these functionalities."
65        )

150class CachedDataset(SQLDataset):
151    """A dataset backed by a SQL table cache.
152
153    Because this dataset includes all records from the underlying table, we also expose the
154    underlying table as a SQLAlchemy Table object.
155    """
156
157    def __init__(
158        self,
159        cache: CacheBase,
160        stream_name: str,
161        stream_configuration: ConfiguredAirbyteStream | Literal[False] | None = None,
162    ) -> None:
163        """We construct the query statement by selecting all columns from the table.
164
165        This prevents the need to scan the table schema to construct the query statement.
166
167        If stream_configuration is None, we attempt to retrieve the stream configuration from the
168        cache processor. This is useful when constructing a dataset from a CachedDataset object,
169        which already has the stream configuration.
170
171        If stream_configuration is set to False, we skip the stream configuration retrieval.
172        """
173        table_name = cache.processor.get_sql_table_name(stream_name)
174        schema_name = cache.schema_name
175        query = select("*").select_from(text(f"{schema_name}.{table_name}"))
176        super().__init__(
177            cache=cache,
178            stream_name=stream_name,
179            query_statement=query,
180            stream_configuration=stream_configuration,
181        )
182
183    @overrides
184    def to_pandas(self) -> DataFrame:
185        """Return the underlying dataset data as a pandas DataFrame."""
186        return self._cache.get_pandas_dataframe(self._stream_name)
187
188    @overrides
189    def to_arrow(
190        self,
191        *,
192        max_chunk_size: int = DEFAULT_ARROW_MAX_CHUNK_SIZE,
193    ) -> Dataset:
194        """Return an Arrow Dataset containing the data from the specified stream.
195
196        Args:
197            stream_name (str): Name of the stream to retrieve data from.
198            max_chunk_size (int): max number of records to include in each batch of pyarrow dataset.
199
200        Returns:
201            pa.dataset.Dataset: Arrow Dataset containing the stream's data.
202        """
203        return self._cache.get_arrow_dataset(
204            stream_name=self._stream_name,
205            max_chunk_size=max_chunk_size,
206        )
207
208    def to_sql_table(self) -> Table:
209        """Return the underlying SQL table as a SQLAlchemy Table object."""
210        return self._cache.processor.get_sql_table(self.stream_name)
211
212    def __eq__(self, value: object) -> bool:
213        """Return True if the value is a CachedDataset with the same cache and stream name.
214
215        In the case of CachedDataset objects, we can simply compare the cache and stream name.
216
217        Note that this equality check is only supported on CachedDataset objects and not for
218        the base SQLDataset implementation. This is because of the complexity and computational
219        cost of comparing two arbitrary SQL queries that could be bound to different variables,
220        as well as the chance that two queries can be syntactically equivalent without being
221        text-wise equivalent.
222        """
223        if not isinstance(value, SQLDataset):
224            return False
225
226        if self._cache is not value._cache:
227            return False
228
229        return not self._stream_name != value._stream_name
230
231    def __hash__(self) -> int:
232        return hash(self._stream_name)

 46class Destination(ConnectorBase, AirbyteWriterInterface):
 47    """A class representing a destination that can be called."""
 48
 49    connector_type = "destination"
 50
 51    def __init__(
 52        self,
 53        executor: Executor,
 54        name: str,
 55        config: dict[str, Any] | None = None,
 56        *,
 57        config_change_callback: ConfigChangeCallback | None = None,
 58        validate: bool = False,
 59    ) -> None:
 60        """Initialize the source.
 61
 62        If config is provided, it will be validated against the spec if validate is True.
 63        """
 64        super().__init__(
 65            executor=executor,
 66            name=name,
 67            config=config,
 68            config_change_callback=config_change_callback,
 69            validate=validate,
 70        )
 71
 72    @staticmethod
 73    def _normalize_destination_name(name: str) -> str:
 74        """Normalize a destination name to canonical form (`destination-<type>`).
 75
 76        Accepts either the short form (e.g. `snowflake`) or the canonical
 77        form (e.g. `destination-snowflake`).
 78        """
 79        if not name.startswith(_CANONICAL_PREFIX):
 80            return f"{_CANONICAL_PREFIX}{name}"
 81        return name
 82
 83    @property
 84    def is_cache_supported(self) -> bool:
 85        """Whether this destination has a compatible cache implementation.
 86
 87        Returns `True` when `get_sql_cache()` is expected to succeed for
 88        the destination's connector type.
 89        """
 90        dest_type = self._normalize_destination_name(
 91            self.name,
 92        ).replace(_CANONICAL_PREFIX, "")
 93        return dest_type in get_supported_destination_types()
 94
 95    def get_sql_cache(
 96        self,
 97        *,
 98        schema_name: str | None = None,
 99    ) -> CacheBase:
100        """Return a SQL Cache for querying data written by this destination.
101
102        This follows the same pattern as
103        `SyncResult.get_sql_cache()` in `airbyte.cloud.sync_results`:
104        it builds a cache from the destination's configuration using
105        `destination_to_cache()`.
106
107        Args:
108            schema_name: Override the schema/namespace on the returned cache.
109                When `None` the cache uses the default schema from the
110                destination config.
111
112        Raises:
113            ValueError: If the destination type is not supported.
114        """
115        resolved_name = self._normalize_destination_name(self.name)
116        config = dict(self._hydrated_config)
117
118        # Ensure the config carries a destinationType key so that
119        # destination_to_cache() can dispatch correctly.
120        if "destinationType" not in config and "DESTINATION_TYPE" not in config:
121            dest_type = resolved_name.replace(_CANONICAL_PREFIX, "")
122            config["destinationType"] = dest_type
123
124        return destination_to_cache(config, schema_name=schema_name)
125
126    def write(  # noqa: PLR0912, PLR0915 # Too many arguments/statements
127        self,
128        source_data: Source | ReadResult,
129        *,
130        streams: list[str] | Literal["*"] | None = None,
131        cache: CacheBase | Literal[False] | None = None,
132        state_cache: CacheBase | Literal[False] | None = None,
133        write_strategy: WriteStrategy = WriteStrategy.AUTO,
134        force_full_refresh: bool = False,
135    ) -> WriteResult:
136        """Write data from source connector or already cached source data.
137
138        Caching is enabled by default, unless explicitly disabled.
139
140        Args:
141            source_data: The source data to write. Can be a `Source` or a `ReadResult` object.
142            streams: The streams to write to the destination. If omitted or if "*" is provided,
143                all streams will be written. If `source_data` is a source, then streams must be
144                selected here or on the source. If both are specified, this setting will override
145                the stream selection on the source.
146            cache: The cache to use for reading source_data. If `None`, no cache will be used. If
147                False, the cache will be disabled. This must be `None` if `source_data` is already
148                a `Cache` object.
149            state_cache: A cache to use for storing incremental state. You do not need to set this
150                if `cache` is specified or if `source_data` is a `Cache` object. Set to `False` to
151                disable state management.
152            write_strategy: The strategy to use for writing source_data. If `AUTO`, the connector
153                will decide the best strategy to use.
154            force_full_refresh: Whether to force a full refresh of the source_data. If `True`, any
155                existing state will be ignored and all source data will be reloaded.
156
157        For incremental syncs, `cache` or `state_cache` will be checked for matching state values.
158        If the cache has tracked state, this will be used for the sync. Otherwise, if there is
159        a known destination state, the destination-specific state will be used. If neither are
160        available, a full refresh will be performed.
161        """
162        if not isinstance(source_data, ReadResult | Source):
163            raise exc.PyAirbyteInputError(
164                message="Invalid source_data type for `source_data` arg.",
165                context={
166                    "source_data_type_provided": type(source_data).__name__,
167                },
168            )
169
170        # Resolve `source`, `read_result`, and `source_name`
171        source: Source | None = source_data if isinstance(source_data, Source) else None
172        read_result: ReadResult | None = (
173            source_data if isinstance(source_data, ReadResult) else None
174        )
175        source_name: str = source.name if source else cast("ReadResult", read_result).source_name
176
177        # State providers and writers default to no-op, unless overridden below.
178        cache_state_provider: StateProviderBase = StaticInputState([])
179        """Provides the state of the cache's data."""
180        cache_state_writer: StateWriterBase = NoOpStateWriter()
181        """Writes updates for the state of the cache's data."""
182        destination_state_provider: StateProviderBase = StaticInputState([])
183        """Provides the state of the destination's data, from `cache` or `state_cache`."""
184        destination_state_writer: StateWriterBase = NoOpStateWriter()
185        """Writes updates for the state of the destination's data, to `cache` or `state_cache`."""
186
187        # If caching not explicitly disabled
188        if cache is not False:
189            # Resolve `cache`, `cache_state_provider`, and `cache_state_writer`
190            if isinstance(source_data, ReadResult):
191                cache = source_data.cache
192
193            cache = cache or get_default_cache()
194            cache_state_provider = cache.get_state_provider(
195                source_name=source_name,
196                destination_name=None,  # This will just track the cache state
197            )
198            cache_state_writer = cache.get_state_writer(
199                source_name=source_name,
200                destination_name=None,  # This will just track the cache state
201            )
202
203        # Resolve `state_cache`
204        if state_cache is None:
205            state_cache = cache or get_default_cache()
206
207        # Resolve `destination_state_writer` and `destination_state_provider`
208        if state_cache:
209            destination_state_writer = state_cache.get_state_writer(
210                source_name=source_name,
211                destination_name=self.name,
212            )
213            if not force_full_refresh:
214                destination_state_provider = state_cache.get_state_provider(
215                    source_name=source_name,
216                    destination_name=self.name,
217                )
218        elif state_cache is not False:
219            warnings.warn(
220                "No state backend or cache provided. State will not be tracked."
221                "To track state, provide a cache or state backend."
222                "To silence this warning, set `state_cache=False` explicitly.",
223                category=exc.PyAirbyteWarning,
224                stacklevel=2,
225            )
226
227        # Resolve `catalog_provider`
228        if source:
229            catalog_provider = CatalogProvider(
230                configured_catalog=source.get_configured_catalog(
231                    streams=streams,
232                    force_full_refresh=force_full_refresh,
233                )
234            )
235        elif read_result:
236            catalog_provider = CatalogProvider.from_read_result(read_result)
237        else:
238            raise exc.PyAirbyteInternalError(
239                message="`source_data` must be a `Source` or `ReadResult` object.",
240            )
241
242        progress_tracker = ProgressTracker(
243            source=source if isinstance(source_data, Source) else None,
244            cache=cache or None,
245            destination=self,
246            expected_streams=catalog_provider.stream_names,
247        )
248
249        source_state_provider: StateProviderBase
250        source_state_provider = JoinedStateProvider(
251            primary=cache_state_provider,
252            secondary=destination_state_provider,
253        )
254
255        if source:
256            if cache is False:
257                # Get message iterator for source (caching disabled)
258                message_iterator: AirbyteMessageIterator = source._get_airbyte_message_iterator(  # noqa: SLF001 # Non-public API
259                    streams=streams,
260                    state_provider=source_state_provider,
261                    progress_tracker=progress_tracker,
262                    force_full_refresh=force_full_refresh,
263                )
264            else:
265                # Caching enabled and we are reading from a source.
266                # Read the data to cache if caching is enabled.
267                read_result = source._read_to_cache(  # noqa: SLF001  # Non-public API
268                    cache=cache,
269                    state_provider=source_state_provider,
270                    state_writer=cache_state_writer,
271                    catalog_provider=catalog_provider,
272                    stream_names=catalog_provider.stream_names,
273                    write_strategy=write_strategy,
274                    force_full_refresh=force_full_refresh,
275                    skip_validation=False,
276                    progress_tracker=progress_tracker,
277                )
278                message_iterator = AirbyteMessageIterator.from_read_result(
279                    read_result=read_result,
280                )
281        else:  # Else we are reading from a read result
282            assert read_result is not None
283            message_iterator = AirbyteMessageIterator.from_read_result(
284                read_result=read_result,
285            )
286
287        # Write the data to the destination
288        try:
289            self._write_airbyte_message_stream(
290                stdin=message_iterator,
291                catalog_provider=catalog_provider,
292                write_strategy=write_strategy,
293                state_writer=destination_state_writer,
294                progress_tracker=progress_tracker,
295            )
296        except Exception as ex:
297            progress_tracker.log_failure(exception=ex)
298            raise
299        else:
300            # No exceptions were raised, so log success
301            progress_tracker.log_success()
302
303        return WriteResult(
304            destination=self,
305            source_data=source_data,
306            catalog_provider=catalog_provider,
307            state_writer=destination_state_writer,
308            progress_tracker=progress_tracker,
309        )
310
311    def _write_airbyte_message_stream(
312        self,
313        stdin: IO[str] | AirbyteMessageIterator,
314        *,
315        catalog_provider: CatalogProvider,
316        write_strategy: WriteStrategy,
317        state_writer: StateWriterBase | None = None,
318        progress_tracker: ProgressTracker,
319    ) -> None:
320        """Read from the connector and write to the cache."""
321        # Run optional validation step
322        if state_writer is None:
323            state_writer = StdOutStateWriter()
324
325        # Apply the write strategy to the catalog provider before sending to the destination
326        catalog_provider = catalog_provider.with_write_strategy(write_strategy)
327
328        with as_temp_files(
329            files_contents=[
330                self._hydrated_config,
331                catalog_provider.configured_catalog.model_dump_json(exclude_none=True),
332            ]
333        ) as [
334            config_file,
335            catalog_file,
336        ]:
337            try:
338                # We call the connector to write the data, tallying the inputs and outputs
339                for destination_message in progress_tracker.tally_confirmed_writes(
340                    messages=self._execute(
341                        args=[
342                            "write",
343                            "--config",
344                            config_file,
345                            "--catalog",
346                            catalog_file,
347                        ],
348                        stdin=AirbyteMessageIterator(
349                            progress_tracker.tally_pending_writes(
350                                stdin,
351                            )
352                        ),
353                    )
354                ):
355                    if destination_message.state:
356                        state_writer.write_state(state_message=destination_message.state)
357
358            except exc.AirbyteConnectorFailedError as ex:
359                raise exc.AirbyteConnectorWriteError(
360                    connector_name=self.name,
361                    log_text=self._last_log_messages,
362                    original_exception=ex,
363                ) from None