manual updates (aj)

aaronsteers · aaronsteers · commit af7d32cd56ac · 2025-04-08T23:01:26.000-07:00
diff --git a/airbyte_cdk/sources/declarative/datetime/datetime_parser.py b/airbyte_cdk/sources/declarative/datetime/datetime_parser.py
@@ -5,8 +5,6 @@
 import datetime
 from typing import Union
 
-from airbyte_cdk.utils.datetime_helpers import ab_datetime_format, ab_datetime_parse
-
 
 class DatetimeParser:
     """
@@ -21,82 +19,47 @@ class DatetimeParser:
     _UNIX_EPOCH = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
 
     def parse(self, date: Union[str, int], format: str) -> datetime.datetime:
-        """
-        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
+        # "%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
         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":
-            epoch = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
-            return epoch + datetime.timedelta(microseconds=int(date))
+            return self._UNIX_EPOCH + datetime.timedelta(microseconds=int(date))
         elif format == "%ms":
-            epoch = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc)
-            return epoch + datetime.timedelta(milliseconds=int(date))
+            return self._UNIX_EPOCH + datetime.timedelta(milliseconds=int(date))
         elif "%_ms" in format:
-            # Convert custom millisecond format to standard format
             format = format.replace("%_ms", "%f")
-
-        # 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)
+        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
 
     def format(self, dt: datetime.datetime, format: str) -> str:
-        """
-        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
+        # 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
         if format == "%s":
             return str(int(dt.timestamp()))
-        elif format == "%s_as_float":
+        if format == "%s_as_float":
             return str(float(dt.timestamp()))
-        elif format == "%epoch_microseconds":
+        if format == "%epoch_microseconds":
             return str(int(dt.timestamp() * 1_000_000))
-        elif format == "%ms":
+        if format == "%ms":
+            # timstamp() returns a float representing the number of seconds since the unix epoch
             return str(int(dt.timestamp() * 1000))
-        elif "%_ms" in format:
+        if "%_ms" in format:
             _format = format.replace("%_ms", "%f")
             milliseconds = int(dt.microsecond / 1000)
-            return dt.strftime(_format).replace(dt.strftime("%f"), "%03d" % milliseconds)
-
-        # For standard formats, use ab_datetime_format
-        return ab_datetime_format(dt, format)
+            formatted_dt = dt.strftime(_format).replace(dt.strftime("%f"), "%03d" % milliseconds)
+            return formatted_dt
+        else:
+            return dt.strftime(format)
 
     def _is_naive(self, dt: datetime.datetime) -> bool:
         return dt.tzinfo is None or dt.tzinfo.utcoffset(dt) is None
diff --git a/airbyte_cdk/sources/declarative/incremental/datetime_based_cursor.py b/airbyte_cdk/sources/declarative/incremental/datetime_based_cursor.py
@@ -253,6 +253,11 @@ def _calculate_cursor_datetime_from_state(
         return datetime.datetime.min.replace(tzinfo=datetime.timezone.utc)
 
     def _format_datetime(self, dt: datetime.datetime) -> str:
+        """Format the datetime according to the configured format.
+
+        # TODO: Standardize cursor serialization with ISO 8601 format and deprecate custom formats
+        #       in STATE messages.
+        """
         return ab_datetime_format(dt, self.datetime_format)
 
     def _partition_daterange(
@@ -267,11 +272,7 @@ def _partition_daterange(
 
         while self._is_within_date_range(start, end):
             next_start = self._evaluate_next_start_date_safely(start, step)
-            if hasattr(next_start, "to_datetime"):
-                next_start_dt = next_start.to_datetime()
-            else:
-                next_start_dt = next_start
-            end_date = self._get_date(next_start_dt - self._cursor_granularity, end, min)
+            end_date = self._get_date(next_start - self._cursor_granularity, end, min)
             dates.append(
                 StreamSlice(
                     partition={},
@@ -298,8 +299,6 @@ def _evaluate_next_start_date_safely(
         would have broken anyway.
         """
         try:
-            if hasattr(start, "to_datetime"):
-                start = start.to_datetime()
             return start + step
         except OverflowError:
             return datetime.datetime.max.replace(tzinfo=datetime.timezone.utc)
@@ -314,15 +313,17 @@ def _get_date(
         return comparator(cursor_date, default_date)
 
     def parse_date(self, date: str) -> datetime.datetime:
+        formats = list(set(self.cursor_datetime_formats + [self.datetime_format]))
         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}"
+                date,
+                formats=formats,
+                disallow_other_formats=False,  # TODO: Consider permissive parsing.
             )
+        except ValueError as ex:
+            raise ValueError(
+                f"No format in {formats} matching '{date}'"
+            ) from ex
 
     @classmethod
     def _parse_timedelta(cls, time_str: Optional[str]) -> Union[datetime.timedelta, Duration]:
