centralized timestamp management with logger

This commit is contained in:
ValueOn AG 2025-11-06 14:27:02 +01:00
parent a225c17841
commit 88110d0f9d
26 changed files with 178 additions and 94 deletions

View file

@ -7,7 +7,7 @@ from pydantic import BaseModel
import threading
import time
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
logger = logging.getLogger(__name__)

View file

@ -6,7 +6,7 @@ import uuid
from pydantic import BaseModel, Field
import threading
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
from modules.shared.configuration import APP_CONFIG
logger = logging.getLogger(__name__)

View file

@ -4,7 +4,7 @@ from typing import List, Dict, Any, Optional
from enum import Enum
from pydantic import BaseModel, Field
from modules.shared.attributeUtils import registerModelLabels
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
import uuid

View file

@ -3,7 +3,7 @@
from typing import Dict, Any, Optional, Union
from pydantic import BaseModel, Field
from modules.shared.attributeUtils import registerModelLabels
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
import uuid
import base64

View file

@ -3,7 +3,7 @@
from typing import Optional
from pydantic import BaseModel, Field
from modules.shared.attributeUtils import registerModelLabels
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
from .datamodelUam import AuthAuthority
from enum import Enum
import uuid

View file

@ -5,7 +5,7 @@ from typing import Optional
from enum import Enum
from pydantic import BaseModel, Field, EmailStr
from modules.shared.attributeUtils import registerModelLabels
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
class AuthAuthority(str, Enum):

View file

@ -2,7 +2,7 @@
from pydantic import BaseModel, Field
from modules.shared.attributeUtils import registerModelLabels
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
import uuid

View file

