Replace pendulum with whenever

Author: dgarros

infrahub_sdk/ctl/branch.py

@@ -5,9 +5,10 @@
 from rich.table import Table
 
 from ..async_typer import AsyncTyper
-from ..ctl.client import initialize_client
-from ..ctl.utils import calculate_time_diff, catch_exception
+from ..utils import calculate_time_diff
+from .client import initialize_client
 from .parameters import CONFIG_PARAM
+from .utils import catch_exception
 
 app = AsyncTyper()
 console = Console()

infrahub_sdk/ctl/utils.py

@@ -8,11 +8,9 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable, NoReturn, Optional, TypeVar
 
-import pendulum
 import typer
 from click.exceptions import Exit
 from httpx import HTTPError
-from pendulum.datetime import DateTime
 from rich.console import Console
 from rich.logging import RichHandler
 from rich.markup import escape
@@ -152,20 +150,6 @@ def parse_cli_vars(variables: Optional[list[str]]) -> dict[str, str]:
     return {var.split("=")[0]: var.split("=")[1] for var in variables if "=" in var}
 
 
-def calculate_time_diff(value: str) -> str | None:
-    """Calculate the time in human format between a timedate in string format and now."""
-    try:
-        time_value = pendulum.parse(value)
-    except pendulum.parsing.exceptions.ParserError:
-        return None
-
-    if not isinstance(time_value, DateTime):
-        return None
-
-    pendulum.set_locale("en")
-    return time_value.diff_for_humans(other=pendulum.now(), absolute=True)
-
-
 def find_graphql_query(name: str, directory: str | Path = ".") -> str:
     if isinstance(directory, str):
         directory = Path(directory)

infrahub_sdk/timestamp.py

@@ -1,9 +1,12 @@
 from __future__ import annotations
 
 import re
+import warnings
+from datetime import datetime, timezone
 
-import pendulum
-from pendulum.datetime import DateTime
+from whenever import Date, Instant, LocalDateTime, Time, ZonedDateTime
+
+UTC = timezone.utc  # Required for older versions of Python
 
 REGEX_MAPPING = {
     "seconds": r"(\d+)(s|sec|second|seconds)",
@@ -16,76 +19,107 @@ class TimestampFormatError(ValueError): ...
 
 
 class Timestamp:
-    def __init__(self, value: str | DateTime | Timestamp | None = None):
-        if value and isinstance(value, DateTime):
-            self.obj = value
+    _obj: ZonedDateTime
+
+    def __init__(self, value: str | ZonedDateTime | Timestamp | None = None):
+        if value and isinstance(value, ZonedDateTime):
+            self._obj = value
         elif value and isinstance(value, self.__class__):
-            self.obj = value.obj
+            self._obj = value._obj
         elif isinstance(value, str):
-            self.obj = self._parse_string(value)
+            self._obj = self._parse_string(value)
         else:
-            self.obj = DateTime.now(tz="UTC")
+            self._obj = ZonedDateTime.now("UTC")
+
+    @property
+    def obj(self) -> ZonedDateTime:
+        warnings.warn(
+            "Direct access to obj property is deprecated. Use to_string(), to_timestamp(), or to_datetime() instead.",
+            UserWarning,
+            stacklevel=2,
+        )
+        return self._obj
 
     @classmethod
-    def _parse_string(cls, value: str) -> DateTime:
+    def _parse_string(cls, value: str) -> ZonedDateTime:
         try:
-            parsed_date = pendulum.parse(value)
-            if isinstance(parsed_date, DateTime):
-                return parsed_date
-        except (pendulum.parsing.exceptions.ParserError, ValueError):
+            zoned_date = ZonedDateTime.parse_common_iso(value)
+            return zoned_date
+        except ValueError:
             pass
 
-        params = {}
+        try:
+            instant_date = Instant.parse_common_iso(value)
+            return instant_date.to_tz("UTC")
+        except ValueError:
+            pass
+
+        try:
+            local_date_time = LocalDateTime.parse_common_iso(value)
+            return local_date_time.assume_utc().to_tz("UTC")
+        except ValueError:
+            pass
+
+        try:
+            date = Date.parse_common_iso(value)
+            local_date = date.at(Time(12, 00))
+            return local_date.assume_tz("UTC", disambiguate="compatible")
+        except ValueError:
+            pass
+
+        params: dict[str, float] = {}
         for key, regex in REGEX_MAPPING.items():
             match = re.search(regex, value)
             if match:
-                params[key] = int(match.group(1))
+                params[key] = float(match.group(1))
 
-        if not params:
-            raise TimestampFormatError(f"Invalid time format for {value}")
+        if params:
+            return ZonedDateTime.now("UTC").subtract(**params)  # type: ignore[call-overload]
 
-        return DateTime.now(tz="UTC").subtract(**params)
+        raise TimestampFormatError(f"Invalid time format for {value}")
 
     def __repr__(self) -> str:
         return f"Timestamp: {self.to_string()}"
 
     def to_string(self, with_z: bool = True) -> str:
-        iso8601_string = self.obj.to_iso8601_string()
-        if not with_z and iso8601_string[-1] == "Z":
-            iso8601_string = iso8601_string[:-1] + "+00:00"
-        return iso8601_string
+        if with_z:
+            return self._obj.instant().format_common_iso()
+        return self.to_datetime().isoformat()
 
     def to_timestamp(self) -> int:
-        return self.obj.int_timestamp
+        return self._obj.timestamp()
+
+    def to_datetime(self) -> datetime:
+        return self._obj.py_datetime()
 
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, Timestamp):
             return NotImplemented
