311def get_available_connectors(install_type: InstallType | str | None = None) -> list[str]:
312    """Return a list of all available connectors.
313
314    Connectors will be returned in alphabetical order, with the standard prefix "source-".
315    """
316    if install_type is None:
317        # No install type specified. Filter for whatever is runnable.
318        if is_docker_installed():
319            logging.info("Docker is detected. Returning all connectors.")
320            # If Docker is available, return all connectors.
321            return sorted(conn.name for conn in _get_registry_cache().values())
322
323        logging.info("Docker was not detected. Returning only Python and Manifest-only connectors.")
324
325        # If Docker is not available, return only Python and Manifest-based connectors.
326        return sorted(
327            conn.name
328            for conn in _get_registry_cache().values()
329            if conn.language in {Language.PYTHON, Language.MANIFEST_ONLY}
330        )
331
332    if not isinstance(install_type, InstallType):
333        install_type = InstallType(install_type)
334
335    if install_type == InstallType.PYTHON:
336        return sorted(
337            conn.name
338            for conn in _get_registry_cache().values()
339            if conn.pypi_package_name is not None
340        )
341
342    if install_type == InstallType.JAVA:
343        warnings.warn(
344            message="Java connectors are not yet supported.",
345            stacklevel=2,
346        )
347        return sorted(
348            conn.name for conn in _get_registry_cache().values() if conn.language == Language.JAVA
349        )
350
351    if install_type == InstallType.DOCKER:
352        return sorted(conn.name for conn in _get_registry_cache().values())
353
354    if install_type == InstallType.YAML:
355        return sorted(
356            conn.name
357            for conn in _get_registry_cache().values()
358            if InstallType.YAML in conn.install_types
359            and conn.name not in _LOWCODE_CONNECTORS_EXCLUDED
360        )
361
362    # pragma: no cover  # Should never be reached.
363    raise exc.PyAirbyteInputError(
364        message="Invalid install type.",
365        context={
366            "install_type": install_type,
367        },
368    )

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

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

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    pip_url: str | None = None,
29    local_executable: Path | str | None = None,
30    docker_image: str | bool | None = None,
31    use_host_network: bool = False,
32    install_if_missing: bool = True,
33) -> Destination:
34    """Get a connector by name and version.
35
36    Args:
37        name: connector name
38        config: connector config - if not provided, you need to set it later via the set_config
39            method.
40        config_change_callback: callback function to be called when the connector config changes.
41        streams: list of stream names to select for reading. If set to "*", all streams will be
42            selected. If not provided, you can set it later via the `select_streams()` or
43            `select_all_streams()` method.
44        version: connector version - if not provided, the currently installed version will be used.
45            If no version is installed, the latest available version will be used. The version can
46            also be set to "latest" to force the use of the latest available version.
47        pip_url: connector pip URL - if not provided, the pip url will be inferred from the
48            connector name.
49        local_executable: If set, the connector will be assumed to already be installed and will be
50            executed using this path or executable name. Otherwise, the connector will be installed
51            automatically in a virtual environment.
52        docker_image: If set, the connector will be executed using Docker. You can specify `True`
53            to use the default image for the connector, or you can specify a custom image name.
54            If `version` is specified and your image name does not already contain a tag
55            (e.g. `my-image:latest`), the version will be appended as a tag (e.g. `my-image:0.1.0`).
56        use_host_network: If set, along with docker_image, the connector will be executed using
57            the host network. This is useful for connectors that need to access resources on
58            the host machine, such as a local database. This parameter is ignored when
59            `docker_image` is not set.
60        install_if_missing: Whether to install the connector if it is not available locally. This
61            parameter is ignored when local_executable is set.
62    """
63    return Destination(
64        name=name,
65        config=config,
66        config_change_callback=config_change_callback,
67        executor=get_connector_executor(
68            name=name,
69            version=version,
70            pip_url=pip_url,
71            local_executable=local_executable,
72            docker_image=docker_image,
73            use_host_network=use_host_network,
74            install_if_missing=install_if_missing,
75        ),
76    )