diff --git a/airbyte_cdk/sources/declarative/interpolation/macros.py b/airbyte_cdk/sources/declarative/interpolation/macros.py
@@ -31,7 +31,7 @@ def now_utc() -> datetime.datetime:
     Usage:
     `"{{ now_utc() }}"`
     """
-    return ab_datetime_now().to_datetime()
+    return ab_datetime_now()
 
 
 def today_utc() -> datetime.date:
@@ -67,10 +67,7 @@ def timestamp(dt: Union[float, str]) -> Union[int, float]:
     :param dt: datetime to convert to timestamp
     :return: unix timestamp
     """
-    if isinstance(dt, (int, float)):
-        return int(dt)
-    else:
-        return str_to_datetime(dt).astimezone(pytz.utc).timestamp()
+    return ab_datetime_parse(dt).timestamp()
 
 
 def str_to_datetime(s: str) -> datetime.datetime:
@@ -87,10 +84,7 @@ def str_to_datetime(s: str) -> datetime.datetime:
     :param s: string to parse as datetime
     :return: datetime object in UTC timezone
     """
-    dt = ab_datetime_parse(s).to_datetime()
-    if dt.tzinfo is None:
-        dt = dt.replace(tzinfo=pytz.utc)
-    return dt.astimezone(pytz.utc)
+    return ab_datetime_parse(s)
 
 
 def max(*args: typing.Any) -> typing.Any:
@@ -144,7 +138,8 @@ def day_delta(num_days: int, format: str = "%Y-%m-%dT%H:%M:%S.%f%z") -> str:
     :return: datetime formatted as RFC3339
     """
     return ab_datetime_format(
-        ab_datetime_now().to_datetime() + datetime.timedelta(days=num_days), format
+        ab_datetime_now() + datetime.timedelta(days=num_days),
+        format=format,
     )
 
 
