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,9 @@
 from __future__ import annotations
 
 import re
+from datetime import UTC, datetime
 
-import pendulum
-from pendulum.datetime import DateTime
+from whenever import Date, Instant, Time, ZonedDateTime
 
 REGEX_MAPPING = {
     "seconds": r"(\d+)(s|sec|second|seconds)",
@@ -16,47 +16,67 @@ class TimestampFormatError(ValueError): ...
 
 
 class Timestamp:
-    def __init__(self, value: str | DateTime | Timestamp | None = None):
-        if value and isinstance(value, DateTime):
+    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
         elif isinstance(value, str):
             self.obj = self._parse_string(value)
         else:
-            self.obj = DateTime.now(tz="UTC")
+            self.obj = ZonedDateTime.now("UTC")
 
     @classmethod
-    def _parse_string(cls, value: str) -> DateTime:
+    def _parse_string(cls, value: str) -> ZonedDateTime:
+        try:
+            zoned_date = ZonedDateTime.parse_common_iso(value)
+            return zoned_date
+        except ValueError:
+            pass
+
         try:
-            parsed_date = pendulum.parse(value)
-            if isinstance(parsed_date, DateTime):
-                return parsed_date
-        except (pendulum.parsing.exceptions.ParserError, ValueError):
+            instant_date_ = Instant.parse_common_iso(value)
+            return instant_date_.to_tz("UTC")
+        except ValueError:
             pass
 
-        params = {}
+        try:
+            date = Date.parse_common_iso(value)
+            local_date = date.at(Time(12, 00))
+            return local_date.assume_tz("UTC")
+        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":
+        if with_z:
+            return self.obj.instant().format_common_iso()
+        iso8601_string = self.obj.format_common_iso()
+        if iso8601_string[-1] == "Z":
             iso8601_string = iso8601_string[:-1] + "+00:00"
         return iso8601_string
 
     def to_timestamp(self) -> int:
-        return self.obj.int_timestamp
+        return self.obj.timestamp()
+
+    def to_datetime(self) -> datetime:
+        time_str = self.to_string()
+        return datetime.strptime(time_str[:-1], "%Y-%m-%dT%H:%M:%S.%f").replace(tzinfo=UTC)
 
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, Timestamp):

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"