Merge branch 'main' into dbgold17/provide-additional-context-to-async-job-steps

Author: dbgold17

airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py

@@ -189,6 +189,7 @@ def __init__(
         # FIXME this is a temporary field the time of the migration from declarative cursors to concurrent ones
         self._attempt_to_create_cursor_if_not_provided = attempt_to_create_cursor_if_not_provided
         self._synced_some_data = False
+        self._logged_regarding_datetime_format_error = False
 
     @property
     def cursor_field(self) -> CursorField:
@@ -518,10 +519,23 @@ def observe(self, record: Record) -> None:
         except ValueError:
             return
 
+        try:
+            record_cursor = self._connector_state_converter.output_format(
+                self._connector_state_converter.parse_value(record_cursor_value)
+            )
+        except ValueError as exception:
+            if not self._logged_regarding_datetime_format_error:
+                logger.warning(
+                    "Skipping cursor update for stream '%s': failed to parse cursor field '%s' value %r: %s",
+                    self._stream_name,
+                    self._cursor_field.cursor_field_key,
+                    record_cursor_value,
+                    exception,
+                )
+                self._logged_regarding_datetime_format_error = True
+            return
+
         self._synced_some_data = True
-        record_cursor = self._connector_state_converter.output_format(
-            self._connector_state_converter.parse_value(record_cursor_value)
-        )
         self._update_global_cursor(record_cursor)
         if not self._use_global_cursor:
             self._cursor_per_partition[

debug_manifest/README.md

@@ -31,6 +31,10 @@ To configure the debugger in VSCode to run the `debug_manifest`, follow these st
         "--manifest-path",
         // PATH TO THE MANIFEST FILE
         "resources/manifest.yaml",
+        // SPECIFY A COMPONENTS.PY FILE (OPTIONAL)
+        "--components-path",
+        // PATH TO THE COMPONENTS FILE
+        "resources/components.py",
         // SPECIFY THE CONFIG
         "--config",
         // PATH TO THE CONFIG FILE

debug_manifest/debug_manifest.py

@@ -17,10 +17,49 @@ def debug_manifest(source: YamlDeclarativeSource, args: list[str]) -> None:
     launch(source, args)
 
 
+def _register_components_from_file(filepath: str) -> None:
+    """
+    Dynamically load a Python file containing custom component definitions and register it
+    under specific module names in sys.modules to ensure that these classes can be properly
+    resolved during hydration of the manifest yaml file.
+
+    This is a somewhat hacky replacement for the file structure manipulation we do when building
+    connector images to ensure the custom components can be imported.
+    """
+    import importlib.util
+    import sys
+    from pathlib import Path
+
+    components_path = Path(filepath)
+    if not components_path.exists():
+        raise FileNotFoundError(f"Components file not found: {components_path}")
+
+    module_name = "components"
+    sdm_module_name = "source_declarative_manifest.components"
+
+    spec = importlib.util.spec_from_file_location(module_name, components_path)
+    if spec is None or spec.loader is None:
+        raise ImportError(f"Could not load module from {components_path}")
+
+    # Create module and execute code
+    module = importlib.util.module_from_spec(spec)
+
+    # Register then execute the module
+    # we dual-register the module to mirror what is done elsewhere in the CDK
+    sys.modules[module_name] = module
+    sys.modules[sdm_module_name] = module
+
+    spec.loader.exec_module(module)
+
+
 if __name__ == "__main__":
     args = sys.argv[1:]
     parsed_args = AirbyteEntrypoint.parse_args(args)
+
     manifest_path = getattr(parsed_args, "manifest_path", None) or "resources/manifest.yaml"
+    components_path = getattr(parsed_args, "components_path", None)
+    if components_path:
+        _register_components_from_file(components_path)
     catalog_path = AirbyteEntrypoint.extract_catalog(args)
     config_path = AirbyteEntrypoint.extract_config(args)
     state_path = AirbyteEntrypoint.extract_state(args)

unit_tests/sources/declarative/incremental/test_concurrent_perpartitioncursor.py

@@ -20,6 +20,7 @@
     ConcurrentDeclarativeSource,
 )
 from airbyte_cdk.sources.declarative.incremental import ConcurrentPerPartitionCursor
+from airbyte_cdk.sources.declarative.partition_routers import ListPartitionRouter
 from airbyte_cdk.sources.declarative.schema import InlineSchemaLoader
 from airbyte_cdk.sources.declarative.stream_slicers.declarative_partition_generator import (
     DeclarativePartition,
@@ -29,7 +30,7 @@
 from airbyte_cdk.sources.streams.concurrent.state_converters.datetime_stream_state_converter import (
     CustomFormatConcurrentStreamStateConverter,
 )
-from airbyte_cdk.sources.types import StreamSlice
+from airbyte_cdk.sources.types import Record, StreamSlice
 from airbyte_cdk.test.catalog_builder import CatalogBuilder, ConfiguredAirbyteStreamBuilder
 from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput, read
 
@@ -4400,3 +4401,34 @@ def test_duplicate_partition_while_processing():
     assert len(cursor._processing_partitions_indexes) == 0
     assert len(cursor._partition_key_to_index) == 0
     assert len(cursor._partitions_done_generating_stream_slices) == 0
+
+
+def test_given_record_with_bad_cursor_value_the_global_state_parsing_does_not_break_sync():
+    cursor_factory_mock = MagicMock()
+    cursor_factory_mock.create.side_effect = [_make_inner_cursor("2024-01-01T00:00:00Z")]
+    cursor = ConcurrentPerPartitionCursor(
+        cursor_factory=MagicMock(),
+        partition_router=ListPartitionRouter(
+            values=["1"], cursor_field="partition_id", config={}, parameters={}
+        ),
+        stream_name="test_stream",
+        stream_namespace=None,
+        stream_state={},
+        message_repository=MagicMock(),
+        connector_state_manager=MagicMock(),
+        connector_state_converter=CustomFormatConcurrentStreamStateConverter(
+            datetime_format="%Y-%m-%dT%H:%M:%SZ",
+            input_datetime_formats=["%Y-%m-%dT%H:%M:%SZ"],
+            is_sequential_state=True,
+            cursor_granularity=timedelta(0),
+        ),
+        cursor_field=CursorField(cursor_field_key="updated_at"),
+    )
+
+    cursor.observe(
+        Record(
+            data={"updated_at": ""},
+            stream_name="test_stream",
+            associated_slice=StreamSlice(partition={"partition_id": "1"}, cursor_slice={}),
+        )
+    )