airbyte.sources.base

  1# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
  2"""Base class implementation for sources."""
  3
  4from __future__ import annotations
  5
  6import sys
  7import warnings
  8from typing import TYPE_CHECKING, Any, Literal
  9
 10import yaml
 11from rich import print  # noqa: A004  # Allow shadowing the built-in
 12
 13from airbyte_protocol.models import (
 14    AirbyteCatalog,
 15    AirbyteMessage,
 16    ConfiguredAirbyteCatalog,
 17    ConfiguredAirbyteStream,
 18    DestinationSyncMode,
 19    SyncMode,
 20    Type,
 21)
 22
 23from airbyte import exceptions as exc
 24from airbyte._connector_base import ConnectorBase
 25from airbyte._message_iterators import AirbyteMessageIterator
 26from airbyte._util.temp_files import as_temp_files
 27from airbyte.caches.util import get_default_cache
 28from airbyte.datasets._lazy import LazyDataset
 29from airbyte.progress import ProgressStyle, ProgressTracker
 30from airbyte.records import StreamRecord, StreamRecordHandler
 31from airbyte.results import ReadResult
 32from airbyte.shared.catalog_providers import CatalogProvider
 33from airbyte.strategies import WriteStrategy
 34
 35
 36if TYPE_CHECKING:
 37    from collections.abc import Generator, Iterable, Iterator
 38
 39    from airbyte_cdk import ConnectorSpecification
 40    from airbyte_protocol.models import AirbyteStream
 41
 42    from airbyte._executors.base import Executor
 43    from airbyte.caches import CacheBase
 44    from airbyte.callbacks import ConfigChangeCallback
 45    from airbyte.documents import Document
 46    from airbyte.shared.state_providers import StateProviderBase
 47    from airbyte.shared.state_writers import StateWriterBase
 48
 49
 50class Source(ConnectorBase):
 51    """A class representing a source that can be called."""
 52
 53    connector_type = "source"
 54
 55    def __init__(
 56        self,
 57        executor: Executor,
 58        name: str,
 59        config: dict[str, Any] | None = None,
 60        *,
 61        config_change_callback: ConfigChangeCallback | None = None,
 62        streams: str | list[str] | None = None,
 63        validate: bool = False,
 64        cursor_key_overrides: dict[str, str] | None = None,
 65        primary_key_overrides: dict[str, str | list[str]] | None = None,
 66    ) -> None:
 67        """Initialize the source.
 68
 69        If config is provided, it will be validated against the spec if validate is True.
 70        """
 71        self._to_be_selected_streams: list[str] | str = []
 72        """Used to hold selection criteria before catalog is known."""
 73
 74        super().__init__(
 75            executor=executor,
 76            name=name,
 77            config=config,
 78            config_change_callback=config_change_callback,
 79            validate=validate,
 80        )
 81        self._config_dict: dict[str, Any] | None = None
 82        self._last_log_messages: list[str] = []
 83        self._discovered_catalog: AirbyteCatalog | None = None
 84        self._selected_stream_names: list[str] = []
 85
 86        self._cursor_key_overrides: dict[str, str] = {}
 87        """A mapping of lower-cased stream names to cursor key overrides."""
 88
 89        self._primary_key_overrides: dict[str, list[str]] = {}
 90        """A mapping of lower-cased stream names to primary key overrides."""
 91
 92        if config is not None:
 93            self.set_config(config, validate=validate)
 94        if streams is not None:
 95            self.select_streams(streams)
 96        if cursor_key_overrides is not None:
 97            self.set_cursor_keys(**cursor_key_overrides)
 98        if primary_key_overrides is not None:
 99            self.set_primary_keys(**primary_key_overrides)
