164 lines
4.6 KiB
Python
164 lines
4.6 KiB
Python
"""
|
|
Timezone utilities for consistent timestamp handling across the gateway.
|
|
Ensures all timestamps are properly handled as UTC.
|
|
"""
|
|
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import Union, Optional
|
|
|
|
def get_utc_now() -> datetime:
|
|
"""
|
|
Get current time in UTC with timezone info.
|
|
|
|
Returns:
|
|
datetime: Current UTC time with timezone info
|
|
"""
|
|
return datetime.now(timezone.utc)
|
|
|
|
def get_utc_timestamp() -> float:
|
|
"""
|
|
Get current UTC timestamp (seconds since epoch).
|
|
|
|
Returns:
|
|
float: Current UTC timestamp in seconds
|
|
"""
|
|
return datetime.now(timezone.utc).timestamp()
|
|
|
|
def to_utc_timestamp(dt: datetime) -> float:
|
|
"""
|
|
Convert datetime object to UTC timestamp.
|
|
|
|
Args:
|
|
dt (datetime): Datetime object to convert
|
|
|
|
Returns:
|
|
float: UTC timestamp in seconds
|
|
"""
|
|
if dt.tzinfo is None:
|
|
# If naive datetime, assume it's UTC
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt.timestamp()
|
|
|
|
def from_utc_timestamp(timestamp: Union[int, float]) -> datetime:
|
|
"""
|
|
Convert UTC timestamp to datetime object.
|
|
|
|
Args:
|
|
timestamp (Union[int, float]): UTC timestamp in seconds
|
|
|
|
Returns:
|
|
datetime: Datetime object in UTC
|
|
"""
|
|
return datetime.fromtimestamp(timestamp, tz=timezone.utc)
|
|
|
|
def add_seconds_to_utc(seconds: int) -> datetime:
|
|
"""
|
|
Add seconds to current UTC time.
|
|
|
|
Args:
|
|
seconds (int): Seconds to add (can be negative)
|
|
|
|
Returns:
|
|
datetime: UTC time with seconds added
|
|
"""
|
|
return get_utc_now() + timedelta(seconds=seconds)
|
|
|
|
def add_seconds_to_utc_timestamp(seconds: int) -> float:
|
|
"""
|
|
Add seconds to current UTC timestamp.
|
|
|
|
Args:
|
|
seconds (int): Seconds to add (can be negative)
|
|
|
|
Returns:
|
|
float: UTC timestamp with seconds added
|
|
"""
|
|
return get_utc_timestamp() + seconds
|
|
|
|
def format_utc_for_display(dt: datetime, format_str: str = "%Y-%m-%d %H:%M:%S UTC") -> str:
|
|
"""
|
|
Format UTC datetime for display.
|
|
|
|
Args:
|
|
dt (datetime): UTC datetime to format
|
|
format_str (str): Format string (default: ISO-like with UTC indicator)
|
|
|
|
Returns:
|
|
str: Formatted datetime string
|
|
"""
|
|
if dt.tzinfo is None:
|
|
# If naive datetime, assume it's UTC
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt.strftime(format_str)
|
|
|
|
def is_expired_utc(expires_at: Union[datetime, float, str]) -> bool:
|
|
"""
|
|
Check if a UTC timestamp has expired.
|
|
|
|
Args:
|
|
expires_at (Union[datetime, float, str]): Expiration timestamp
|
|
|
|
Returns:
|
|
bool: True if expired, False otherwise
|
|
"""
|
|
if not expires_at:
|
|
return False
|
|
|
|
current_utc = get_utc_timestamp()
|
|
|
|
if isinstance(expires_at, datetime):
|
|
expires_timestamp = to_utc_timestamp(expires_at)
|
|
elif isinstance(expires_at, str):
|
|
try:
|
|
# Try to parse ISO string
|
|
dt = datetime.fromisoformat(expires_at.replace('Z', '+00:00'))
|
|
expires_timestamp = to_utc_timestamp(dt)
|
|
except ValueError:
|
|
# If parsing fails, try float conversion
|
|
expires_timestamp = float(expires_at)
|
|
else:
|
|
expires_timestamp = float(expires_at)
|
|
|
|
return current_utc > expires_timestamp
|
|
|
|
def get_expires_in_seconds(expires_at: Union[datetime, float, str]) -> Optional[int]:
|
|
"""
|
|
Get seconds until expiration (negative if expired).
|
|
|
|
Args:
|
|
expires_at (Union[datetime, float, str]): Expiration timestamp
|
|
|
|
Returns:
|
|
Optional[int]: Seconds until expiration, None if no expiration
|
|
"""
|
|
if not expires_at:
|
|
return None
|
|
|
|
current_utc = get_utc_timestamp()
|
|
|
|
if isinstance(expires_at, datetime):
|
|
expires_timestamp = to_utc_timestamp(expires_at)
|
|
elif isinstance(expires_at, str):
|
|
try:
|
|
# Try to parse ISO string
|
|
dt = datetime.fromisoformat(expires_at.replace('Z', '+00:00'))
|
|
expires_timestamp = to_utc_timestamp(dt)
|
|
except ValueError:
|
|
# If parsing fails, try float conversion
|
|
expires_timestamp = float(expires_at)
|
|
else:
|
|
expires_timestamp = float(expires_at)
|
|
|
|
return int(expires_timestamp - current_utc)
|
|
|
|
def create_expiration_timestamp(expires_in_seconds: int) -> float:
|
|
"""
|
|
Create a new expiration timestamp from seconds until expiration.
|
|
|
|
Args:
|
|
expires_in_seconds (int): Seconds until expiration
|
|
|
|
Returns:
|
|
float: UTC timestamp in seconds
|
|
"""
|
|
return get_utc_timestamp() + expires_in_seconds
|