Coverage for functions \ flipdare \ util \ time_util.py: 78%
271 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
1#!/usr/bin/env python
2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved.
3#
4# This file is part of Flipdare's proprietary software and contains
5# confidential and copyrighted material. Unauthorised copying,
6# modification, distribution, or use of this file is strictly
7# prohibited without prior written permission from Flipdare Pty Ltd.
8#
9# This software includes third-party components licensed under MIT,
10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.
11#
13from dataclasses import dataclass
14import pytz
15from datetime import UTC, datetime, timedelta, tzinfo
16from google.cloud.firestore_v1 import SERVER_TIMESTAMP as FIRESTORE_SERVER_TIMESTAMP
17from google.cloud.firestore_v1.transforms import Sentinel
18from flipdare.app_types import DatabaseTimeType
20__all__ = ["TypesenseTime", "FirestoreTime", "TimeUtil"]
23@dataclass
24class DateRange:
25 from_date: datetime
26 to_date: datetime
29class TimeUtil:
31 def __init__(self) -> None:
32 pass
34 # ----------------------------------------------------------------------------------------------
35 # TIMEs
36 # ----------------------------------------------------------------------------------------------
38 @staticmethod
39 def get_current_utc_dt() -> datetime:
40 """Get the current UTC time."""
41 return datetime.now(UTC)
43 @staticmethod
44 def get_current_utc_timestamp() -> int:
45 """Get the current UTC timestamp in seconds since epoch."""
46 return int(datetime.now(UTC).timestamp())
48 @staticmethod
49 def get_current_utc_float_time() -> float:
50 """Get the current UTC time as a float timestamp (seconds since epoch)."""
51 return float(datetime.now(UTC).timestamp())
53 # ----------------------------------------------------------------------------------------------
54 # Modifiers
55 # ----------------------------------------------------------------------------------------------
57 @staticmethod
58 def epoch_to_user_dt(epoch: float, user_tz_str: str | None = None) -> datetime:
59 """Convert a float timestamp (seconds since epoch) to a datetime in user's timezone."""
60 utc_dt = TimeUtil.epoch_to_utc_dt(epoch)
61 if user_tz_str is None:
62 return utc_dt
64 user_tz = TimeUtil._get_user_tz(user_tz_str)
65 return utc_dt.astimezone(user_tz)
67 @staticmethod
68 def simple_epoch_to_utc_dt(epoch: int) -> datetime:
69 """Convert a simple integer timestamp (seconds since epoch) to a UTC datetime."""
70 return datetime.fromtimestamp(epoch, tz=UTC)
72 @staticmethod
73 def epoch_to_utc_dt(epoch: float) -> datetime:
74 """Convert a float timestamp (seconds since epoch) to a UTC datetime."""
75 return datetime.fromtimestamp(epoch, tz=UTC)
77 @staticmethod
78 def short_to_utc_dt(label: str) -> datetime:
79 return datetime.strptime(label, "%d_%B_%Y").replace(tzinfo=UTC)
81 @staticmethod
82 def dt_to_epoch(dt: datetime) -> int:
83 """Convert a datetime to an integer timestamp (seconds since epoch)."""
84 if dt.tzinfo is None:
85 dt = dt.replace(tzinfo=UTC)
86 return int(dt.timestamp())
88 @staticmethod
89 def dt_to_simple_epoch(dt: datetime) -> int:
90 """Convert a datetime to a simple integer timestamp (seconds since epoch)."""
91 if dt.tzinfo is None:
92 dt = dt.replace(tzinfo=UTC)
93 return int(dt.timestamp())
95 # ----------------------------------------------------------------------------------------------
96 # PAST Times
97 # ----------------------------------------------------------------------------------------------
99 @staticmethod
100 def get_start_of_day_utc() -> datetime:
101 """Get the start of the current day in UTC."""
102 now = datetime.now(UTC)
103 return datetime(year=now.year, month=now.month, day=now.day, tzinfo=UTC)
105 @staticmethod
106 def get_utc_time_days_ago(days: int) -> datetime:
107 """Get the UTC time a certain number of days ago."""
108 return datetime.now(UTC) - timedelta(days=days)
110 @staticmethod
111 def get_utc_time_days_ago_from(from_when: datetime, days: int) -> datetime:
112 """Get the UTC time a certain number of days ago from a specific datetime."""
113 return from_when - timedelta(days=days)
115 @staticmethod
116 def get_utc_time_hours_ago(hours: int) -> datetime:
117 """Get the UTC time a certain number of hours ago."""
118 return datetime.now(UTC) - timedelta(hours=hours)
120 @staticmethod
121 def get_utc_time_minutes_ago(minutes: int) -> datetime:
122 """Get the UTC time a certain number of minutes ago."""
123 return datetime.now(UTC) - timedelta(minutes=minutes)
125 @staticmethod
126 def get_utc_time_future_hours(from_when: datetime, hours: int) -> datetime:
127 """Get the UTC time a certain number of hours in the future from a specific datetime."""
128 return from_when + timedelta(hours=hours)
130 @staticmethod
131 def get_utc_time_future_days_now(days: int) -> datetime:
132 """Get the UTC time a certain number of days in the future."""
133 return datetime.now(UTC) + timedelta(days=days)
135 @staticmethod
136 def get_utc_time_future_days(from_when: datetime, days: int) -> datetime:
137 """Get the UTC time a certain number of days in the future."""
138 return from_when + timedelta(days=days)
140 # ----------------------------------------------------------------------------------------------
141 # Times as Strings
142 # ----------------------------------------------------------------------------------------------
144 @staticmethod
145 def formatted_now() -> str:
146 now = TimeUtil.get_current_utc_dt()
147 return TimeUtil.formatted_dt(now)
149 @staticmethod
150 def formatted_now_utc() -> str:
151 """Get the current UTC time as an ISO formatted string."""
152 return TimeUtil.formatted_dt(datetime.now(UTC))
154 @staticmethod
155 def formatted_dt(dt: datetime) -> str:
156 """Format a datetime object to a string."""
157 return dt.strftime("%Y-%m-%d %H:%M:%S UTC")
159 @staticmethod
160 def formatted_seconds(seconds: int) -> str:
161 """Format seconds into H:M:S string."""
162 hrs, rem = divmod(seconds, 3600)
163 mins, secs = divmod(rem, 60)
164 return f"{hrs}h {mins}m {secs}s"
166 @staticmethod
167 def formatted_epoch(epoch: float) -> str:
168 """Format a float timestamp (seconds since epoch) to a string."""
169 dt = TimeUtil.epoch_to_utc_dt(epoch)
170 return TimeUtil.formatted_dt(dt)
172 @staticmethod
173 def formatted_short(day: datetime) -> str:
174 """Format a datetime object to a simple string format (e.g., '1_January_2024')."""
175 return day.strftime("%#d_%B_%Y")
177 # ----------------------------------------------------------------------------------------------
178 # USER strings
179 # ----------------------------------------------------------------------------------------------
181 @staticmethod
182 def formatted_user(dt: datetime, user_tz_str: str | None = None) -> str:
183 """Format a datetime object to a string in user's timezone."""
184 from flipdare.app_log import LOG
186 user_tz = TimeUtil._get_user_tz(user_tz_str)
187 try:
188 user_dt = dt.astimezone(user_tz)
189 return user_dt.strftime("%Y-%m-%d %H:%M:%S")
190 except Exception as e:
191 LOG().error(f"Error formatting datetime for user timezone {user_tz_str}: {e}")
192 return TimeUtil.formatted_dt(dt)
194 @staticmethod
195 def formatted_user_accurate(dt: datetime, user_tz_str: str | None = None) -> str:
196 """Format a datetime object to a string in user's timezone with important format."""
197 from flipdare.app_log import LOG
199 if user_tz_str is None:
200 return dt.strftime("%#d %B %Y at %#I:%M %p UTC")
201 try:
202 user_tz = pytz.timezone(user_tz_str)
203 user_dt = dt.astimezone(user_tz)
204 return user_dt.strftime("%#d %B %Y at %#I:%M %p")
205 except Exception as e:
206 LOG().error(f"Error formatting datetime for user timezone {user_tz_str}: {e}")
207 return TimeUtil.formatted_dt(dt)
209 @staticmethod
210 def formatted_user_day(dt: datetime, user_tz_str: str | None = None) -> str:
211 """Format a datetime object to a string in user's timezone."""
212 from flipdare.app_log import LOG
214 if user_tz_str is None:
215 return dt.strftime("%#d %B %Y")
216 try:
217 user_tz = pytz.timezone(user_tz_str)
218 user_dt = dt.astimezone(user_tz)
219 return user_dt.strftime("%#d %B %Y")
220 except Exception as e:
221 LOG().error(f"Error formatting datetime for user timezone {user_tz_str}: {e}")
222 return TimeUtil.formatted_dt(dt)
224 # ----------------------------------------------------------------------------------------------
225 # TIME Ranges
226 # ----------------------------------------------------------------------------------------------
228 @staticmethod
229 def last_week_epoch() -> tuple[float, float]:
230 # 1. Start of 'this' week (Monday at 00:00:00)
231 now = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)
232 days_since_monday = now.weekday()
233 this_monday_start = now - timedelta(days=days_since_monday)
235 # 2. Last week's boundaries
236 last_monday_start = this_monday_start - timedelta(days=7)
237 last_sunday_end = this_monday_start # This Monday 00:00 is the end of last Sunday
239 return last_monday_start.timestamp(), last_sunday_end.timestamp()
241 # ----------------------------------------------------------------------------------------------
242 # Operations
243 # ----------------------------------------------------------------------------------------------
245 @staticmethod
246 def get_date_range(
247 days: int, start: datetime | None = None, reverse: bool = False
248 ) -> list[DateRange]:
249 """Get a list of datetime ranges for the past number of days."""
250 if start is None:
251 start = TimeUtil.get_start_of_day_utc()
252 ranges = [
253 DateRange(start - timedelta(days=i + 1), start - timedelta(days=i))
254 for i in range(days)
255 ]
256 if reverse:
257 ranges.reverse()
258 return ranges
260 @staticmethod
261 def duration_in_seconds(start: datetime, end: datetime) -> int:
262 """Calculate the duration in seconds between two datetime objects."""
263 if start.tzinfo is None:
264 start = start.replace(tzinfo=UTC)
265 if end.tzinfo is None:
266 end = end.replace(tzinfo=UTC)
267 return int((end - start).total_seconds())
269 @staticmethod
270 def is_older_than(
271 dt: datetime,
272 seconds: int | None = None,
273 minutes: int | None = None,
274 hours: int | None = None,
275 days: int | None = None,
276 ) -> bool:
277 """Check if a datetime is older than a certain number of seconds from now."""
278 if minutes is not None:
279 seconds = minutes * 60
280 if hours is not None:
281 seconds = hours * 3600
282 if days is not None:
283 seconds = days * 86400
284 if seconds is None:
285 raise ValueError("Either seconds, minutes, hours, or days must be provided.")
287 cutoff = datetime.now(UTC) - timedelta(seconds=seconds)
288 return dt < cutoff
290 @staticmethod
291 def is_newer_than(
292 dt: datetime,
293 seconds: int | None = None,
294 minutes: int | None = None,
295 hours: int | None = None,
296 ) -> bool:
297 """Check if a datetime is newer than a certain number of seconds from now."""
298 if minutes is not None:
299 seconds = minutes * 60
300 if hours is not None:
301 seconds = hours * 3600
302 if seconds is None:
303 raise ValueError("Either seconds, minutes, or hours must be provided.")
304 cutoff = datetime.now(UTC) - timedelta(seconds=seconds)
305 return dt > cutoff
307 # ----------------------------------------------------------------------------------------------
308 # Helpers
309 # ----------------------------------------------------------------------------------------------
311 @staticmethod
312 def _get_user_tz(user_tz_str: str | None = None) -> pytz.BaseTzInfo | tzinfo:
313 if user_tz_str is None:
314 return UTC
316 from flipdare.app_log import LOG
318 try:
319 return pytz.timezone(user_tz_str)
320 except Exception as e:
321 LOG().error(f"Error getting timezone for user timezone string {user_tz_str}: {e}")
322 return UTC
325class TypesenseTime:
327 @staticmethod
328 def server_timestamp() -> int:
329 return TimeUtil.get_current_utc_timestamp()
331 @staticmethod
332 def from_any(timestamp: int | Sentinel | datetime | str | None) -> int:
333 """Convert Firestore UTC time to integer timestamp (seconds since epoch)."""
334 if isinstance(timestamp, int):
335 return timestamp
337 if isinstance(timestamp, Sentinel):
338 return TypesenseTime.server_timestamp()
340 firestore_time = FirestoreTime.from_firestore(timestamp)
342 if firestore_time is None:
343 return TimeUtil.get_current_utc_timestamp()
344 if firestore_time.tzinfo is None:
345 firestore_time = firestore_time.replace(tzinfo=UTC)
346 return int(firestore_time.timestamp())
348 @staticmethod
349 def from_firestore(timestamp: DatabaseTimeType) -> int:
350 """Convert Firestore UTC time to integer timestamp (seconds since epoch)."""
351 if isinstance(timestamp, Sentinel):
352 return TypesenseTime.server_timestamp()
354 firestore_time = FirestoreTime.from_firestore(timestamp)
356 if firestore_time is None:
357 return TimeUtil.get_current_utc_timestamp()
358 if firestore_time.tzinfo is None:
359 firestore_time = firestore_time.replace(tzinfo=UTC)
360 return int(firestore_time.timestamp())
362 @staticmethod
363 def to_utc_datetime(timestamp: int | None = None) -> datetime | None:
364 """Convert integer timestamp (seconds since epoch) to UTC datetime."""
365 if timestamp is None:
366 return None
367 return datetime.fromtimestamp(timestamp, tz=UTC)
370class FirestoreTime:
372 @staticmethod
373 def server_timestamp() -> Sentinel:
374 return FIRESTORE_SERVER_TIMESTAMP
376 @staticmethod
377 def age_in_days(timestamp: DatabaseTimeType) -> int:
378 """Calculate the age of a Firestore timestamp from now in days."""
379 dt = FirestoreTime.from_firestore(timestamp)
380 if dt is None:
381 return 0
383 return (datetime.now(UTC) - dt).days
385 @staticmethod
386 def formatted(timestamp: DatabaseTimeType) -> str:
387 """Format Firestore timestamp to string."""
388 dt = FirestoreTime.from_firestore(timestamp)
389 if dt is None:
390 # most likely a new object with a sentinel..
391 # special case, since not stored in db, just set to current server time
392 dt = TimeUtil.get_current_utc_dt()
394 return dt.strftime("%Y-%m-%d %H:%M:%S UTC")
396 @staticmethod
397 def to_firestore(timestamp: DatabaseTimeType) -> Sentinel | datetime | None:
398 """
399 Convert various timestamp formats to Firestore timestamp format.
400 Note, this will be in UTC time.
401 """
402 from flipdare.app_log import LOG
404 if timestamp is None:
405 return FirestoreTime.server_timestamp()
407 try:
408 # Sentinel check
409 if isinstance(timestamp, Sentinel):
410 return timestamp
412 # Datetime check (handles DatetimeWithNanoseconds too)
413 if isinstance(timestamp, datetime):
414 # Convert to UTC if timezone-aware, add UTC if naive
415 return (
416 timestamp.astimezone(UTC)
417 if timestamp.tzinfo
418 else timestamp.replace(tzinfo=UTC)
419 )
421 # If it's a string, fromisoformat handles 'Z' and offsets automatically in 3.11+
422 # if isinstance(timestamp, str):
423 return datetime.fromisoformat(timestamp)
425 except Exception as error:
426 LOG().error(f"Error converting timestamp {timestamp}: {error}")
428 return None
430 @staticmethod
431 def internal_str(timestamp: DatabaseTimeType) -> str:
432 """
433 Convert Firestore timestamp to Python datetime.
434 Note, this will be in UTC time
435 """
436 timestamp = FirestoreTime.from_firestore(timestamp)
437 if timestamp is None:
438 return "N/A"
439 return TimeUtil.formatted_dt(timestamp)
441 @staticmethod
442 def from_firestore(timestamp: DatabaseTimeType) -> datetime | None:
443 """Convert Firestore timestamp to a standardized UTC datetime."""
444 match timestamp:
445 case None:
446 return None
448 # Sentinel check - ensure this is handled before attribute checks
449 case obj if "Sentinel" in str(type(obj)):
450 return None
452 case datetime() as dt:
453 # DatetimeWithNanoseconds is caught here as it's a datetime subclass.
454 # .astimezone(timezone.utc) is the safest method in 3.13.
455 return dt.astimezone(UTC) if dt.tzinfo else dt.replace(tzinfo=UTC)
457 case str() as ts_str:
458 try:
459 # 3.11+ handles 'Z' and ISO formats natively
460 dt = datetime.fromisoformat(ts_str.removesuffix(" UTC"))
461 # If naive (no tzinfo), assume it's already UTC; if aware, convert to UTC
462 return dt.replace(tzinfo=UTC) if not dt.tzinfo else dt.astimezone(UTC)
463 except ValueError:
464 return None
466 # 2. Use getattr to safely check for attributes on unknown objects.
467 # This will only be reached if the object is NOT a datetime or str.
468 case obj if (s := getattr(obj, "seconds", None)) is not None and (
469 ns := getattr(obj, "nanoseconds", None)
470 ) is not None:
471 return datetime.fromtimestamp(s + ns / 1e9, tz=UTC)
473 case _:
474 return None