100
101    def set_streams(self, streams: list[str]) -> None:
102        """Deprecated. See select_streams()."""
103        warnings.warn(
104            "The 'set_streams' method is deprecated and will be removed in a future version. "
105            "Please use the 'select_streams' method instead.",
106            DeprecationWarning,
107            stacklevel=2,
108        )
109        self.select_streams(streams)
110
111    def set_cursor_key(
112        self,
113        stream_name: str,
114        cursor_key: str,
115    ) -> None:
116        """Set the cursor for a single stream.
117
118        Note:
119        - This does not unset previously set cursors.
120        - The cursor key must be a single field name.
121        - Not all streams support custom cursors. If a stream does not support custom cursors,
122          the override may be ignored.
123        - Stream names are case insensitive, while field names are case sensitive.
124        - Stream names are not validated by PyAirbyte. If the stream name
125          does not exist in the catalog, the override may be ignored.
126        """
127        self._cursor_key_overrides[stream_name.lower()] = cursor_key
128
129    def set_cursor_keys(
130        self,
131        **kwargs: str,
132    ) -> None:
133        """Override the cursor key for one or more streams.
134
135        Usage:
136            source.set_cursor_keys(
137                stream1="cursor1",
138                stream2="cursor2",
139            )
140
141        Note:
142        - This does not unset previously set cursors.
143        - The cursor key must be a single field name.
144        - Not all streams support custom cursors. If a stream does not support custom cursors,
145          the override may be ignored.
146        - Stream names are case insensitive, while field names are case sensitive.
147        - Stream names are not validated by PyAirbyte. If the stream name
148          does not exist in the catalog, the override may be ignored.
149        """
150        self._cursor_key_overrides.update({k.lower(): v for k, v in kwargs.items()})
151
152    def set_primary_key(
153        self,
154        stream_name: str,
155        primary_key: str | list[str],
156    ) -> None:
157        """Set the primary key for a single stream.
158
159        Note:
160        - This does not unset previously set primary keys.
161        - The primary key must be a single field name or a list of field names.
162        - Not all streams support overriding primary keys. If a stream does not support overriding
163          primary keys, the override may be ignored.
164        - Stream names are case insensitive, while field names are case sensitive.
165        - Stream names are not validated by PyAirbyte. If the stream name
166          does not exist in the catalog, the override may be ignored.
167        """
168        self._primary_key_overrides[stream_name.lower()] = (
169            primary_key if isinstance(primary_key, list) else [primary_key]
170        )
171
172    def set_primary_keys(
173        self,
174        **kwargs: str | list[str],
175    ) -> None:
176        """Override the primary keys for one or more streams.
177
178        This does not unset previously set primary keys.
179
180        Usage:
181            source.set_primary_keys(
182                stream1="pk1",
183                stream2=["pk1", "pk2"],
184            )
185
186        Note:
187        - This does not unset previously set primary keys.
188        - The primary key must be a single field name or a list of field names.
189        - Not all streams support overriding primary keys. If a stream does not support overriding
190          primary keys, the override may be ignored.
191        - Stream names are case insensitive, while field names are case sensitive.
192        - Stream names are not validated by PyAirbyte. If the stream name
193          does not exist in the catalog, the override may be ignored.
194        """
195        self._primary_key_overrides.update(
196            {k.lower(): v if isinstance(v, list) else [v] for k, v in kwargs.items()}
197        )
198
199    def _log_warning_preselected_stream(self, streams: str | list[str]) -> None:
200        """Logs a warning message indicating stream selection which are not selected yet."""
201        if streams == "*":
202            print(
203                "Warning: Config is not set yet. All streams will be selected after config is set.",
204                file=sys.stderr,
205            )
206        else:
207            print(
208                "Warning: Config is not set yet. "
209                f"Streams to be selected after config is set: {streams}",
210                file=sys.stderr,
211            )
212
213    def select_all_streams(self) -> None:
214        """Select all streams.
215
216        This is a more streamlined equivalent to:
217        > source.select_streams(source.get_available_streams()).
218        """
219        if self._config_dict is None:
220            self._to_be_selected_streams = "*"
221            self._log_warning_preselected_stream(self._to_be_selected_streams)
222            return
223
224        self._selected_stream_names = self.get_available_streams()
225
226    def select_streams(self, streams: str | list[str]) -> None:
227        """Select the stream names that should be read from the connector.
228
229        Args:
230            streams: A list of stream names to select. If set to "*", all streams will be selected.
231
232        Currently, if this is not set, all streams will be read.
233        """
234        if self._config_dict is None:
235            self._to_be_selected_streams = streams
236            self._log_warning_preselected_stream(streams)
237            return
238
239        if streams == "*":
240            self.select_all_streams()
241            return
242
243        if isinstance(streams, str):
244            # If a single stream is provided, convert it to a one-item list
245            streams = [streams]
246
247        available_streams = self.get_available_streams()
248        for stream in streams:
249            if stream not in available_streams:
250                raise exc.AirbyteStreamNotFoundError(
251                    stream_name=stream,
252                    connector_name=self.name,
253                    available_streams=available_streams,
254                )
255        self._selected_stream_names = streams
256
257    def get_selected_streams(self) -> list[str]:
258        """Get the selected streams.
259
260        If no streams are selected, return an empty list.
261        """
262        return self._selected_stream_names
263
264    def set_config(
265        self,
266        config: dict[str, Any],
267        *,
268        validate: bool = True,
269    ) -> None:
270        """Set the config for the connector.
271
272        If validate is True, raise an exception if the config fails validation.
273
274        If validate is False, validation will be deferred until check() or validate_config()
275        is called.
276        """
277        if validate:
278            self.validate_config(config)
279
280        self._config_dict = config
281
282        if self._to_be_selected_streams:
283            self.select_streams(self._to_be_selected_streams)
284            self._to_be_selected_streams = []
285
286    def _discover(self) -> AirbyteCatalog:
287        """Call discover on the connector.
288
289        This involves the following steps:
290        - Write the config to a temporary file
291        - execute the connector with discover --config <config_file>
292        - Listen to the messages and return the first AirbyteCatalog that comes along.
293        - Make sure the subprocess is killed when the function returns.
294        """
295        with as_temp_files([self._hydrated_config]) as [config_file]:
296            for msg in self._execute(["discover", "--config", config_file]):
297                if msg.type == Type.CATALOG and msg.catalog:
298                    return msg.catalog
299            raise exc.AirbyteConnectorMissingCatalogError(
300                connector_name=self.name,
301                log_text=self._last_log_messages,
302            )
303
304    def get_available_streams(self) -> list[str]:
305        """Get the available streams from the spec."""
306        return [s.name for s in self.discovered_catalog.streams]
307
308    def _get_incremental_stream_names(self) -> list[str]:
309        """Get the name of streams that support incremental sync."""
310        return [
311            stream.name
312            for stream in self.discovered_catalog.streams
313            if SyncMode.incremental in stream.supported_sync_modes
314        ]
315
316    def _get_spec(self, *, force_refresh: bool = False) -> ConnectorSpecification:
317        """Call spec on the connector.
318
319        This involves the following steps:
320        * execute the connector with spec
321        * Listen to the messages and return the first AirbyteCatalog that comes along.
322        * Make sure the subprocess is killed when the function returns.
323        """
324        if force_refresh or self._spec is None:
325            for msg in self._execute(["spec"]):
326                if msg.type == Type.SPEC and msg.spec:
327                    self._spec = msg.spec
328                    break
329
330        if self._spec:
331            return self._spec
332
333        raise exc.AirbyteConnectorMissingSpecError(
334            connector_name=self.name,
335            log_text=self._last_log_messages,
336        )
337
338    @property
339    def config_spec(self) -> dict[str, Any]:
340        """Generate a configuration spec for this connector, as a JSON Schema definition.
341
342        This function generates a JSON Schema dictionary with configuration specs for the
343        current connector, as a dictionary.
344
345        Returns:
346            dict: The JSON Schema configuration spec as a dictionary.
347        """
348        return self._get_spec(force_refresh=True).connectionSpecification
349
350    @property
351    def _yaml_spec(self) -> str:
352        """Get the spec as a yaml string.
353
354        For now, the primary use case is for writing and debugging a valid config for a source.
355
356        This is private for now because we probably want better polish before exposing this
357        as a stable interface. This will also get easier when we have docs links with this info
358        for each connector.
359        """
360        spec_obj: ConnectorSpecification = self._get_spec()
361        spec_dict: dict[str, Any] = spec_obj.model_dump(exclude_unset=True)
362        # convert to a yaml string
363        return yaml.dump(spec_dict)
364
365    @property
366    def docs_url(self) -> str:
367        """Get the URL to the connector's documentation."""
368        return "https://docs.airbyte.com/integrations/sources/" + self.name.lower().replace(
369            "source-", ""
370        )
371
372    @property
373    def discovered_catalog(self) -> AirbyteCatalog:
374        """Get the raw catalog for the given streams.
375
376        If the catalog is not yet known, we call discover to get it.
377        """
378        if self._discovered_catalog is None:
379            self._discovered_catalog = self._discover()
380
381        return self._discovered_catalog
382
383    @property
384    def configured_catalog(self) -> ConfiguredAirbyteCatalog:
385        """Get the configured catalog for the given streams.
386
387        If the raw catalog is not yet known, we call discover to get it.
388
389        If no specific streams are selected, we return a catalog that syncs all available streams.
390
391        TODO: We should consider disabling by default the streams that the connector would
392        disable by default. (For instance, streams that require a premium license are sometimes
393        disabled by default within the connector.)
394        """
395        # Ensure discovered catalog is cached before we start
396        _ = self.discovered_catalog
397
398        # Filter for selected streams if set, otherwise use all available streams:
399        streams_filter: list[str] = self._selected_stream_names or self.get_available_streams()
400        return self.get_configured_catalog(streams=streams_filter)
401
402    def get_configured_catalog(
403        self,
404        streams: Literal["*"] | list[str] | None = None,
405    ) -> ConfiguredAirbyteCatalog:
406        """Get a configured catalog for the given streams.
407
408        If no streams are provided, the selected streams will be used. If no streams are selected,
409        all available streams will be used.
410
411        If '*' is provided, all available streams will be used.
412        """
413        selected_streams: list[str] = []
414        if streams is None:
415            selected_streams = self._selected_stream_names or self.get_available_streams()
416        elif streams == "*":
417            selected_streams = self.get_available_streams()
418        elif isinstance(streams, list):
419            selected_streams = streams
420        else:
421            raise exc.PyAirbyteInputError(
422                message="Invalid streams argument.",
423                input_value=streams,
424            )
425
426        return ConfiguredAirbyteCatalog(
427            streams=[
428                ConfiguredAirbyteStream(
429                    stream=stream,
430                    destination_sync_mode=DestinationSyncMode.overwrite,
431                    sync_mode=SyncMode.incremental,
432                    primary_key=(
433                        [self._primary_key_overrides[stream.name.lower()]]
434                        if stream.name.lower() in self._primary_key_overrides
435                        else stream.source_defined_primary_key
436                    ),
437                    cursor_field=(
438                        [self._cursor_key_overrides[stream.name.lower()]]
439                        if stream.name.lower() in self._cursor_key_overrides
440                        else stream.default_cursor_field
441                    ),
442                    # These are unused in the current implementation:
443                    generation_id=None,
444                    minimum_generation_id=None,
445                    sync_id=None,
446                )
447                for stream in self.discovered_catalog.streams
448                if stream.name in selected_streams
449            ],
450        )
451
452    def get_stream_json_schema(self, stream_name: str) -> dict[str, Any]:
453        """Return the JSON Schema spec for the specified stream name."""
454        catalog: AirbyteCatalog = self.discovered_catalog
455        found: list[AirbyteStream] = [
456            stream for stream in catalog.streams if stream.name == stream_name
457        ]
458
459        if len(found) == 0:
460            raise exc.PyAirbyteInputError(
461                message="Stream name does not exist in catalog.",
462                input_value=stream_name,
463            )
464
465        if len(found) > 1:
466            raise exc.PyAirbyteInternalError(
467                message="Duplicate streams found with the same name.",
468                context={
469                    "found_streams": found,
470                },
471            )
472
473        return found[0].json_schema
474
475    def get_records(
476        self,
477        stream: str,
478        *,
479        normalize_field_names: bool = False,
480        prune_undeclared_fields: bool = True,
481    ) -> LazyDataset:
482        """Read a stream from the connector.
483
484        Args:
485            stream: The name of the stream to read.
486            normalize_field_names: When `True`, field names will be normalized to lower case, with
487                special characters removed. This matches the behavior of PyAirbyte caches and most
488                Airbyte destinations.
489            prune_undeclared_fields: When `True`, undeclared fields will be pruned from the records,
490                which generally matches the behavior of PyAirbyte caches and most Airbyte
491                destinations, specifically when you expect the catalog may be stale. You can disable
492                this to keep all fields in the records.
493
494        This involves the following steps:
495        * Call discover to get the catalog
496        * Generate a configured catalog that syncs the given stream in full_refresh mode
497        * Write the configured catalog and the config to a temporary file
498        * execute the connector with read --config <config_file> --catalog <catalog_file>
499        * Listen to the messages and return the first AirbyteRecordMessages that come along.
500        * Make sure the subprocess is killed when the function returns.
501        """
502        configured_catalog = self.get_configured_catalog(streams=[stream])
503        if len(configured_catalog.streams) == 0:
504            raise exc.PyAirbyteInputError(
505                message="Requested stream does not exist.",
506                context={
507                    "stream": stream,
508                    "available_streams": self.get_available_streams(),
509                    "connector_name": self.name,
510                },
511            ) from KeyError(stream)
512
513        configured_stream = configured_catalog.streams[0]
514
515        def _with_logging(records: Iterable[dict[str, Any]]) -> Iterator[dict[str, Any]]:
516            yield from records
517
518        stream_record_handler = StreamRecordHandler(
519            json_schema=self.get_stream_json_schema(stream),
520            prune_extra_fields=prune_undeclared_fields,
521            normalize_keys=normalize_field_names,
522        )
523
524        # This method is non-blocking, so we use "PLAIN" to avoid a live progress display
525        progress_tracker = ProgressTracker(
526            ProgressStyle.PLAIN,
527            source=self,
528            cache=None,
529            destination=None,
530            expected_streams=[stream],
531        )
532
533        iterator: Iterator[dict[str, Any]] = (
534            StreamRecord.from_record_message(
535                record_message=record.record,
536                stream_record_handler=stream_record_handler,
537            )
538            for record in self._read_with_catalog(
539                catalog=configured_catalog,
540                progress_tracker=progress_tracker,
541            )
542            if record.record
543        )
544        progress_tracker.log_success()
545        return LazyDataset(
546            iterator,
547            stream_metadata=configured_stream,
548        )
549
550    def get_documents(
551        self,
552        stream: str,
553        title_property: str | None = None,
554        content_properties: list[str] | None = None,
555        metadata_properties: list[str] | None = None,
556        *,
557        render_metadata: bool = False,
558    ) -> Iterable[Document]:
559        """Read a stream from the connector and return the records as documents.
560
561        If metadata_properties is not set, all properties that are not content will be added to
562        the metadata.
563
564        If render_metadata is True, metadata will be rendered in the document, as well as the
565        the main content.
566        """
567        return self.get_records(stream).to_documents(
568            title_property=title_property,
569            content_properties=content_properties,
570            metadata_properties=metadata_properties,
571            render_metadata=render_metadata,
572        )
573
574    def _get_airbyte_message_iterator(
575        self,
576        *,
577        streams: Literal["*"] | list[str] | None = None,
578        state_provider: StateProviderBase | None = None,
579        progress_tracker: ProgressTracker,
580        force_full_refresh: bool = False,
581    ) -> AirbyteMessageIterator:
582        """Get an AirbyteMessageIterator for this source."""
583        return AirbyteMessageIterator(
584            self._read_with_catalog(
585                catalog=self.get_configured_catalog(streams=streams),
586                state=state_provider if not force_full_refresh else None,
587                progress_tracker=progress_tracker,
588            )
589        )
590
591    def _read_with_catalog(
592        self,
593        catalog: ConfiguredAirbyteCatalog,
594        progress_tracker: ProgressTracker,
595        state: StateProviderBase | None = None,
596    ) -> Generator[AirbyteMessage, None, None]:
597        """Call read on the connector.
598
599        This involves the following steps:
600        * Write the config to a temporary file
601        * execute the connector with read --config <config_file> --catalog <catalog_file>
602        * Listen to the messages and return the AirbyteRecordMessages that come along.
603        * Send out telemetry on the performed sync (with information about which source was used and
604          the type of the cache)
605        """
606        with as_temp_files(
607            [
608                self._hydrated_config,
609                catalog.model_dump_json(),
610                state.to_state_input_file_text() if state else "[]",
611            ]
612        ) as [
613            config_file,
614            catalog_file,
615            state_file,
616        ]:
617            message_generator = self._execute(
618                [
619                    "read",
620                    "--config",
621                    config_file,
622                    "--catalog",
623                    catalog_file,
624                    "--state",
625                    state_file,
626                ],
627                progress_tracker=progress_tracker,
628            )
629            yield from progress_tracker.tally_records_read(message_generator)
630        progress_tracker.log_read_complete()
631
632    def _peek_airbyte_message(
633        self,
634        message: AirbyteMessage,
635        *,
636        raise_on_error: bool = True,
637    ) -> None:
638        """Process an Airbyte message.
639
640        This method handles reading Airbyte messages and taking action, if needed, based on the
641        message type. For instance, log messages are logged, records are tallied, and errors are
642        raised as exceptions if `raise_on_error` is True.
643
644        Raises:
645            AirbyteConnectorFailedError: If a TRACE message of type ERROR is emitted.
646        """
647        super()._peek_airbyte_message(message, raise_on_error=raise_on_error)
648
649    def _log_incremental_streams(
650        self,
651        *,
652        incremental_streams: set[str] | None = None,
653    ) -> None:
654        """Log the streams which are using incremental sync mode."""
655        log_message = (
656            "The following streams are currently using incremental sync:\n"
657            f"{incremental_streams}\n"
658            "To perform a full refresh, set 'force_full_refresh=True' in 'airbyte.read()' method."
659        )
660        print(log_message, file=sys.stderr)
661
662    def read(
663        self,
664        cache: CacheBase | None = None,
665        *,
666        streams: str | list[str] | None = None,
667        write_strategy: str | WriteStrategy = WriteStrategy.AUTO,
668        force_full_refresh: bool = False,
669        skip_validation: bool = False,
670    ) -> ReadResult:
671        """Read from the connector and write to the cache.
672
673        Args:
674            cache: The cache to write to. If not set, a default cache will be used.
675            streams: Optional if already set. A list of stream names to select for reading. If set
676                to "*", all streams will be selected.
677            write_strategy: The strategy to use when writing to the cache. If a string, it must be
678                one of "append", "merge", "replace", or "auto". If a WriteStrategy, it must be one
679                of WriteStrategy.APPEND, WriteStrategy.MERGE, WriteStrategy.REPLACE, or
680                WriteStrategy.AUTO.
681            force_full_refresh: If True, the source will operate in full refresh mode. Otherwise,
682                streams will be read in incremental mode if supported by the connector. This option
683                must be True when using the "replace" strategy.
684            skip_validation: If True, PyAirbyte will not pre-validate the input configuration before
685                running the connector. This can be helpful in debugging, when you want to send
686                configurations to the connector that otherwise might be rejected by JSON Schema
687                validation rules.
688        """
689        cache = cache or get_default_cache()
690        progress_tracker = ProgressTracker(
691            source=self,
692            cache=cache,
693            destination=None,
694            expected_streams=None,  # Will be set later
695        )
696
697        # Set up state provider if not in full refresh mode
698        if force_full_refresh:
699            state_provider: StateProviderBase | None = None
700        else:
701            state_provider = cache.get_state_provider(
702                source_name=self._name,
703            )
704        state_writer = cache.get_state_writer(source_name=self._name)
705
706        if streams:
707            self.select_streams(streams)
708
709        if not self._selected_stream_names:
710            raise exc.PyAirbyteNoStreamsSelectedError(
711                connector_name=self.name,
712                available_streams=self.get_available_streams(),
713            )
714
715        try:
716            result = self._read_to_cache(
717                cache=cache,
718                catalog_provider=CatalogProvider(self.configured_catalog),
719                stream_names=self._selected_stream_names,
720                state_provider=state_provider,
721                state_writer=state_writer,
722                write_strategy=write_strategy,
723                force_full_refresh=force_full_refresh,
724                skip_validation=skip_validation,
725                progress_tracker=progress_tracker,
726            )
727        except exc.PyAirbyteInternalError as ex:
728            progress_tracker.log_failure(exception=ex)
729            raise exc.AirbyteConnectorFailedError(
730                connector_name=self.name,
731                log_text=self._last_log_messages,
732            ) from ex
733        except Exception as ex:
734            progress_tracker.log_failure(exception=ex)
735            raise
736
737        progress_tracker.log_success()
738        return result
739
740    def _read_to_cache(  # noqa: PLR0913  # Too many arguments
741        self,
742        cache: CacheBase,
743        *,
744        catalog_provider: CatalogProvider,
745        stream_names: list[str],
746        state_provider: StateProviderBase | None,
747        state_writer: StateWriterBase | None,
748        write_strategy: str | WriteStrategy = WriteStrategy.AUTO,
749        force_full_refresh: bool = False,
750        skip_validation: bool = False,
751        progress_tracker: ProgressTracker,
752    ) -> ReadResult:
753        """Internal read method."""
754        if write_strategy == WriteStrategy.REPLACE and not force_full_refresh:
755            warnings.warn(
756                message=(
757                    "Using `REPLACE` strategy without also setting `full_refresh_mode=True` "
758                    "could result in data loss. "
759                    "To silence this warning, use the following: "
760                    'warnings.filterwarnings("ignore", '
761                    'category="airbyte.warnings.PyAirbyteDataLossWarning")`'
762                ),
763                category=exc.PyAirbyteDataLossWarning,
764                stacklevel=1,
765            )
766        if isinstance(write_strategy, str):
767            try:
768                write_strategy = WriteStrategy(write_strategy)
769            except ValueError:
770                raise exc.PyAirbyteInputError(
771                    message="Invalid strategy",
772                    context={
773                        "write_strategy": write_strategy,
774                        "available_strategies": [s.value for s in WriteStrategy],
775                    },
776                ) from None
777
778        # Run optional validation step
779        if not skip_validation:
780            self.validate_config()
781
782        # Log incremental stream if incremental streams are known
783        if state_provider and state_provider.known_stream_names:
784            # Retrieve set of the known streams support which support incremental sync
785            incremental_streams = (
786                set(self._get_incremental_stream_names())
787                & state_provider.known_stream_names
788                & set(self.get_selected_streams())
789            )
790            if incremental_streams:
791                self._log_incremental_streams(incremental_streams=incremental_streams)
792
793        airbyte_message_iterator = AirbyteMessageIterator(
794            self._read_with_catalog(
795                catalog=catalog_provider.configured_catalog,
796                state=state_provider,
797                progress_tracker=progress_tracker,
798            )
799        )
800        cache._write_airbyte_message_stream(  # noqa: SLF001  # Non-public API
801            stdin=airbyte_message_iterator,
802            catalog_provider=catalog_provider,
803            write_strategy=write_strategy,
804            state_writer=state_writer,
805            progress_tracker=progress_tracker,
806        )
807
808        # Flush the WAL, if applicable
809        cache.processor._do_checkpoint()  # noqa: SLF001  # Non-public API
810
811        return ReadResult(
812            source_name=self.name,
813            progress_tracker=progress_tracker,
814            processed_streams=stream_names,
815            cache=cache,
816        )
817
818
819__all__ = [
820    "Source",
821]