chore: remove pendulum dependency (#258)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Aaron <AJ> Steers <[email protected]>
Co-authored-by: octavia-squidington-iii <[email protected]>

Author: 3 people

airbyte_cdk/cli/source_declarative_manifest/_run.py

@@ -21,7 +21,6 @@
 import sys
 import traceback
 from collections.abc import Mapping
-from datetime import datetime
 from pathlib import Path
 from typing import Any, cast
 
@@ -44,6 +43,7 @@
 )
 from airbyte_cdk.sources.declarative.yaml_declarative_source import YamlDeclarativeSource
 from airbyte_cdk.sources.source import TState
+from airbyte_cdk.utils.datetime_helpers import ab_datetime_now
 
 
 class SourceLocalYaml(YamlDeclarativeSource):
@@ -101,7 +101,7 @@ def _get_local_yaml_source(args: list[str]) -> SourceLocalYaml:
                         type=Type.TRACE,
                         trace=AirbyteTraceMessage(
                             type=TraceType.ERROR,
-                            emitted_at=int(datetime.now().timestamp() * 1000),
+                            emitted_at=ab_datetime_now().to_epoch_millis(),
                             error=AirbyteErrorTraceMessage(
                                 message=f"Error starting the sync. This could be due to an invalid configuration or catalog. Please contact Support for assistance. Error: {error}",
                                 stack_trace=traceback.format_exc(),
@@ -191,7 +191,7 @@ def create_declarative_source(
                         type=Type.TRACE,
                         trace=AirbyteTraceMessage(
                             type=TraceType.ERROR,
-                            emitted_at=int(datetime.now().timestamp() * 1000),
+                            emitted_at=ab_datetime_now().to_epoch_millis(),
                             error=AirbyteErrorTraceMessage(
                                 message=f"Error starting the sync. This could be due to an invalid configuration or catalog. Please contact Support for assistance. Error: {error}",
                                 stack_trace=traceback.format_exc(),

airbyte_cdk/connector_builder/connector_builder_handler.py

@@ -3,7 +3,6 @@
 #
 
 import dataclasses
-from datetime import datetime
 from typing import Any, List, Mapping
 
 from airbyte_cdk.connector_builder.message_grouper import MessageGrouper
@@ -21,6 +20,7 @@
     ModelToComponentFactory,
 )
 from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets
+from airbyte_cdk.utils.datetime_helpers import ab_datetime_now
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
 DEFAULT_MAXIMUM_NUMBER_OF_PAGES_PER_SLICE = 5
@@ -114,4 +114,4 @@ def resolve_manifest(source: ManifestDeclarativeSource) -> AirbyteMessage:
 
 
 def _emitted_at() -> int:
-    return int(datetime.now().timestamp()) * 1000
+    return ab_datetime_now().to_epoch_millis()

airbyte_cdk/sources/declarative/auth/oauth.py

@@ -3,10 +3,9 @@
 #
 
 from dataclasses import InitVar, dataclass, field
+from datetime import timedelta
 from typing import Any, List, Mapping, MutableMapping, Optional, Union
 
-import pendulum
-
 from airbyte_cdk.sources.declarative.auth.declarative_authenticator import DeclarativeAuthenticator
 from airbyte_cdk.sources.declarative.interpolation.interpolated_boolean import InterpolatedBoolean
 from airbyte_cdk.sources.declarative.interpolation.interpolated_mapping import InterpolatedMapping
@@ -18,6 +17,7 @@
 from airbyte_cdk.sources.streams.http.requests_native_auth.oauth import (
     SingleUseRefreshTokenOauth2Authenticator,
 )
+from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse
 
 
 @dataclass
@@ -53,7 +53,7 @@ class DeclarativeOauth2Authenticator(AbstractOauth2Authenticator, DeclarativeAut
     refresh_token: Optional[Union[InterpolatedString, str]] = None
     scopes: Optional[List[str]] = None
     token_expiry_date: Optional[Union[InterpolatedString, str]] = None
-    _token_expiry_date: Optional[pendulum.DateTime] = field(init=False, repr=False, default=None)
+    _token_expiry_date: Optional[AirbyteDateTime] = field(init=False, repr=False, default=None)
     token_expiry_date_format: Optional[str] = None
     token_expiry_is_time_of_expiration: bool = False
     access_token_name: Union[InterpolatedString, str] = "access_token"
@@ -122,15 +122,24 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self._refresh_request_headers = InterpolatedMapping(
             self.refresh_request_headers or {}, parameters=parameters
         )
-        self._token_expiry_date: pendulum.DateTime = (
-            pendulum.parse(
-                InterpolatedString.create(self.token_expiry_date, parameters=parameters).eval(
-                    self.config
+        try:
+            if (
+                isinstance(self.token_expiry_date, (int, str))
+                and str(self.token_expiry_date).isdigit()
+            ):
+                self._token_expiry_date = ab_datetime_parse(self.token_expiry_date)
+            else:
+                self._token_expiry_date = (
+                    ab_datetime_parse(
+                        InterpolatedString.create(
+                            self.token_expiry_date, parameters=parameters
+                        ).eval(self.config)
+                    )
+                    if self.token_expiry_date
+                    else ab_datetime_now() - timedelta(days=1)
                 )
-            )  # type: ignore # pendulum.parse returns a datetime in this context
-            if self.token_expiry_date
-            else pendulum.now().subtract(days=1)  # type: ignore # substract does not have type hints
-        )
+        except ValueError as e:
+            raise ValueError(f"Invalid token expiry date format: {e}")
         self.use_profile_assertion = (
             InterpolatedBoolean(self.use_profile_assertion, parameters=parameters)
             if isinstance(self.use_profile_assertion, str)
@@ -222,8 +231,8 @@ def get_refresh_request_body(self) -> Mapping[str, Any]:
     def get_refresh_request_headers(self) -> Mapping[str, Any]:
         return self._refresh_request_headers.eval(self.config)
 
-    def get_token_expiry_date(self) -> pendulum.DateTime:
-        return self._token_expiry_date  # type: ignore # _token_expiry_date is a pendulum.DateTime. It is never None despite what mypy thinks
+    def get_token_expiry_date(self) -> AirbyteDateTime:
+        return self._token_expiry_date  # type: ignore # _token_expiry_date is an AirbyteDateTime. It is never None despite what mypy thinks
 
     def set_token_expiry_date(self, value: Union[str, int]) -> None:
         self._token_expiry_date = self._parse_token_expiration_date(value)

airbyte_cdk/sources/declarative/auth/token_provider.py

@@ -9,9 +9,7 @@
 from typing import Any, List, Mapping, Optional, Union
 
 import dpath
-import pendulum
 from isodate import Duration
-from pendulum import DateTime
 
 from airbyte_cdk.sources.declarative.decoders.decoder import Decoder
 from airbyte_cdk.sources.declarative.decoders.json_decoder import JsonDecoder
@@ -21,6 +19,7 @@
 from airbyte_cdk.sources.http_logger import format_http_message
 from airbyte_cdk.sources.message import MessageRepository, NoopMessageRepository
 from airbyte_cdk.sources.types import Config
+from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now
 
 
 class TokenProvider:
@@ -38,7 +37,7 @@ class SessionTokenProvider(TokenProvider):
     message_repository: MessageRepository = NoopMessageRepository()
     decoder: Decoder = field(default_factory=lambda: JsonDecoder(parameters={}))
 
-    _next_expiration_time: Optional[DateTime] = None
+    _next_expiration_time: Optional[AirbyteDateTime] = None
     _token: Optional[str] = None
 
     def get_token(self) -> str:
@@ -48,7 +47,7 @@ def get_token(self) -> str:
         return self._token
 
     def _refresh_if_necessary(self) -> None:
-        if self._next_expiration_time is None or self._next_expiration_time < pendulum.now():
+        if self._next_expiration_time is None or self._next_expiration_time < ab_datetime_now():
             self._refresh()
 
     def _refresh(self) -> None:
@@ -65,7 +64,7 @@ def _refresh(self) -> None:
             raise ReadException("Failed to get session token, response got ignored by requester")
         session_token = dpath.get(next(self.decoder.decode(response)), self.session_token_path)
         if self.expiration_duration is not None:
-            self._next_expiration_time = pendulum.now() + self.expiration_duration
+            self._next_expiration_time = ab_datetime_now() + self.expiration_duration
         self._token = session_token  # type: ignore # Returned decoded response will be Mapping and therefore session_token will be str or None
 
 

airbyte_cdk/sources/streams/concurrent/state_converters/datetime_stream_state_converter.py

@@ -6,9 +6,6 @@
 from datetime import datetime, timedelta, timezone
 from typing import Any, Callable, List, MutableMapping, Optional, Tuple
 
-import pendulum
-from pendulum.datetime import DateTime
-
 # 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
@@ -17,6 +14,7 @@
     AbstractStreamStateConverter,
     ConcurrencyCompatibleStateType,
 )
+from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse
 
 
 class DateTimeStreamStateConverter(AbstractStreamStateConverter):
@@ -36,7 +34,7 @@ def zero_value(self) -> datetime:
 
     @classmethod
     def get_end_provider(cls) -> Callable[[], datetime]:
-        return lambda: datetime.now(timezone.utc)
+        return ab_datetime_now
 
     @abstractmethod
     def increment(self, timestamp: datetime) -> datetime: ...
@@ -136,10 +134,10 @@ def output_format(self, timestamp: datetime) -> int:
         return int(timestamp.timestamp())
 
     def parse_timestamp(self, timestamp: int) -> datetime:
-        dt_object = pendulum.from_timestamp(timestamp)
-        if not isinstance(dt_object, DateTime):
+        dt_object = AirbyteDateTime.fromtimestamp(timestamp, timezone.utc)
+        if not isinstance(dt_object, AirbyteDateTime):
             raise ValueError(
-                f"DateTime object was expected but got {type(dt_object)} from pendulum.parse({timestamp})"
+                f"AirbyteDateTime object was expected but got {type(dt_object)} from AirbyteDateTime.fromtimestamp({timestamp})"
             )
         return dt_object
 
@@ -169,22 +167,33 @@ def __init__(
     def increment(self, timestamp: datetime) -> datetime:
         return timestamp + self._cursor_granularity
 
-    def output_format(self, timestamp: datetime) -> Any:
-        return timestamp.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
+    def output_format(self, timestamp: datetime) -> str:
+        """Format datetime with milliseconds always included.
+
+        Args:
+            timestamp: The datetime to format.
+
+        Returns:
+            str: ISO8601/RFC3339 formatted string with milliseconds.
+        """
+        dt = AirbyteDateTime.from_datetime(timestamp)
+        # Always include milliseconds, even if zero
+        millis = dt.microsecond // 1000 if dt.microsecond else 0
+        return f"{dt.year:04d}-{dt.month:02d}-{dt.day:02d}T{dt.hour:02d}:{dt.minute:02d}:{dt.second:02d}.{millis:03d}Z"
 
     def parse_timestamp(self, timestamp: str) -> datetime:
-        dt_object = pendulum.parse(timestamp)
-        if not isinstance(dt_object, DateTime):
+        dt_object = ab_datetime_parse(timestamp)
+        if not isinstance(dt_object, AirbyteDateTime):
             raise ValueError(
-                f"DateTime object was expected but got {type(dt_object)} from pendulum.parse({timestamp})"
+                f"AirbyteDateTime object was expected but got {type(dt_object)} from parse({timestamp})"
             )
         return dt_object
 
 
 class CustomFormatConcurrentStreamStateConverter(IsoMillisConcurrentStreamStateConverter):
     """
     Datetime State converter that emits state according to the supplied datetime format. The converter supports reading
-    incoming state in any valid datetime format via Pendulum.
+    incoming state in any valid datetime format using AirbyteDateTime parsing utilities.
     """
 
     def __init__(

airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py

@@ -4,11 +4,11 @@
 
 import logging
 from abc import abstractmethod
+from datetime import timedelta
 from json import JSONDecodeError
 from typing import Any, List, Mapping, MutableMapping, Optional, Tuple, Union
 
 import backoff
-import pendulum
 import requests
 from requests.auth import AuthBase
 
@@ -17,6 +17,7 @@
 from airbyte_cdk.sources.message import MessageRepository, NoopMessageRepository
 from airbyte_cdk.utils import AirbyteTracedException
 from airbyte_cdk.utils.airbyte_secrets_utils import add_to_secrets
+from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse
 
 from ..exceptions import DefaultBackoffException
 
@@ -72,7 +73,7 @@ def get_access_token(self) -> str:
 
     def token_has_expired(self) -> bool:
         """Returns True if the token is expired"""
-        return pendulum.now() > self.get_token_expiry_date()  # type: ignore # this is always a bool despite what mypy thinks
+        return ab_datetime_now() > self.get_token_expiry_date()
 
     def build_refresh_request_body(self) -> Mapping[str, Any]:
         """
@@ -179,7 +180,7 @@ def refresh_access_token(self) -> Tuple[str, Union[str, int]]:
             self.get_expires_in_name()
         ]
 
-    def _parse_token_expiration_date(self, value: Union[str, int]) -> pendulum.DateTime:
+    def _parse_token_expiration_date(self, value: Union[str, int]) -> AirbyteDateTime:
         """
         Return the expiration datetime of the refresh token
 
@@ -191,9 +192,19 @@ def _parse_token_expiration_date(self, value: Union[str, int]) -> pendulum.DateT
                 raise ValueError(
                     f"Invalid token expiry date format {self.token_expiry_date_format}; a string representing the format is required."
                 )
-            return pendulum.from_format(str(value), self.token_expiry_date_format)
+            try:
+                return ab_datetime_parse(str(value))
+            except ValueError as e:
+                raise ValueError(f"Invalid token expiry date format: {e}")
         else:
-            return pendulum.now().add(seconds=int(float(value)))
+            try:
+                # Only accept numeric values (as int/float/string) when no format specified
+                seconds = int(float(str(value)))
+                return ab_datetime_now() + timedelta(seconds=seconds)
+            except (ValueError, TypeError):
+                raise ValueError(
+                    f"Invalid expires_in value: {value}. Expected number of seconds when no format specified."
+                )
 
     @property
     def token_expiry_is_time_of_expiration(self) -> bool:
@@ -244,7 +255,7 @@ def get_scopes(self) -> List[str]:
         """List of requested scopes"""
 
     @abstractmethod
-    def get_token_expiry_date(self) -> pendulum.DateTime:
+    def get_token_expiry_date(self) -> AirbyteDateTime:
         """Expiration date of the access token"""
 
     @abstractmethod