@@ -175,18 +170,12 @@ def format_datetime(
         return ab_datetime_format(dt, format)
 
     if isinstance(dt, int):
-        dt_datetime = ab_datetime_parse(
-            dt, formats=[input_format] if input_format else None
-        ).to_datetime()
+        dt_datetime = ab_datetime_parse(dt, formats=[input_format])
     else:
-        dt_datetime = (
-            ab_datetime_parse(dt, formats=[input_format] if input_format else None).to_datetime()
-            if input_format
-            else str_to_datetime(dt)
-        )
+        dt_datetime = ab_datetime_parse(dt, formats=[input_format])
     if dt_datetime.tzinfo is None:
         dt_datetime = dt_datetime.replace(tzinfo=pytz.utc)
-    return ab_datetime_format(dt_datetime, format)
+    return ab_datetime_format(dt_datetime, format=format)
 
 
 _macros_list = [
diff --git a/airbyte_cdk/sources/file_based/file_based_stream_reader.py b/airbyte_cdk/sources/file_based/file_based_stream_reader.py
@@ -96,7 +96,11 @@ def filter_files_by_globs_and_start_date(
         Utility method for filtering files based on globs.
         """
         start_date = (
-            ab_datetime_parse(self.config.start_date, formats=[self.DATE_TIME_FORMAT]).to_datetime()
+            ab_datetime_parse(
+                self.config.start_date,
+                formats=[self.DATE_TIME_FORMAT],
+                disallow_other_formats=False,
+            )
             if self.config and self.config.start_date
             else None
         )
diff --git a/airbyte_cdk/sources/file_based/stream/concurrent/cursor/file_based_concurrent_cursor.py b/airbyte_cdk/sources/file_based/stream/concurrent/cursor/file_based_concurrent_cursor.py
@@ -112,8 +112,10 @@ 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 ab_datetime_parse(
-            cursor_dt, formats=[self.DATE_TIME_FORMAT]
-        ).to_datetime(), cursor_uri
+            cursor_dt,
+            formats=[self.DATE_TIME_FORMAT],
+            disallow_other_formats=False,
+        ), cursor_uri
 
     def _get_cursor_key_from_file(self, file: Optional[RemoteFile]) -> str:
         if file:
@@ -129,8 +131,10 @@ def _compute_earliest_file_in_history(self) -> Optional[RemoteFile]:
                 return RemoteFile(
                     uri=filename,
                     last_modified=ab_datetime_parse(
-                        last_modified, formats=[self.DATE_TIME_FORMAT]
-                    ).to_datetime(),
+                        last_modified,
+                        formats=[self.DATE_TIME_FORMAT],
+                        disallow_other_formats=False,
+                    ),
                 )
             else:
                 return None
@@ -159,7 +163,8 @@ def add_file(self, file: RemoteFile) -> None:
                 else:
                     self._pending_files.pop(file.uri)
                 self._file_to_datetime_history[file.uri] = ab_datetime_format(
-                    file.last_modified, self.DATE_TIME_FORMAT
+                    file.last_modified,
+                    format=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
@@ -212,12 +217,14 @@ def _compute_latest_file_in_history(self) -> Optional[RemoteFile]:
             if self._file_to_datetime_history:
                 filename, last_modified = max(
                     self._file_to_datetime_history.items(), key=lambda f: (f[1], f[0])
-                )
+                )``
                 return RemoteFile(
                     uri=filename,
                     last_modified=ab_datetime_parse(
-                        last_modified, formats=[self.DATE_TIME_FORMAT]
-                    ).to_datetime(),
+                        last_modified,
+                        formats=[self.DATE_TIME_FORMAT],
+                        disallow_other_formats=False,
+                    ),
                 )
             else:
                 return None
@@ -247,8 +254,10 @@ def _should_sync_file(self, file: RemoteFile, logger: logging.Logger) -> bool:
             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 = ab_datetime_parse(
-                    self._file_to_datetime_history[file.uri], formats=[self.DATE_TIME_FORMAT]
-                ).to_datetime()
+                    self._file_to_datetime_history[file.uri],
+                    formats=[self.DATE_TIME_FORMAT],
+                    disallow_other_formats=False,
+                )
                 if file.last_modified < updated_at_from_history:
                     self._message_repository.emit_message(
                         AirbyteMessage(
@@ -294,7 +303,11 @@ def _compute_start_time(self) -> datetime:
             return datetime.min
         else:
             earliest = min(self._file_to_datetime_history.values())
-            earliest_dt = ab_datetime_parse(earliest, formats=[self.DATE_TIME_FORMAT]).to_datetime()
+            earliest_dt = ab_datetime_parse(
+                earliest,
+                formats=[self.DATE_TIME_FORMAT],
+                disallow_other_formats=False,
+            )
             if self._is_history_full():
                 time_window = datetime.now() - self._time_window_if_history_is_full
                 earliest_dt = min(earliest_dt, time_window)
diff --git a/airbyte_cdk/sources/file_based/stream/cursor/default_file_based_cursor.py b/airbyte_cdk/sources/file_based/stream/cursor/default_file_based_cursor.py
@@ -84,8 +84,10 @@ def _should_sync_file(self, file: RemoteFile, logger: logging.Logger) -> bool:
         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 = ab_datetime_parse(
-                self._file_to_datetime_history[file.uri], formats=[self.DATE_TIME_FORMAT]
-            ).to_datetime()
+                self._file_to_datetime_history[file.uri],
+                formats=[self.DATE_TIME_FORMAT],
+                disallow_other_formats=False,
+            )
             if file.last_modified < updated_at_from_history:
                 logger.warning(
                     f"The file {file.uri}'s last modified date is older than the last time it was synced. This is unexpected. Skipping the file."
@@ -135,8 +137,10 @@ def _compute_earliest_file_in_history(self) -> Optional[RemoteFile]:
             return RemoteFile(
                 uri=filename,
                 last_modified=ab_datetime_parse(
-                    last_modified, formats=[self.DATE_TIME_FORMAT]
-                ).to_datetime(),
+                    last_modified,
+                    formats=[self.DATE_TIME_FORMAT],
+                    disallow_other_formats=False,
+                ),
             )
         else:
             return None
@@ -146,7 +150,11 @@ def _compute_start_time(self) -> datetime:
             return datetime.min
         else:
             earliest = min(self._file_to_datetime_history.values())
-            earliest_dt = ab_datetime_parse(earliest, formats=[self.DATE_TIME_FORMAT]).to_datetime()
+            earliest_dt = ab_datetime_parse(
+                earliest,
+                formats=[self.DATE_TIME_FORMAT],
+                disallow_other_formats=False,
+            )
             if self._is_history_full():
                 time_window = datetime.now() - self._time_window_if_history_is_full
                 earliest_dt = min(earliest_dt, time_window)
diff --git a/airbyte_cdk/sources/streams/call_rate.py b/airbyte_cdk/sources/streams/call_rate.py
@@ -666,7 +666,7 @@ def get_reset_ts_from_response(
         if response.headers.get(self._ratelimit_reset_header):
             return AirbyteDateTime.fromtimestamp(
                 int(response.headers[self._ratelimit_reset_header])
-            ).to_datetime()
+            )
         return None
 
     def get_calls_left_from_response(self, response: requests.Response) -> Optional[int]:
diff --git a/airbyte_cdk/utils/datetime_format_inferrer.py b/airbyte_cdk/utils/datetime_format_inferrer.py
@@ -54,7 +54,11 @@ def _can_be_datetime(self, value: Any) -> bool:
     def _matches_format(self, value: Any, format: str) -> bool:
         """Checks if the value matches the format"""
         try:
-            dt = ab_datetime_try_parse(value, formats=[format], disallow_other_formats=True)
+            dt = ab_datetime_try_parse(
+                value,
+                formats=[format],
+                disallow_other_formats=True,
+            )
             return dt is not None
         except (ValueError, TypeError):
             return False
diff --git a/airbyte_cdk/utils/datetime_helpers.py b/airbyte_cdk/utils/datetime_helpers.py
diff --git a/unit_tests/utils/test_datetime_helpers.py b/unit_tests/utils/test_datetime_helpers.py
