104 lines
No EOL
3.3 KiB
Python
104 lines
No EOL
3.3 KiB
Python
"""
|
|
Timezone utilities for consistent timestamp handling across the gateway.
|
|
Ensures all timestamps are properly handled as UTC.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Optional, Any
|
|
import time
|
|
import logging
|
|
|
|
# Configure logger
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def getUtcNow() -> datetime:
|
|
"""
|
|
Get current time in UTC with timezone info.
|
|
|
|
Returns:
|
|
datetime: Current UTC time with timezone info
|
|
"""
|
|
return datetime.now(timezone.utc)
|
|
|
|
def getUtcTimestamp() -> float:
|
|
"""
|
|
Get current UTC timestamp (seconds since epoch with millisecond precision).
|
|
|
|
Returns:
|
|
float: Current UTC timestamp in seconds with millisecond precision
|
|
"""
|
|
return time.time()
|
|
|
|
def createExpirationTimestamp(expiresInSeconds: int) -> float:
|
|
"""
|
|
Create a new expiration timestamp from seconds until expiration.
|
|
|
|
Args:
|
|
expiresInSeconds (int): Seconds until expiration
|
|
|
|
Returns:
|
|
float: UTC timestamp in seconds
|
|
"""
|
|
return getUtcTimestamp() + expiresInSeconds
|
|
|
|
def parseTimestamp(value: Any, default: Optional[float] = None) -> Optional[float]:
|
|
"""
|
|
Parse a timestamp value from various source formats and return as float.
|
|
|
|
Handles conversion from:
|
|
- float: Returns as-is
|
|
- int: Converts to float
|
|
- str: Attempts to parse as numeric string (e.g., "1234567890.123")
|
|
- None: Returns default value (or None if no default)
|
|
|
|
This function is useful when database connectors (e.g., psycopg2) may return
|
|
numeric fields as strings in some environments (e.g., Azure PostgreSQL).
|
|
|
|
Args:
|
|
value: The timestamp value to parse (can be float, int, str, or None)
|
|
default: Optional default value to return if value is None or invalid.
|
|
If None and value is invalid, returns None.
|
|
|
|
Returns:
|
|
float: Parsed timestamp as float, or default value if provided, or None
|
|
|
|
Examples:
|
|
>>> parseTimestamp(1234567890.123)
|
|
1234567890.123
|
|
>>> parseTimestamp("1234567890.123")
|
|
1234567890.123
|
|
>>> parseTimestamp(None, default=0.0)
|
|
0.0
|
|
>>> parseTimestamp("invalid", default=getUtcTimestamp())
|
|
# Returns current timestamp
|
|
"""
|
|
if value is None:
|
|
return default
|
|
|
|
# Already a float
|
|
if isinstance(value, float):
|
|
return value
|
|
|
|
# Integer - convert to float
|
|
if isinstance(value, int):
|
|
return float(value)
|
|
|
|
# String - try to parse as numeric
|
|
if isinstance(value, str):
|
|
# Empty string
|
|
if not value.strip():
|
|
logger.warning(f"parseTimestamp: Received empty string, returning default={default}")
|
|
return default
|
|
try:
|
|
return float(value)
|
|
except (ValueError, TypeError) as e:
|
|
# Invalid string format
|
|
logger.warning(f"parseTimestamp: Failed to parse string '{value}' as float: {type(e).__name__}: {str(e)}, returning default={default}")
|
|
return default
|
|
|
|
# Unknown type - try to convert anyway
|
|
try:
|
|
return float(value)
|
|
except (ValueError, TypeError) as e:
|
|
logger.warning(f"parseTimestamp: Failed to convert value of type {type(value).__name__} '{value}' to float: {type(e).__name__}: {str(e)}, returning default={default}")
|
|
return default |