fix(cdk): replace direct datetime parsing/formatting with datetime_helpers

Co-Authored-By: Aaron <AJ> Steers <[email protected]>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/datetime/datetime_parser.py

@@ -5,6 +5,8 @@
 import datetime
 from typing import Union
 
+from airbyte_cdk.utils.datetime_helpers import ab_datetime_format, ab_datetime_parse
+
 
 class DatetimeParser:
     """
@@ -19,47 +21,82 @@ class DatetimeParser:
     _UNIX_EPOCH = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
 
     def parse(self, date: Union[str, int], format: str) -> datetime.datetime:
-        # "%s" is a valid (but unreliable) directive for formatting, but not for parsing
-        # It is defined as
-        # The number of seconds since the Epoch, 1970-01-01 00:00:00+0000 (UTC). https://man7.org/linux/man-pages/man3/strptime.3.html
-        #
-        # The recommended way to parse a date from its timestamp representation is to use datetime.fromtimestamp
-        # See https://stackoverflow.com/a/4974930
+        """
+        Parse a datetime string according to a specified format.
+
+        Args:
+            date: String to parse
+            format: Format string as described in https://docs.python.org/3/library/datetime.html#strftime-and-strptime-behavior
+                with the following extensions:
+                - %s: Unix timestamp
+                - %s_as_float: Unix timestamp as float
+                - %epoch_microseconds: Microseconds since epoch
+                - %ms: Milliseconds since epoch
+                - %_ms: Custom millisecond format
+
+        Returns:
+            The parsed datetime
+        """
+        # Special format handling
         if format == "%s":
             return datetime.datetime.fromtimestamp(int(date), tz=datetime.timezone.utc)
         elif format == "%s_as_float":
             return datetime.datetime.fromtimestamp(float(date), tz=datetime.timezone.utc)
         elif format == "%epoch_microseconds":
-            return self._UNIX_EPOCH + datetime.timedelta(microseconds=int(date))
+            epoch = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
+            return epoch + datetime.timedelta(microseconds=int(date))
         elif format == "%ms":
-            return self._UNIX_EPOCH + datetime.timedelta(milliseconds=int(date))
+            epoch = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
+            return epoch + datetime.timedelta(milliseconds=int(date))
         elif "%_ms" in format:
+            # Convert custom millisecond format to standard format
             format = format.replace("%_ms", "%f")
-        parsed_datetime = datetime.datetime.strptime(str(date), format)
-        if self._is_naive(parsed_datetime):
+            
+        # For standard formats, use ab_datetime_parse with the specific format
+        try:
+            result = ab_datetime_parse(date, formats=[format], disallow_other_formats=True)
+            return result.to_datetime()  # Convert AirbyteDateTime to standard datetime
+        except ValueError:
+            # Fallback to original implementation for backward compatibility
+            parsed_datetime = datetime.datetime.strptime(str(date), format)
+            if self._is_naive(parsed_datetime):
+                return parsed_datetime.replace(tzinfo=datetime.timezone.utc)
             return parsed_datetime.replace(tzinfo=datetime.timezone.utc)
-        return parsed_datetime
+        return parsed_datetime.replace(tzinfo=datetime.timezone.utc)
 
     def format(self, dt: datetime.datetime, format: str) -> str:
-        # strftime("%s") is unreliable because it ignores the time zone information and assumes the time zone of the system it's running on
-        # It's safer to use the timestamp() method than the %s directive
-        # See https://stackoverflow.com/a/4974930
+        """
+        Format a datetime object according to a specified format.
+
+        Args:
+            dt: The datetime object to format
+            format: Format string as described in https://docs.python.org/3/library/datetime.html#strftime-and-strptime-behavior
+                with the following extensions:
+                - %s: Unix timestamp
+                - %s_as_float: Unix timestamp as float
+                - %epoch_microseconds: Microseconds since epoch
+                - %ms: Milliseconds since epoch
+                - %_ms: Custom millisecond format
+
+        Returns:
+            The formatted string
+        """
+        # Handle special formats
         if format == "%s":
             return str(int(dt.timestamp()))
-        if format == "%s_as_float":
+        elif format == "%s_as_float":
             return str(float(dt.timestamp()))
-        if format == "%epoch_microseconds":
+        elif format == "%epoch_microseconds":
             return str(int(dt.timestamp() * 1_000_000))
-        if format == "%ms":
-            # timstamp() returns a float representing the number of seconds since the unix epoch
+        elif format == "%ms":
             return str(int(dt.timestamp() * 1000))
-        if "%_ms" in format:
+        elif "%_ms" in format:
             _format = format.replace("%_ms", "%f")
             milliseconds = int(dt.microsecond / 1000)
-            formatted_dt = dt.strftime(_format).replace(dt.strftime("%f"), "%03d" % milliseconds)
-            return formatted_dt
-        else:
-            return dt.strftime(format)
+            return dt.strftime(_format).replace(dt.strftime("%f"), "%03d" % milliseconds)
+
+        # For standard formats, use ab_datetime_format
+        return ab_datetime_format(dt, format)
 
     def _is_naive(self, dt: datetime.datetime) -> bool:
         return dt.tzinfo is None or dt.tzinfo.utcoffset(dt) is None

airbyte_cdk/sources/declarative/datetime/min_max_datetime.py

@@ -8,6 +8,7 @@
 
 from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeParser
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
+from airbyte_cdk.utils.datetime_helpers import ab_datetime_format, ab_datetime_parse
 
 
 @dataclass
@@ -39,7 +40,6 @@ class MinMaxDatetime:
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self.datetime = InterpolatedString.create(self.datetime, parameters=parameters or {})
-        self._parser = DatetimeParser()
         self.min_datetime = (
             InterpolatedString.create(self.min_datetime, parameters=parameters)  # type: ignore [assignment]  #  expression has type "InterpolatedString | None", variable has type "InterpolatedString | str"
             if self.min_datetime
@@ -65,25 +65,25 @@ def get_datetime(
         if not datetime_format:
             datetime_format = "%Y-%m-%dT%H:%M:%S.%f%z"
 
-        time = self._parser.parse(
+        time = ab_datetime_parse(
             str(
                 self.datetime.eval(  # type: ignore[union-attr] # str has no attribute "eval"
                     config,
                     **additional_parameters,
                 )
             ),
-            datetime_format,
-        )  # type: ignore # datetime is always cast to an interpolated string
+            formats=[datetime_format] if datetime_format else None
+        ).to_datetime()  # type: ignore # datetime is always cast to an interpolated string
 
         if self.min_datetime:
             min_time = str(self.min_datetime.eval(config, **additional_parameters))  # type: ignore # min_datetime is always cast to an interpolated string
             if min_time:
-                min_datetime = self._parser.parse(min_time, datetime_format)  # type: ignore # min_datetime is always cast to an interpolated string
+                min_datetime = ab_datetime_parse(min_time, formats=[datetime_format] if datetime_format else None).to_datetime()  # type: ignore # min_datetime is always cast to an interpolated string
                 time = max(time, min_datetime)
         if self.max_datetime:
             max_time = str(self.max_datetime.eval(config, **additional_parameters))  # type: ignore # max_datetime is always cast to an interpolated string
             if max_time:
-                max_datetime = self._parser.parse(max_time, datetime_format)
+                max_datetime = ab_datetime_parse(max_time, formats=[datetime_format] if datetime_format else None).to_datetime()
                 time = min(time, max_datetime)
         return time
 

airbyte_cdk/sources/declarative/incremental/datetime_based_cursor.py

@@ -21,6 +21,7 @@
 )
 from airbyte_cdk.sources.message import MessageRepository
 from airbyte_cdk.sources.types import Config, Record, StreamSlice, StreamState
+from airbyte_cdk.utils.datetime_helpers import ab_datetime_format, ab_datetime_parse
 from airbyte_cdk.utils.mapping_helpers import _validate_component_request_option_paths
 
 
@@ -112,7 +113,6 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._partition_field_end = InterpolatedString.create(
             self.partition_field_end or "end_time", parameters=parameters
         )
-        self._parser = DatetimeParser()
 
         # If datetime format is not specified then start/end datetime should inherit it from the stream slicer
         if not self._start_datetime.datetime_format:
@@ -253,7 +253,7 @@ def _calculate_cursor_datetime_from_state(
         return datetime.datetime.min.replace(tzinfo=datetime.timezone.utc)
 
     def _format_datetime(self, dt: datetime.datetime) -> str:
-        return self._parser.format(dt, self.datetime_format)
+        return ab_datetime_format(dt, self.datetime_format)
 
     def _partition_daterange(
         self,
@@ -308,12 +308,11 @@ 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]:
-            try:
-                return self._parser.parse(date, datetime_format)
-            except ValueError:
-                pass
-        raise ValueError(f"No format in {self.cursor_datetime_formats} matching {date}")
+        try:
+            # Try each format in the list, falling back to the default format
+            return ab_datetime_parse(date, formats=self.cursor_datetime_formats + [self.datetime_format]).to_datetime()
+        except ValueError:
+            raise ValueError(f"No format in {self.cursor_datetime_formats + [self.datetime_format]} matching {date}")
 
     @classmethod
     def _parse_timedelta(cls, time_str: Optional[str]) -> Union[datetime.timedelta, Duration]:

airbyte_cdk/sources/declarative/interpolation/macros.py

@@ -13,6 +13,11 @@
 from isodate import parse_duration
 
 from airbyte_cdk.sources.declarative.datetime.datetime_parser import DatetimeParser
+from airbyte_cdk.utils.datetime_helpers import (
+    ab_datetime_format,
+    ab_datetime_now,
+    ab_datetime_parse,
+)
 
 """
 This file contains macros that can be evaluated by a `JinjaInterpolation` object