-        return self.obj == other.obj
+        return self._obj == other._obj
 
     def __lt__(self, other: object) -> bool:
         if not isinstance(other, Timestamp):
             return NotImplemented
-        return self.obj < other.obj
+        return self._obj < other._obj
 
     def __gt__(self, other: object) -> bool:
         if not isinstance(other, Timestamp):
             return NotImplemented
-        return self.obj > other.obj
+        return self._obj > other._obj
 
     def __le__(self, other: object) -> bool:
         if not isinstance(other, Timestamp):
             return NotImplemented
-        return self.obj <= other.obj
+        return self._obj <= other._obj
 
     def __ge__(self, other: object) -> bool:
         if not isinstance(other, Timestamp):
             return NotImplemented
-        return self.obj >= other.obj
+        return self._obj >= other._obj
 
     def __hash__(self) -> int:
         return hash(self.to_string())
 
     def add_delta(self, hours: int = 0, minutes: int = 0, seconds: int = 0, microseconds: int = 0) -> Timestamp:
-        time = self.obj.add(hours=hours, minutes=minutes, seconds=seconds, microseconds=microseconds)
+        time = self._obj.add(hours=hours, minutes=minutes, seconds=seconds, microseconds=microseconds)
         return Timestamp(time)

infrahub_sdk/utils.py

@@ -18,9 +18,11 @@
 from infrahub_sdk.repository import GitRepoManager
 
 from .exceptions import FileNotValidError, JsonDecodeError
+from .timestamp import Timestamp, TimestampFormatError
 
 if TYPE_CHECKING:
     from graphql import GraphQLResolveInfo
+    from whenever import TimeDelta
 
 
 def base36encode(number: int) -> str:
@@ -367,3 +369,26 @@ def get_user_permissions(data: list[dict]) -> dict:
         groups[group_name] = permissions
 
     return groups
+
+
+def calculate_time_diff(value: str) -> str | None:
+    """Calculate the time in human format between a timedate in string format and now."""
+    try:
+        time_value = Timestamp(value)
+    except TimestampFormatError:
+        return None
+
+    delta: TimeDelta = Timestamp().obj.difference(time_value.obj)
+    (hrs, mins, secs, _) = delta.in_hrs_mins_secs_nanos()
+
+    if hrs and hrs < 24 and mins:
+        return f"{hrs}h {mins}m and {secs}s ago"
+    if hrs and hrs > 24:
+        remaining_hrs = hrs % 24
+        days = int((hrs - remaining_hrs) / 24)
+        return f"{days}d and {remaining_hrs}h ago"
+    if hrs == 0 and mins and secs:
+        return f"{mins}m and {secs}s ago"
+    if hrs == 0 and mins == 0 and secs:
+        return f"{secs}s ago"
+    return "now"