diff --git a/airbyte_cdk/legacy/sources/declarative/incremental/datetime_based_cursor.py b/airbyte_cdk/legacy/sources/declarative/incremental/datetime_based_cursor.py index 616a13d8cf..9650f3dc97 100644 --- a/airbyte_cdk/legacy/sources/declarative/incremental/datetime_based_cursor.py +++ b/airbyte_cdk/legacy/sources/declarative/incremental/datetime_based_cursor.py @@ -11,7 +11,10 @@ from airbyte_cdk.legacy.sources.declarative.incremental.declarative_cursor import DeclarativeCursor from airbyte_cdk.models import AirbyteLogMessage, AirbyteMessage, Level, Type -from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeParser +from airbyte_cdk.sources.declarative.datetime.datetime_parser import ( + DatetimeFormatMismatchError, + DatetimeParser, +) from airbyte_cdk.sources.declarative.datetime.min_max_datetime import MinMaxDatetime from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString from airbyte_cdk.sources.declarative.interpolation.jinja import JinjaInterpolation @@ -308,12 +311,19 @@ def _get_date( return comparator(cursor_date, default_date) def parse_date(self, date: str) -> datetime.datetime: - for datetime_format in self.cursor_datetime_formats + [self.datetime_format]: + formats = self.cursor_datetime_formats + [self.datetime_format] + last_error: Optional[ValueError] = None + for datetime_format in formats: try: return self._parser.parse(date, datetime_format) - except ValueError: - pass - raise ValueError(f"No format in {self.cursor_datetime_formats} matching {date}") + except ValueError as error: + last_error = error + raise DatetimeFormatMismatchError( + value=date, + formats=formats, + cursor_field=self.cursor_field.eval(self.config), # type: ignore # cursor_field is converted to an InterpolatedString in __post_init__ + original_error=last_error, + ) @classmethod def _parse_timedelta(cls, time_str: Optional[str]) -> Union[datetime.timedelta, Duration]: diff --git a/airbyte_cdk/sources/declarative/datetime/datetime_parser.py b/airbyte_cdk/sources/declarative/datetime/datetime_parser.py index 2707ffe114..89d7b01d0f 100644 --- a/airbyte_cdk/sources/declarative/datetime/datetime_parser.py +++ b/airbyte_cdk/sources/declarative/datetime/datetime_parser.py @@ -3,7 +3,66 @@ # import datetime -from typing import Union +from typing import List, Optional, Union + +from airbyte_cdk.models import FailureType +from airbyte_cdk.utils.traced_exception import AirbyteTracedException + + +class DatetimeFormatMismatchError(AirbyteTracedException, ValueError): + """Raised when a datetime value matches none of the configured datetime formats. + + Inherits from `ValueError` so that callers which already handle datetime parsing + failures (multi-format fallback loops, cursor value extraction) keep working, and + from `AirbyteTracedException` so that the failure surfaces with a user-facing + message instead of a raw Python error. + + The user-facing `message` is deterministic: it names the stream and cursor field + when they are known, and never embeds the offending value or the format tokens. + Those live in `internal_message`. + """ + + def __init__( + self, + value: object, + formats: List[str], + cursor_field: Optional[str] = None, + stream_name: Optional[str] = None, + original_error: Optional[BaseException] = None, + ) -> None: + self.value = value + self.formats = formats + self.cursor_field = cursor_field + self.stream_name = stream_name + super().__init__( + message=self._build_message(cursor_field, stream_name), + internal_message=f"No format in {formats} matching {value}", + failure_type=FailureType.system_error, + exception=original_error, + ) + + @staticmethod + def _build_message(cursor_field: Optional[str], stream_name: Optional[str]) -> str: + if cursor_field and stream_name: + return ( + f'Cursor field "{cursor_field}" of stream "{stream_name}" matches none of the ' + "configured datetime formats." + ) + if cursor_field: + return f'Cursor field "{cursor_field}" matches none of the configured datetime formats.' + return "Datetime value matches none of the configured datetime formats." + + def with_context( + self, cursor_field: Optional[str] = None, stream_name: Optional[str] = None + ) -> "DatetimeFormatMismatchError": + """Return an equivalent error naming the stream and cursor field.""" + return DatetimeFormatMismatchError( + value=self.value, + formats=self.formats, + cursor_field=cursor_field or self.cursor_field, + stream_name=stream_name or self.stream_name, + original_error=self, + ) class DatetimeParser: @@ -35,7 +94,12 @@ def parse(self, date: Union[str, int], format: str) -> datetime.datetime: return self._UNIX_EPOCH + datetime.timedelta(milliseconds=int(date)) elif "%_ms" in format: format = format.replace("%_ms", "%f") - parsed_datetime = datetime.datetime.strptime(str(date), format) + try: + parsed_datetime = datetime.datetime.strptime(str(date), format) + except ValueError as error: + raise DatetimeFormatMismatchError( + value=date, formats=[format], original_error=error + ) from error if self._is_naive(parsed_datetime): return parsed_datetime.replace(tzinfo=datetime.timezone.utc) return parsed_datetime diff --git a/airbyte_cdk/sources/streams/concurrent/cursor.py b/airbyte_cdk/sources/streams/concurrent/cursor.py index 11eaad235c..ba6d177b9c 100644 --- a/airbyte_cdk/sources/streams/concurrent/cursor.py +++ b/airbyte_cdk/sources/streams/concurrent/cursor.py @@ -20,6 +20,7 @@ ) from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager +from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeFormatMismatchError from airbyte_cdk.sources.message import MessageRepository, NoopMessageRepository from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY from airbyte_cdk.sources.streams.concurrent.clamping import ClampingStrategy, NoClamping @@ -230,8 +231,9 @@ def __init__( ] = {} self._has_closed_at_least_one_slice = False self._cursor_granularity = cursor_granularity - # Flag to track if the logger has been triggered (per stream) + # Flags to track if the loggers have been triggered (per stream) self._should_be_synced_logger_triggered = False + self._unparsable_cursor_value_logger_triggered = False self._clamping_strategy = clamping_strategy self._is_ascending_order = True @@ -264,6 +266,16 @@ def _slice_boundary_fields_wrapper(self) -> Tuple[str, str]: def _get_concurrent_state( self, state: MutableMapping[str, Any] + ) -> Tuple[CursorValueType, MutableMapping[str, Any]]: + try: + return self._get_concurrent_state_without_context(state) + except DatetimeFormatMismatchError as error: + raise error.with_context( + cursor_field=self._cursor_field.cursor_field_key, stream_name=self._stream_name + ) from error + + def _get_concurrent_state_without_context( + self, state: MutableMapping[str, Any] ) -> Tuple[CursorValueType, MutableMapping[str, Any]]: if self._connector_state_converter.is_state_message_compatible(state): partitioned_state = self._connector_state_converter.deserialize(state) @@ -307,6 +319,8 @@ def observe(self, record: Record) -> None: self._most_recent_cursor_value_per_partition[record.associated_slice] = cursor_value elif most_recent_cursor_value > cursor_value: self._is_ascending_order = False + except DatetimeFormatMismatchError: + self._log_for_record_with_unparsable_cursor_value() except ValueError: self._log_for_record_without_cursor_value() @@ -567,6 +581,9 @@ def should_be_synced(self, record: Record) -> bool: """ try: record_cursor_value: CursorValueType = self._extract_cursor_value(record) + except DatetimeFormatMismatchError: + self._log_for_record_with_unparsable_cursor_value() + return True except ValueError: self._log_for_record_without_cursor_value() return True @@ -579,6 +596,13 @@ def _log_for_record_without_cursor_value(self) -> None: ) self._should_be_synced_logger_triggered = True + def _log_for_record_with_unparsable_cursor_value(self) -> None: + if not self._unparsable_cursor_value_logger_triggered: + LOGGER.warning( + f"Cursor field `{self.cursor_field.cursor_field_key}` in record for stream {self._stream_name} matches none of the configured datetime formats. The incremental sync will assume the record needs to be synced" + ) + self._unparsable_cursor_value_logger_triggered = True + def reduce_slice_range(self, stream_slice: StreamSlice) -> StreamSlice: # In theory, we might be more flexible here meaning that it doesn't need to be in ascending order but it just # needs to be ordered. For now though, we will only support ascending order. diff --git a/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py b/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py index fdb5d4d774..0fd618fe25 100644 --- a/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py +++ b/airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py @@ -8,7 +8,10 @@ # FIXME We would eventually like the Concurrent package do be agnostic of the declarative package. However, this is a breaking change and # the goal in the short term is only to fix the issue we are seeing for source-declarative-manifest. -from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeParser +from airbyte_cdk.sources.declarative.datetime.datetime_parser import ( + DatetimeFormatMismatchError, + DatetimeParser, +) from airbyte_cdk.sources.streams.concurrent.cursor import CursorField from airbyte_cdk.sources.streams.concurrent.state_converters.abstract_stream_state_converter import ( AbstractStreamStateConverter, @@ -215,9 +218,14 @@ def output_format(self, timestamp: datetime) -> str: return self._parser.format(timestamp, self._datetime_format) def parse_timestamp(self, timestamp: str) -> datetime: + last_error: Optional[ValueError] = None for datetime_format in self._input_datetime_formats: try: return self._parser.parse(timestamp, datetime_format) - except ValueError: - pass - raise ValueError(f"No format in {self._input_datetime_formats} matching {timestamp}") + except ValueError as error: + last_error = error + raise DatetimeFormatMismatchError( + value=timestamp, + formats=self._input_datetime_formats, + original_error=last_error, + ) diff --git a/unit_tests/legacy/sources/declarative/incremental/test_datetime_based_cursor.py b/unit_tests/legacy/sources/declarative/incremental/test_datetime_based_cursor.py index 800f103771..f0e4a4439f 100644 --- a/unit_tests/legacy/sources/declarative/incremental/test_datetime_based_cursor.py +++ b/unit_tests/legacy/sources/declarative/incremental/test_datetime_based_cursor.py @@ -8,6 +8,10 @@ import pytest from airbyte_cdk.legacy.sources.declarative.incremental import DatetimeBasedCursor +from airbyte_cdk.models import FailureType +from airbyte_cdk.sources.declarative.datetime.datetime_parser import ( + DatetimeFormatMismatchError, +) from airbyte_cdk.sources.declarative.datetime.min_max_datetime import MinMaxDatetime from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString from airbyte_cdk.sources.declarative.requesters.request_option import ( @@ -1020,9 +1024,17 @@ def test_given_unknown_format_when_parse_date_then_raise_error(): config=config, parameters={}, ) - with pytest.raises(ValueError): + with pytest.raises(DatetimeFormatMismatchError) as exc_info: slicer.parse_date("2021-01-01T00:00:00.000000+0000") + assert isinstance(exc_info.value, ValueError) + assert ( + exc_info.value.message + == f'Cursor field "{cursor_field}" matches none of the configured datetime formats.' + ) + assert exc_info.value.failure_type == FailureType.system_error + assert "2021-01-01T00:00:00.000000+0000" in (exc_info.value.internal_message or "") + @pytest.mark.parametrize( "test_name, input_dt, datetimeformat, datetimeformat_granularity, expected_output", diff --git a/unit_tests/sources/declarative/datetime/test_datetime_parser.py b/unit_tests/sources/declarative/datetime/test_datetime_parser.py index b9da9852f9..fce74b0e8e 100644 --- a/unit_tests/sources/declarative/datetime/test_datetime_parser.py +++ b/unit_tests/sources/declarative/datetime/test_datetime_parser.py @@ -6,7 +6,11 @@ import pytest -from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeParser +from airbyte_cdk.models import FailureType +from airbyte_cdk.sources.declarative.datetime.datetime_parser import ( + DatetimeFormatMismatchError, + DatetimeParser, +) @pytest.mark.parametrize( @@ -125,3 +129,33 @@ def test_format_datetime(input_dt: datetime.datetime, datetimeformat: str, expec parser = DatetimeParser() output_date = parser.format(input_dt, datetimeformat) assert output_date == expected_output + + +def test_given_value_not_matching_format_when_parse_then_raise_traced_error(): + with pytest.raises(DatetimeFormatMismatchError) as exc_info: + DatetimeParser().parse("2025-06-28T18:35:16Z", "%Y-%m-%dT%H:%M:%S.%fZ") + + assert isinstance(exc_info.value, ValueError) + assert ( + exc_info.value.message == "Datetime value matches none of the configured datetime formats." + ) + assert exc_info.value.failure_type == FailureType.system_error + assert ( + exc_info.value.internal_message + == "No format in ['%Y-%m-%dT%H:%M:%S.%fZ'] matching 2025-06-28T18:35:16Z" + ) + + +def test_given_context_when_with_context_then_message_names_stream_and_cursor_field(): + error = DatetimeFormatMismatchError( + value="2025-06-28T18:35:16Z", formats=["%Y-%m-%dT%H:%M:%S.%fZ"] + ).with_context(cursor_field="updated_at", stream_name="items") + + assert error.message == ( + 'Cursor field "updated_at" of stream "items" matches none of the configured datetime ' + "formats." + ) + assert ( + error.internal_message + == "No format in ['%Y-%m-%dT%H:%M:%S.%fZ'] matching 2025-06-28T18:35:16Z" + ) diff --git a/unit_tests/sources/declarative/test_state_delegating_stream.py b/unit_tests/sources/declarative/test_state_delegating_stream.py index 90bd0d76cb..8a97bb8ba6 100644 --- a/unit_tests/sources/declarative/test_state_delegating_stream.py +++ b/unit_tests/sources/declarative/test_state_delegating_stream.py @@ -24,6 +24,9 @@ from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( ConcurrentDeclarativeSource, ) +from airbyte_cdk.sources.declarative.datetime.datetime_parser import ( + DatetimeFormatMismatchError, +) from airbyte_cdk.test.mock_http import HttpMocker, HttpRequest, HttpResponse _CONFIG = {"start_date": "2024-07-01T00:00:00.000Z"} @@ -608,9 +611,16 @@ def test_cursor_age_validation_raises_error_for_unparseable_cursor(): source_config=manifest, config=_CONFIG, catalog=None, state=state ) - with pytest.raises(ValueError, match="not-a-date"): + with pytest.raises(DatetimeFormatMismatchError) as exc_info: source.discover(logger=MagicMock(), config=_CONFIG) + assert isinstance(exc_info.value, ValueError) + assert exc_info.value.message == ( + 'Cursor field "updated_at" of stream "TestStream" matches none of the configured ' + "datetime formats." + ) + assert "not-a-date" in (exc_info.value.internal_message or "") + @freezegun.freeze_time("2024-07-15") def test_final_state_cursor_falls_back_to_full_refresh_when_state_unparseable(): diff --git a/unit_tests/sources/streams/concurrent/test_cursor.py b/unit_tests/sources/streams/concurrent/test_cursor.py index 13fe1df875..70d871d140 100644 --- a/unit_tests/sources/streams/concurrent/test_cursor.py +++ b/unit_tests/sources/streams/concurrent/test_cursor.py @@ -2,6 +2,7 @@ # Copyright (c) 2023 Airbyte, Inc., all rights reserved. # +import logging from copy import deepcopy from datetime import datetime, timedelta, timezone from typing import Any, Mapping, Optional @@ -11,7 +12,11 @@ import freezegun import pytest +from airbyte_cdk.models import FailureType from airbyte_cdk.sources.connector_state_manager import ConnectorStateManager +from airbyte_cdk.sources.declarative.datetime.datetime_parser import ( + DatetimeFormatMismatchError, +) from airbyte_cdk.sources.message import MessageRepository from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY from airbyte_cdk.sources.streams.concurrent.clamping import ( @@ -1411,3 +1416,53 @@ def test_final_state_cursor_get_cursor_datetime_from_state_returns_now_for_no_cu result_with_empty_state = cursor.get_cursor_datetime_from_state({}) assert result_with_empty_state is None + + +_MILLIS_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ" +_WHOLE_SECOND_CURSOR_VALUE = "2025-06-28T18:35:16Z" + + +def _cursor_with_custom_format(stream_state: Mapping[str, Any]) -> ConcurrentCursor: + return ConcurrentCursor( + _A_STREAM_NAME, + _A_STREAM_NAMESPACE, + deepcopy(stream_state), + Mock(spec=MessageRepository), + Mock(spec=ConnectorStateManager), + CustomFormatConcurrentStreamStateConverter(_MILLIS_FORMAT), + CursorField(_A_CURSOR_FIELD_KEY), + _SLICE_BOUNDARY_FIELDS, + None, + CustomFormatConcurrentStreamStateConverter.get_end_provider(), + _NO_LOOKBACK_WINDOW, + ) + + +def test_given_state_value_not_matching_format_when_create_cursor_then_raise_traced_error() -> None: + with pytest.raises(DatetimeFormatMismatchError) as exc_info: + _cursor_with_custom_format({_A_CURSOR_FIELD_KEY: _WHOLE_SECOND_CURSOR_VALUE}) + + assert isinstance(exc_info.value, ValueError) + assert exc_info.value.message == ( + f'Cursor field "{_A_CURSOR_FIELD_KEY}" of stream "{_A_STREAM_NAME}" matches none of the ' + "configured datetime formats." + ) + assert exc_info.value.failure_type == FailureType.system_error + assert _WHOLE_SECOND_CURSOR_VALUE in (exc_info.value.internal_message or "") + assert _MILLIS_FORMAT in (exc_info.value.internal_message or "") + + +def test_given_record_cursor_value_not_matching_format_when_should_be_synced_then_log_and_sync( + caplog, +) -> None: + cursor = _cursor_with_custom_format(_NO_STATE) + record = Record( + data={_A_CURSOR_FIELD_KEY: _WHOLE_SECOND_CURSOR_VALUE}, + associated_slice=None, + stream_name=_A_STREAM_NAME, + ) + + with caplog.at_level(logging.WARNING, logger="airbyte"): + assert cursor.should_be_synced(record) is True + + assert "matches none of the configured datetime formats" in caplog.text diff --git a/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py b/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py index ebcb3c20c4..51667f9073 100644 --- a/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py +++ b/unit_tests/sources/streams/concurrent/test_datetime_state_converter.py @@ -6,6 +6,10 @@ import pytest +from airbyte_cdk.models import FailureType +from airbyte_cdk.sources.declarative.datetime.datetime_parser import ( + DatetimeFormatMismatchError, +) from airbyte_cdk.sources.streams.concurrent.cursor import CursorField from airbyte_cdk.sources.streams.concurrent.state_converters.abstract_stream_state_converter import ( ConcurrencyCompatibleStateType, @@ -406,3 +410,19 @@ def test_given_when_parse_timestamp_then_eventually_fallback_on_output_format(): parsed_datetime = converter.parse_timestamp("2024-01-01T02:00:00") assert parsed_datetime == datetime(2024, 1, 1, 2, 0, 0, tzinfo=timezone.utc) + + +def test_given_no_matching_input_format_when_parse_timestamp_then_raise_traced_error(): + converter = CustomFormatConcurrentStreamStateConverter( + "%Y-%m-%dT%H:%M:%S.%fZ", ["%Y-%m-%dT%H:%M:%S.%fZ"] + ) + + with pytest.raises(DatetimeFormatMismatchError) as exc_info: + converter.parse_timestamp("2025-06-28T18:35:16Z") + + assert isinstance(exc_info.value, ValueError) + assert ( + exc_info.value.message == "Datetime value matches none of the configured datetime formats." + ) + assert exc_info.value.failure_type == FailureType.system_error + assert "2025-06-28T18:35:16Z" in (exc_info.value.internal_message or "")