@@ -26,7 +31,7 @@ def now_utc() -> datetime.datetime:
     Usage:
     `"{{ now_utc() }}"`
     """
-    return datetime.datetime.now(datetime.timezone.utc)
+    return ab_datetime_now().to_datetime()
 
 
 def today_utc() -> datetime.date:
@@ -82,12 +87,10 @@ def str_to_datetime(s: str) -> datetime.datetime:
     :param s: string to parse as datetime
     :return: datetime object in UTC timezone
     """
-
-    parsed_date = parser.isoparse(s)
-    if not parsed_date.tzinfo:
-        # Assume UTC if the input does not contain a timezone
-        parsed_date = parsed_date.replace(tzinfo=pytz.utc)
-    return parsed_date.astimezone(pytz.utc)
+    dt = ab_datetime_parse(s).to_datetime()
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=pytz.utc)
+    return dt.astimezone(pytz.utc)
 
 
 def max(*args: typing.Any) -> typing.Any:
@@ -140,9 +143,10 @@ def day_delta(num_days: int, format: str = "%Y-%m-%dT%H:%M:%S.%f%z") -> str:
     :param num_days: number of days to add to current date time
     :return: datetime formatted as RFC3339
     """
-    return (
-        datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(days=num_days)
-    ).strftime(format)
+    return ab_datetime_format(
+        ab_datetime_now().to_datetime() + datetime.timedelta(days=num_days),
+        format
+    )
 
 
 def duration(datestring: str) -> Union[datetime.timedelta, isodate.Duration]:
@@ -169,17 +173,17 @@ def format_datetime(
     https://github.com/python/cpython/issues/56959
     """
     if isinstance(dt, datetime.datetime):
