222def get_available_connectors(install_type: InstallType | str | None = None) -> list[str]:
223    """Return a list of all available connectors.
224
225    Connectors will be returned in alphabetical order, with the standard prefix "source-".
226    """
227    if install_type is None:
228        # No install type specified. Filter for whatever is runnable.
229        if is_docker_installed():
230            logger.info("Docker is detected. Returning all connectors.")
231            # If Docker is available, return all connectors.
232            return sorted(conn.name for conn in _get_registry_cache().values())
233
234        logger.info("Docker was not detected. Returning only Python and Manifest-only connectors.")
235
236        # If Docker is not available, return only Python and Manifest-based connectors.
237        return sorted(
238            conn.name
239            for conn in _get_registry_cache().values()
240            if conn.language in {Language.PYTHON, Language.MANIFEST_ONLY}
241        )
242
243    if not isinstance(install_type, InstallType):
244        install_type = InstallType(install_type)
245
246    if install_type == InstallType.PYTHON:
247        return sorted(
248            conn.name
249            for conn in _get_registry_cache().values()
250            if conn.pypi_package_name is not None
251        )
252
253    if install_type == InstallType.JAVA:
254        warnings.warn(
255            message="Java connectors are not yet supported.",
256            stacklevel=2,
257        )
258        return sorted(
259            conn.name for conn in _get_registry_cache().values() if conn.language == Language.JAVA
260        )
261
262    if install_type == InstallType.DOCKER:
263        return sorted(conn.name for conn in _get_registry_cache().values())
264
265    if install_type == InstallType.YAML:
266        return sorted(
267            conn.name
268            for conn in _get_registry_cache().values()
269            if InstallType.YAML in conn.install_types
270        )
271
272    # pragma: no cover  # Should never be reached.
273    raise exc.PyAirbyteInputError(
274        message="Invalid install type.",
275        context={
276            "install_type": install_type,
277        },
278    )

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

 33def get_secret(
 34    secret_name: str,
 35    /,
 36    *,
 37    sources: list[SecretManager | SecretSourceEnum] | None = None,
 38    allow_prompt: bool = True,
 39    **kwargs: dict[str, Any],
 40) -> SecretString:
 41    """Get a secret from the environment.
 42
 43    The optional `sources` argument of enum type `SecretSourceEnum` or list of `SecretSourceEnum`
 44    options. If left blank, all available sources will be checked. If a list of `SecretSourceEnum`
 45    entries is passed, then the sources will be checked using the provided ordering.
 46
 47    If `allow_prompt` is `True` or if SecretSourceEnum.PROMPT is declared in the `source` arg, then
 48    the user will be prompted to enter the secret if it is not found in any of the other sources.
 49    """
 50    if secret_name.startswith(SECRETS_HYDRATION_PREFIX):
 51        # If the secret name starts with the hydration prefix, we assume it's a secret reference.
 52        # We strip the prefix and get the actual secret name.
 53        secret_name = secret_name.removeprefix(SECRETS_HYDRATION_PREFIX).lstrip()
 54
 55    if "source" in kwargs:
 56        warnings.warn(
 57            message="The `source` argument is deprecated. Use the `sources` argument instead.",
 58            category=DeprecationWarning,
 59            stacklevel=2,
 60        )
 61        sources = kwargs.pop("source")  # type: ignore [assignment]
 62
 63    available_sources: dict[str, SecretManager] = {}
 64    for available_source in _get_secret_sources():
 65        # Add available sources to the dict. Order matters.
 66        available_sources[available_source.name] = available_source
 67
 68    if sources is None:
 69        # If ANY is in the list, then we don't need to check any other sources.
 70        # This is the default behavior.
 71        sources = list(available_sources.values())
 72
 73    elif not isinstance(sources, list):
 74        sources = [sources]  # type: ignore [unreachable]  # This is a 'just in case' catch.
 75
 76    # Replace any SecretSourceEnum strings with the matching SecretManager object
 77    for source in list(sources):
 78        if isinstance(source, SecretSourceEnum):
 79            if source not in available_sources:
 80                raise exc.PyAirbyteInputError(
 81                    guidance="Invalid secret source name.",
 82                    input_value=source,
 83                    context={
 84                        "Available Sources": list(available_sources.keys()),
 85                    },
 86                )
 87
 88            sources[sources.index(source)] = available_sources[source]
 89
 90    secret_managers = cast("list[SecretManager]", sources)
 91
 92    if SecretSourceEnum.PROMPT in secret_managers:
 93        prompt_source = secret_managers.pop(
 94            # Mis-typed, but okay here since we have equality logic for the enum comparison:
 95            secret_managers.index(SecretSourceEnum.PROMPT),  # type: ignore [arg-type]
 96        )
 97
 98        if allow_prompt:
 99            # Always check prompt last. Add it to the end of the list.
100            secret_managers.append(prompt_source)
101
102    for secret_mgr in secret_managers:
103        val = secret_mgr.get_secret(secret_name)
104        if val:
105            return SecretString(val)
106
107    raise exc.PyAirbyteSecretNotFoundError(
108        secret_name=secret_name,
109        sources=[str(s) for s in available_sources],
110    )

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

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    )

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)

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