@ -11,7 +11,7 @@ import uuid
from modules.connectors.connectorDbPostgre import DatabaseConnector
from modules.shared.configuration import APP_CONFIG
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp, parseTimestamp
from modules.interfaces.interfaceDbAppAccess import AppAccess
from modules.datamodels.datamodelUam import (
User,
@ -1019,7 +1019,7 @@ class AppObjects:
return None
# Sort by expiration date and get the latest (most recent expiration)
tokens.sort(key=lambda x: x.get("expiresAt", 0), reverse=True)
tokens.sort(key=lambda x: parseTimestamp(x.get("expiresAt"), default=0), reverse=True)
latest_token = Token(**tokens[0])
# No auto-refresh here. Callers should use a higher-level service to refresh when needed.
@ -1170,10 +1170,8 @@ class AppObjects:
all_tokens = self.db.getRecordset(Token, recordFilter={})
for token_data in all_tokens:
if (
token_data.get("expiresAt")
and token_data.get("expiresAt") < current_time
):
expiresAt = parseTimestamp(token_data.get("expiresAt"))
if expiresAt and expiresAt < current_time:
# Token is expired, delete it
self.db.recordDelete(Token, token_data["id"])
cleaned_count += 1

View file

@ -27,7 +27,7 @@ from modules.datamodels.datamodelUam import User
# DYNAMIC PART: Connectors to the Interface
from modules.connectors.connectorDbPostgre import DatabaseConnector
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp, parseTimestamp
from modules.datamodels.datamodelPagination import PaginationParams, PaginatedResult
# Basic Configurations
@ -995,7 +995,7 @@ class ChatObjects:
# Apply default sorting by timestamp if no sort specified
if pagination is None or not pagination.sort:
logDicts.sort(key=lambda x: float(x.get("timestamp", 0)))
logDicts.sort(key=lambda x: parseTimestamp(x.get("timestamp"), default=0))
# Apply filtering (if filters provided)
if pagination and pagination.filters:
@ -1143,15 +1143,15 @@ class ChatObjects:
messages = self.db.getRecordset(ChatMessage, recordFilter={"workflowId": workflowId})
for msg in messages:
# Apply timestamp filtering in Python
msg_timestamp = msg.get("publishedAt", getUtcTimestamp())
if afterTimestamp is not None and msg_timestamp <= afterTimestamp:
msgTimestamp = parseTimestamp(msg.get("publishedAt"), default=getUtcTimestamp())
if afterTimestamp is not None and msgTimestamp <= afterTimestamp:
continue
# Load documents for each message
documents = self.getDocuments(msg["id"])
# Create ChatMessage object with loaded documents
chat_message = ChatMessage(
chatMessage = ChatMessage(
id=msg["id"],
workflowId=msg["workflowId"],
parentMessageId=msg.get("parentMessageId"),
@ -1176,23 +1176,23 @@ class ChatObjects:
# Use publishedAt as the timestamp for chronological ordering
items.append({
"type": "message",
"createdAt": msg_timestamp,
"item": chat_message
"createdAt": msgTimestamp,
"item": chatMessage
})
# Get logs
logs = self.db.getRecordset(ChatLog, recordFilter={"workflowId": workflowId})
for log in logs:
# Apply timestamp filtering in Python
log_timestamp = log.get("timestamp", getUtcTimestamp())
if afterTimestamp is not None and log_timestamp <= afterTimestamp:
logTimestamp = parseTimestamp(log.get("timestamp"), default=getUtcTimestamp())
if afterTimestamp is not None and logTimestamp <= afterTimestamp:
continue
chat_log = ChatLog(**log)
chatLog = ChatLog(**log)
items.append({
"type": "log",
"createdAt": log_timestamp,
"item": chat_log
"createdAt": logTimestamp,
"item": chatLog
})
# Get stats list
@ -1210,7 +1210,7 @@ class ChatObjects:
})
# Sort all items by createdAt timestamp for chronological order
items.sort(key=lambda x: x["createdAt"])
items.sort(key=lambda x: parseTimestamp(x.get("createdAt"), default=0))
return {"items": items}

View file

@ -17,7 +17,7 @@ from modules.datamodels.datamodelUtils import Prompt
from modules.datamodels.datamodelVoice import VoiceSettings
from modules.datamodels.datamodelUam import User, Mandate
from modules.shared.configuration import APP_CONFIG
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
from modules.datamodels.datamodelPagination import PaginationParams, PaginatedResult
logger = logging.getLogger(__name__)

View file

@ -10,7 +10,7 @@ from typing import Dict, Any, Optional, List
from modules.connectors.connectorVoiceGoogle import ConnectorGoogleSpeech
from modules.datamodels.datamodelVoice import VoiceSettings
from modules.datamodels.datamodelUam import User
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
logger = logging.getLogger(__name__)

View file

@ -18,7 +18,7 @@ from modules.datamodels.datamodelUam import User, UserConnection, AuthAuthority,
from modules.datamodels.datamodelSecurity import Token
from modules.security.auth import getCurrentUser, limiter
from modules.interfaces.interfaceDbAppObjects import getInterface
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp, parseTimestamp
# Configure logger
logger = logging.getLogger(__name__)
@ -51,7 +51,7 @@ def getTokenStatusForConnection(interface, connectionId: str) -> tuple[str, Opti
latestCreatedAt = 0
for tokenData in tokens:
createdAt = tokenData.get("createdAt", 0)
createdAt = parseTimestamp(tokenData.get("createdAt"), default=0)
if createdAt > latestCreatedAt:
latestCreatedAt = createdAt
latestToken = tokenData
@ -60,7 +60,7 @@ def getTokenStatusForConnection(interface, connectionId: str) -> tuple[str, Opti
return "none", None
# Check if token is expired
expiresAt = latestToken.get("expiresAt")
expiresAt = parseTimestamp(latestToken.get("expiresAt"))
if not expiresAt:
return "none", None

View file

@ -126,13 +126,24 @@ async def get_user(
async def create_user(
request: Request,
user_data: User = Body(...),
password: Optional[str] = Body(None, embed=True),
currentUser: User = Depends(getCurrentUser)
) -> User:
"""Create a new user"""
appInterface = interfaceDbAppObjects.getInterface(currentUser)
# Create user
newUser = appInterface.createUser(user_data)
# Extract fields from User model and call createUser with individual parameters
from modules.datamodels.datamodelUam import AuthAuthority
newUser = appInterface.createUser(
username=user_data.username,
password=password,
email=user_data.email,
fullName=user_data.fullName,
language=user_data.language,
enabled=user_data.enabled,
privilege=user_data.privilege,
authenticationAuthority=user_data.authenticationAuthority
)
return newUser

View file

@ -15,7 +15,7 @@ from modules.interfaces.interfaceDbAppObjects import getInterface, getRootInterf
from modules.datamodels.datamodelUam import AuthAuthority, User, ConnectionStatus, UserConnection
from modules.security.auth import getCurrentUser, limiter
from modules.security.jwtService import createAccessToken, setAccessTokenCookie, createRefreshToken, setRefreshTokenCookie
from modules.shared.timezoneUtils import createExpirationTimestamp, getUtcTimestamp
from modules.shared.timeUtils import createExpirationTimestamp, getUtcTimestamp, parseTimestamp
# Configure logger
logger = logging.getLogger(__name__)
@ -264,7 +264,7 @@ async def auth_callback(code: str, state: str, request: Request, response: Respo
})
if existing_tokens:
# Use most recent by createdAt
existing_tokens.sort(key=lambda x: x.get("createdAt", 0), reverse=True)
existing_tokens.sort(key=lambda x: parseTimestamp(x.get("createdAt"), default=0), reverse=True)
token_response["refresh_token"] = existing_tokens[0].get("tokenRefresh", "")
if not token_response.get("refresh_token") and user_id:
existing_access_tokens = rootInterface.db.getRecordset(Token, recordFilter={
@ -273,7 +273,7 @@ async def auth_callback(code: str, state: str, request: Request, response: Respo
"authority": AuthAuthority.GOOGLE
})
if existing_access_tokens:
existing_access_tokens.sort(key=lambda x: x.get("createdAt", 0), reverse=True)
existing_access_tokens.sort(key=lambda x: parseTimestamp(x.get("createdAt"), default=0), reverse=True)
token_response["refresh_token"] = existing_access_tokens[0].get("tokenRefresh", "")
except Exception:
# Non-fatal; continue without refresh token
@ -748,19 +748,21 @@ async def refresh_token(
raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to refresh token")
# Update the connection status and timing
google_connection.expiresAt = float(current_token.expiresAt) if current_token.expiresAt else google_connection.expiresAt
expiresAtValue = parseTimestamp(current_token.expiresAt)
google_connection.expiresAt = expiresAtValue if expiresAtValue else google_connection.expiresAt
google_connection.lastChecked = getUtcTimestamp()
google_connection.status = ConnectionStatus.ACTIVE
appInterface.db.recordModify(UserConnection, google_connection.id, google_connection.model_dump())
# Calculate time until expiration
current_time = getUtcTimestamp()
expires_in = int(current_token.expiresAt - current_time) if current_token.expiresAt else 0
currentTime = getUtcTimestamp()
expiresAt = parseTimestamp(current_token.expiresAt)
expiresIn = int(expiresAt - currentTime) if expiresAt else 0
return {
"message": "Token refreshed successfully",
"expires_at": current_token.expiresAt,
"expires_in_seconds": expires_in
"expires_at": expiresAt,
"expires_in_seconds": expiresIn
}
except HTTPException:

View file

@ -16,7 +16,7 @@ from modules.datamodels.datamodelUam import AuthAuthority, User, ConnectionStatu
from modules.datamodels.datamodelSecurity import Token
from modules.security.auth import getCurrentUser, limiter
from modules.security.jwtService import createAccessToken, setAccessTokenCookie, createRefreshToken, setRefreshTokenCookie
from modules.shared.timezoneUtils import createExpirationTimestamp, getUtcTimestamp
from modules.shared.timeUtils import createExpirationTimestamp, getUtcTimestamp, parseTimestamp
# Configure logger
logger = logging.getLogger(__name__)
@ -639,7 +639,9 @@ async def refresh_token(
appInterface.saveConnectionToken(refreshedToken)
# Update the connection's expiration time
msft_connection.expiresAt = float(refreshedToken.expiresAt)
expiresAtValue = parseTimestamp(refreshedToken.expiresAt)
if expiresAtValue:
msft_connection.expiresAt = expiresAtValue
msft_connection.lastChecked = getUtcTimestamp()
msft_connection.status = ConnectionStatus.ACTIVE
@ -647,12 +649,13 @@ async def refresh_token(
appInterface.db.recordModify(UserConnection, msft_connection.id, msft_connection.model_dump())
# Calculate time until expiration
current_time = getUtcTimestamp()
expiresIn = int(refreshedToken.expiresAt - current_time)
currentTime = getUtcTimestamp()
expiresAt = parseTimestamp(refreshedToken.expiresAt)
expiresIn = int(expiresAt - currentTime) if expiresAt else 0
return {
"message": "Token refreshed successfully",
"expires_at": refreshedToken.expiresAt,
"expires_at": expiresAt,
"expires_in_seconds": expiresIn
}
else:

View file

@ -9,7 +9,7 @@ from fastapi import Response
from jose import jwt
from modules.shared.configuration import APP_CONFIG
from modules.shared.timezoneUtils import getUtcNow
from modules.shared.timeUtils import getUtcNow
# Config
SECRET_KEY = APP_CONFIG.get("APP_JWT_KEY_SECRET")

View file

@ -10,7 +10,7 @@ from typing import Optional, Dict, Any, Callable
from modules.datamodels.datamodelSecurity import Token
from modules.datamodels.datamodelUam import AuthAuthority
from modules.shared.configuration import APP_CONFIG
from modules.shared.timezoneUtils import getUtcTimestamp, createExpirationTimestamp
from modules.shared.timeUtils import getUtcTimestamp, createExpirationTimestamp, parseTimestamp
logger = logging.getLogger(__name__)
@ -167,7 +167,7 @@ class TokenManager:
# Only allow a new refresh if at least 10 minutes passed since the token was created/refreshed
try:
nowTs = getUtcTimestamp()
createdTs = float(oldToken.createdAt) if oldToken.createdAt is not None else 0.0
createdTs = parseTimestamp(oldToken.createdAt, default=0.0)
secondsSinceLastRefresh = nowTs - createdTs
if secondsSinceLastRefresh < 10 * 60:
logger.info(

View file

@ -11,7 +11,7 @@ from starlette.middleware.base import BaseHTTPMiddleware
from typing import Callable
import asyncio
from modules.security.tokenRefreshService import token_refresh_service
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
logger = logging.getLogger(__name__)

View file

@ -9,7 +9,7 @@ to ensure users don't experience token expiration issues.
import logging
from typing import Dict, Any
from modules.datamodels.datamodelUam import UserConnection, AuthAuthority
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
from modules.shared.auditLogger import audit_logger
logger = logging.getLogger(__name__)

View file

@ -7,7 +7,7 @@ import logging
from typing import Any, Optional, Dict, Callable, List
from modules.shared.configuration import APP_CONFIG
from modules.shared.eventManagement import eventManager
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
from modules.shared import jsonUtils
logger = logging.getLogger(__name__)

View file

@ -120,7 +120,7 @@ def debugLogToFile(message: str, context: str = "DEBUG") -> None:
debug_file = os.path.join(debug_dir, "debug_workflow.log")
# Format the debug entry
from modules.shared.timezoneUtils import getUtcTimestamp
from modules.shared.timeUtils import getUtcTimestamp
timestamp = getUtcTimestamp()
debug_entry = f"[{timestamp}] [{context}] {message}\n"

104
modules/shared/timeUtils.py Normal file
View file

@ -0,0 +1,104 @@
"""
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

View file

@ -1,37 +0,0 @@
"""
Timezone utilities for consistent timestamp handling across the gateway.
Ensures all timestamps are properly handled as UTC.
"""
from datetime import datetime, timezone
import time
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

View file

@ -14,6 +14,7 @@ from modules.datamodels.datamodelChat import ChatWorkflow
from modules.datamodels.datamodelAi import AiCallOptions, OperationTypeEnum, ProcessingModeEnum, PriorityEnum
from modules.workflows.processing.modes.modeBase import BaseMode
from modules.workflows.processing.shared.stateTools import checkWorkflowStopped
from modules.shared.timeUtils import parseTimestamp
from modules.workflows.processing.shared.executionState import TaskExecutionState
from modules.workflows.processing.shared.promptGenerationActionsActionplan import (
generateActionDefinitionPrompt,
@ -662,7 +663,7 @@ class ActionplanMode(BaseMode):
retryCount=createdAction.get("retryCount", 0),
retryMax=createdAction.get("retryMax", 3),
processingTime=createdAction.get("processingTime"),
timestamp=float(createdAction.get("timestamp", self.services.utils.timestampGetUtc())),
timestamp=parseTimestamp(createdAction.get("timestamp"), default=self.services.utils.timestampGetUtc()),
result=createdAction.get("result"),
resultDocuments=createdAction.get("resultDocuments", []),
userMessage=createdAction.get("userMessage")

View file

@ -12,6 +12,7 @@ from modules.datamodels.datamodelChat import (
from modules.datamodels.datamodelChat import ChatWorkflow
from modules.workflows.processing.modes.modeBase import BaseMode
from modules.workflows.processing.shared.stateTools import checkWorkflowStopped
from modules.shared.timeUtils import parseTimestamp
logger = logging.getLogger(__name__)
@ -342,7 +343,7 @@ class AutomationMode(BaseMode):
retryCount=createdAction.get("retryCount", 0),
retryMax=createdAction.get("retryMax", 3),
processingTime=createdAction.get("processingTime"),
timestamp=float(createdAction.get("timestamp", self.services.utils.timestampGetUtc())),
timestamp=parseTimestamp(createdAction.get("timestamp"), default=self.services.utils.timestampGetUtc()),
result=createdAction.get("result"),
userMessage=createdAction.get("userMessage")
)

View file

@ -14,6 +14,7 @@ from modules.datamodels.datamodelChat import (
from modules.datamodels.datamodelChat import ChatWorkflow
from modules.workflows.processing.modes.modeBase import BaseMode
from modules.workflows.processing.shared.stateTools import checkWorkflowStopped
from modules.shared.timeUtils import parseTimestamp
from modules.workflows.processing.shared.executionState import TaskExecutionState, shouldContinue
from modules.workflows.processing.shared.promptGenerationActionsDynamic import (
generateDynamicPlanSelectionPrompt,
@ -867,7 +868,7 @@ Return only the user-friendly message, no technical details."""
retryCount=createdAction.get("retryCount", 0),
retryMax=createdAction.get("retryMax", 3),
processingTime=createdAction.get("processingTime"),
timestamp=float(createdAction.get("timestamp", self.services.utils.timestampGetUtc())),
timestamp=parseTimestamp(createdAction.get("timestamp"), default=self.services.utils.timestampGetUtc()),
result=createdAction.get("result"),
resultDocuments=createdAction.get("resultDocuments", []),
userMessage=createdAction.get("userMessage")
@ -960,7 +961,7 @@ Return only the user-friendly message, no technical details."""
retryCount=createdAction.get("retryCount", 0),
retryMax=createdAction.get("retryMax", 3),
processingTime=createdAction.get("processingTime"),
timestamp=float(createdAction.get("timestamp", self.services.utils.timestampGetUtc())),
timestamp=parseTimestamp(createdAction.get("timestamp"), default=self.services.utils.timestampGetUtc()),
result=createdAction.get("result"),
resultDocuments=createdAction.get("resultDocuments", []),
userMessage=createdAction.get("userMessage")