-        return dt.strftime(format)
+        return ab_datetime_format(dt, format)
 
     if isinstance(dt, int):
-        dt_datetime = DatetimeParser().parse(dt, input_format if input_format else "%s")
+        dt_datetime = ab_datetime_parse(dt, formats=[input_format] if input_format else None).to_datetime()
     else:
         dt_datetime = (
-            datetime.datetime.strptime(dt, input_format) if input_format else str_to_datetime(dt)
+            ab_datetime_parse(dt, formats=[input_format] if input_format else None).to_datetime() if input_format else str_to_datetime(dt)
         )
     if dt_datetime.tzinfo is None:
         dt_datetime = dt_datetime.replace(tzinfo=pytz.utc)
-    return DatetimeParser().format(dt=dt_datetime, format=format)
+    return ab_datetime_format(dt_datetime, format)
 
 
 _macros_list = [

airbyte_cdk/sources/file_based/file_based_stream_reader.py

@@ -19,6 +19,7 @@
     use_file_transfer,
 )
 from airbyte_cdk.sources.file_based.remote_file import RemoteFile
+from airbyte_cdk.utils.datetime_helpers import ab_datetime_parse
 
 
 class FileReadMode(Enum):
@@ -95,7 +96,7 @@ def filter_files_by_globs_and_start_date(
         Utility method for filtering files based on globs.
         """
         start_date = (
-            datetime.strptime(self.config.start_date, self.DATE_TIME_FORMAT)
+            ab_datetime_parse(self.config.start_date, formats=[self.DATE_TIME_FORMAT]).to_datetime()
             if self.config and self.config.start_date
             else None
         )

airbyte_cdk/sources/file_based/stream/concurrent/cursor/file_based_concurrent_cursor.py

@@ -20,6 +20,7 @@
 from airbyte_cdk.sources.streams.concurrent.cursor import CursorField
 from airbyte_cdk.sources.streams.concurrent.partitions.partition import Partition
 from airbyte_cdk.sources.types import Record
+from airbyte_cdk.utils.datetime_helpers import ab_datetime_format, ab_datetime_parse
 
 if TYPE_CHECKING:
     from airbyte_cdk.sources.file_based.stream.concurrent.adapters import FileBasedStreamPartition
@@ -110,11 +111,11 @@ def _compute_prev_sync_cursor(self, value: Optional[StreamState]) -> Tuple[datet
         )
         cursor_str = min(prev_cursor_str, earliest_file_cursor_value)
         cursor_dt, cursor_uri = cursor_str.split("_", 1)
-        return datetime.strptime(cursor_dt, self.DATE_TIME_FORMAT), cursor_uri
+        return ab_datetime_parse(cursor_dt, formats=[self.DATE_TIME_FORMAT]).to_datetime(), cursor_uri
 
     def _get_cursor_key_from_file(self, file: Optional[RemoteFile]) -> str:
         if file:
-            return f"{datetime.strftime(file.last_modified, self.DATE_TIME_FORMAT)}_{file.uri}"
+            return f"{ab_datetime_format(file.last_modified, self.DATE_TIME_FORMAT)}_{file.uri}"
         return self.zero_cursor_value
 
     def _compute_earliest_file_in_history(self) -> Optional[RemoteFile]:
@@ -125,7 +126,7 @@ def _compute_earliest_file_in_history(self) -> Optional[RemoteFile]:
                 )
                 return RemoteFile(
                     uri=filename,
-                    last_modified=datetime.strptime(last_modified, self.DATE_TIME_FORMAT),
+                    last_modified=ab_datetime_parse(last_modified, formats=[self.DATE_TIME_FORMAT]).to_datetime(),
                 )
             else:
                 return None
@@ -153,8 +154,8 @@ def add_file(self, file: RemoteFile) -> None:
                     )
                 else:
                     self._pending_files.pop(file.uri)
-                self._file_to_datetime_history[file.uri] = file.last_modified.strftime(
-                    self.DATE_TIME_FORMAT
+                self._file_to_datetime_history[file.uri] = ab_datetime_format(
+                    file.last_modified, self.DATE_TIME_FORMAT
                 )
                 if len(self._file_to_datetime_history) > self.DEFAULT_MAX_HISTORY_SIZE:
                     # Get the earliest file based on its last modified date and its uri
@@ -210,7 +211,7 @@ def _compute_latest_file_in_history(self) -> Optional[RemoteFile]:
                 )
                 return RemoteFile(
                     uri=filename,
-                    last_modified=datetime.strptime(last_modified, self.DATE_TIME_FORMAT),
+                    last_modified=ab_datetime_parse(last_modified, formats=[self.DATE_TIME_FORMAT]).to_datetime(),
                 )
             else:
                 return None
@@ -239,9 +240,9 @@ def _should_sync_file(self, file: RemoteFile, logger: logging.Logger) -> bool:
         with self._state_lock:
             if file.uri in self._file_to_datetime_history:
                 # If the file's uri is in the history, we should sync the file if it has been modified since it was synced
-                updated_at_from_history = datetime.strptime(
-                    self._file_to_datetime_history[file.uri], self.DATE_TIME_FORMAT
-                )
+                updated_at_from_history = ab_datetime_parse(
+                    self._file_to_datetime_history[file.uri], formats=[self.DATE_TIME_FORMAT]
+                ).to_datetime()
                 if file.last_modified < updated_at_from_history:
                     self._message_repository.emit_message(
                         AirbyteMessage(
@@ -287,7 +288,7 @@ def _compute_start_time(self) -> datetime:
             return datetime.min
         else:
             earliest = min(self._file_to_datetime_history.values())
-            earliest_dt = datetime.strptime(earliest, self.DATE_TIME_FORMAT)
+            earliest_dt = ab_datetime_parse(earliest, formats=[self.DATE_TIME_FORMAT]).to_datetime()
             if self._is_history_full():
                 time_window = datetime.now() - self._time_window_if_history_is_full
                 earliest_dt = min(earliest_dt, time_window)