15def get_secret(
16    secret_name: str,
17    /,
18    *,
19    sources: list[SecretManager | SecretSourceEnum] | None = None,
20    allow_prompt: bool = True,
21    **kwargs: dict[str, Any],
22) -> SecretString:
23    """Get a secret from the environment.
24
25    The optional `sources` argument of enum type `SecretSourceEnum` or list of `SecretSourceEnum`
26    options. If left blank, all available sources will be checked. If a list of `SecretSourceEnum`
27    entries is passed, then the sources will be checked using the provided ordering.
28
29    If `allow_prompt` is `True` or if SecretSourceEnum.PROMPT is declared in the `source` arg, then
30    the user will be prompted to enter the secret if it is not found in any of the other sources.
31    """
32    if "source" in kwargs:
33        warnings.warn(
34            message="The `source` argument is deprecated. Use the `sources` argument instead.",
35            category=DeprecationWarning,
36            stacklevel=2,
37        )
38        sources = kwargs.pop("source")  # type: ignore [assignment]
39
40    available_sources: dict[str, SecretManager] = {}
41    for available_source in _get_secret_sources():
42        # Add available sources to the dict. Order matters.
43        available_sources[available_source.name] = available_source
44
45    if sources is None:
46        # If ANY is in the list, then we don't need to check any other sources.
47        # This is the default behavior.
48        sources = list(available_sources.values())
49
50    elif not isinstance(sources, list):
51        sources = [sources]  # type: ignore [unreachable]  # This is a 'just in case' catch.
52
53    # Replace any SecretSourceEnum strings with the matching SecretManager object
54    for source in list(sources):
55        if isinstance(source, SecretSourceEnum):
56            if source not in available_sources:
57                raise exc.PyAirbyteInputError(
58                    guidance="Invalid secret source name.",
59                    input_value=source,
60                    context={
61                        "Available Sources": list(available_sources.keys()),
62                    },
63                )
64
65            sources[sources.index(source)] = available_sources[source]
66
67    secret_managers = cast(list[SecretManager], sources)
68
69    if SecretSourceEnum.PROMPT in secret_managers:
70        prompt_source = secret_managers.pop(
71            # Mis-typed, but okay here since we have equality logic for the enum comparison:
72            secret_managers.index(SecretSourceEnum.PROMPT),  # type: ignore [arg-type]
73        )
74
75        if allow_prompt:
76            # Always check prompt last. Add it to the end of the list.
77            secret_managers.append(prompt_source)
78
79    for secret_mgr in secret_managers:
80        val = secret_mgr.get_secret(secret_name)
81        if val:
82            return SecretString(val)
83
84    raise exc.PyAirbyteSecretNotFoundError(
85        secret_name=secret_name,
86        sources=[str(s) for s in available_sources],
87    )

 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    pip_url: str | None = None,
 56    local_executable: Path | str | None = None,
 57    docker_image: bool | str | None = None,
 58    use_host_network: bool = False,
 59    source_manifest: bool | dict | Path | str | None = None,
 60    install_if_missing: bool = True,
 61    install_root: Path | None = None,
 62) -> Source:
 63    """Get a connector by name and version.
 64
 65    If an explicit install or execution method is requested (e.g. `local_executable`,
 66    `docker_image`, `pip_url`, `source_manifest`), the connector will be executed using this method.
 67
 68    Otherwise, an appropriate method will be selected based on the available connector metadata:
 69    1. If the connector is registered and has a YAML source manifest is available, the YAML manifest
 70       will be downloaded and used to to execute the connector.
 71    2. Else, if the connector is registered and has a PyPI package, it will be installed via pip.
 72    3. Else, if the connector is registered and has a Docker image, and if Docker is available, it
 73       will be executed using Docker.
 74
 75    Args:
 76        name: connector name
 77        config: connector config - if not provided, you need to set it later via the set_config
 78            method.
 79        config_change_callback: callback function to be called when the connector config changes.
 80        streams: list of stream names to select for reading. If set to "*", all streams will be
 81            selected. If not provided, you can set it later via the `select_streams()` or
 82            `select_all_streams()` method.
 83        version: connector version - if not provided, the currently installed version will be used.
 84            If no version is installed, the latest available version will be used. The version can
 85            also be set to "latest" to force the use of the latest available version.
 86        pip_url: connector pip URL - if not provided, the pip url will be inferred from the
 87            connector name.
 88        local_executable: If set, the connector will be assumed to already be installed and will be
 89            executed using this path or executable name. Otherwise, the connector will be installed
 90            automatically in a virtual environment.
 91        docker_image: If set, the connector will be executed using Docker. You can specify `True`
 92            to use the default image for the connector, or you can specify a custom image name.
 93            If `version` is specified and your image name does not already contain a tag
 94            (e.g. `my-image:latest`), the version will be appended as a tag (e.g. `my-image:0.1.0`).
 95        use_host_network: If set, along with docker_image, the connector will be executed using
 96            the host network. This is useful for connectors that need to access resources on
 97            the host machine, such as a local database. This parameter is ignored when
 98            `docker_image` is not set.
 99        source_manifest: If set, the connector will be executed based on a declarative YAML
100            source definition. This input can be `True` to attempt to auto-download a YAML spec,
101            `dict` to accept a Python dictionary as the manifest, `Path` to pull a manifest from
102            the local file system, or `str` to pull the definition from a web URL.
103        install_if_missing: Whether to install the connector if it is not available locally. This
104            parameter is ignored when `local_executable` or `source_manifest` are set.
105        install_root: (Optional.) The root directory where the virtual environment will be
106            created. If not provided, the current working directory will be used.
107    """
108    return Source(
109        name=name,
110        config=config,
111        config_change_callback=config_change_callback,
112        streams=streams,
113        executor=get_connector_executor(
114            name=name,
115            version=version,
116            pip_url=pip_url,
117            local_executable=local_executable,
118            docker_image=docker_image,
119            use_host_network=use_host_network,
120            source_manifest=source_manifest,
121            install_if_missing=install_if_missing,
122            install_root=install_root,
123        ),
124    )

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

32class BigQueryCache(BigQueryConfig, CacheBase):
33    """The BigQuery cache implementation."""
34
35    _sql_processor_class: type[BigQuerySqlProcessor] = PrivateAttr(default=BigQuerySqlProcessor)
36
37    def get_arrow_dataset(
38        self,
39        stream_name: str,
40        *,
41        max_chunk_size: int = DEFAULT_ARROW_MAX_CHUNK_SIZE,
42    ) -> NoReturn:
43        """Raises NotImplementedError; BigQuery doesn't support `pd.read_sql_table`.
44
45        See: https://github.com/airbytehq/PyAirbyte/issues/165
46        """
47        raise NotImplementedError(
48            "BigQuery doesn't currently support to_arrow"
49            "Please consider using a different cache implementation for these functionalities."
50        )

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

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