From 2b5287188deea700f2d8bdade4e624e5abc0713c Mon Sep 17 00:00:00 2001 From: ValueOn AG Date: Sat, 23 Aug 2025 01:16:20 +0200 Subject: [PATCH] refactored attributes / timestamps / data objects CRUD --- modules/interfaces/interfaceAppModel.py | 215 +++++++++++++++--- modules/interfaces/interfaceAppObjects.py | 42 +++- modules/interfaces/interfaceChatModel.py | 91 +++++++- modules/interfaces/interfaceChatObjects.py | 2 +- modules/interfaces/interfaceComponentModel.py | 80 ++++++- modules/routes/routeSecurityGoogle.py | 2 +- modules/routes/routeSecurityMsft.py | 30 ++- modules/shared/attributeUtils.py | 124 ++++++---- notes/readme.md | 1 + 9 files changed, 479 insertions(+), 108 deletions(-) diff --git a/modules/interfaces/interfaceAppModel.py b/modules/interfaces/interfaceAppModel.py index be026a27..1a847e19 100644 --- a/modules/interfaces/interfaceAppModel.py +++ b/modules/interfaces/interfaceAppModel.py @@ -31,10 +31,39 @@ class ConnectionStatus(str, Enum): class Mandate(BaseModel, ModelMixin): """Data model for a mandate""" - id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Unique ID of the mandate") - name: str = Field(description="Name of the mandate") - language: str = Field(default="en", description="Default language of the mandate") - enabled: bool = Field(default=True, description="Indicates whether the mandate is enabled") + id: str = Field( + default_factory=lambda: str(uuid.uuid4()), + description="Unique ID of the mandate", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + name: str = Field( + description="Name of the mandate", + frontend_type="text", + frontend_readonly=False, + frontend_required=True + ) + language: str = Field( + default="en", + description="Default language of the mandate", + frontend_type="select", + frontend_readonly=False, + frontend_required=True, + frontend_options=[ + {"value": "de", "label": {"en": "Deutsch", "fr": "Allemand"}}, + {"value": "en", "label": {"en": "English", "fr": "Anglais"}}, + {"value": "fr", "label": {"en": "Français", "fr": "Français"}}, + {"value": "it", "label": {"en": "Italiano", "fr": "Italien"}} + ] + ) + enabled: bool = Field( + default=True, + description="Indicates whether the mandate is enabled", + frontend_type="checkbox", + frontend_readonly=False, + frontend_required=False + ) # Register labels for Mandate register_model_labels( @@ -50,20 +79,83 @@ register_model_labels( class UserConnection(BaseModel, ModelMixin): """Data model for a user's connection to an external service""" - id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Unique ID of the connection") - userId: str = Field(description="ID of the user this connection belongs to") - authority: AuthAuthority = Field(description="Authentication authority") - externalId: str = Field(description="User ID in the external system") - externalUsername: str = Field(description="Username in the external system") - externalEmail: Optional[EmailStr] = Field(None, description="Email in the external system") - status: ConnectionStatus = Field(default=ConnectionStatus.ACTIVE, description="Connection status") - connectedAt: float = Field(default_factory=get_utc_timestamp, description="When the connection was established (UTC timestamp in seconds)") - lastChecked: float = Field(default_factory=get_utc_timestamp, description="When the connection was last verified (UTC timestamp in seconds)") - expiresAt: Optional[float] = Field(None, description="When the connection expires (UTC timestamp in seconds)") - - def to_dict(self) -> Dict[str, Any]: - """Convert the model to a dictionary""" - return super().to_dict() + id: str = Field( + default_factory=lambda: str(uuid.uuid4()), + description="Unique ID of the connection", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + userId: str = Field( + description="ID of the user this connection belongs to", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + authority: AuthAuthority = Field( + description="Authentication authority", + frontend_type="select", + frontend_readonly=True, + frontend_required=False, + frontend_options=[ + {"value": "local", "label": {"en": "Local", "fr": "Local"}}, + {"value": "google", "label": {"en": "Google", "fr": "Google"}}, + {"value": "msft", "label": {"en": "Microsoft", "fr": "Microsoft"}} + ] + ) + externalId: str = Field( + description="User ID in the external system", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + externalUsername: str = Field( + description="Username in the external system", + frontend_type="text", + frontend_readonly=False, + frontend_required=False + ) + externalEmail: Optional[EmailStr] = Field( + None, + description="Email in the external system", + frontend_type="email", + frontend_readonly=False, + frontend_required=False + ) + status: ConnectionStatus = Field( + default=ConnectionStatus.ACTIVE, + description="Connection status", + frontend_type="select", + frontend_readonly=False, + frontend_required=False, + frontend_options=[ + {"value": "active", "label": {"en": "Active", "fr": "Actif"}}, + {"value": "inactive", "label": {"en": "Inactive", "fr": "Inactif"}}, + {"value": "expired", "label": {"en": "Expired", "fr": "Expiré"}}, + {"value": "pending", "label": {"en": "Pending", "fr": "En attente"}} + ] + ) + connectedAt: float = Field( + default_factory=get_utc_timestamp, + description="When the connection was established (UTC timestamp in seconds)", + frontend_type="timestamp", + frontend_readonly=True, + frontend_required=False + ) + lastChecked: float = Field( + default_factory=get_utc_timestamp, + description="When the connection was last verified (UTC timestamp in seconds)", + frontend_type="timestamp", + frontend_readonly=True, + frontend_required=False + ) + expiresAt: Optional[float] = Field( + None, + description="When the connection expires (UTC timestamp in seconds)", + frontend_type="timestamp", + frontend_readonly=True, + frontend_required=False + ) # Register labels for UserConnection register_model_labels( @@ -135,15 +227,84 @@ register_model_labels( class User(BaseModel, ModelMixin): """Data model for a user""" - id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Unique ID of the user") - username: str = Field(description="Username for login") - email: Optional[EmailStr] = Field(None, description="Email address of the user") - fullName: Optional[str] = Field(None, description="Full name of the user") - language: str = Field(default="en", description="Preferred language of the user") - enabled: bool = Field(default=True, description="Indicates whether the user is enabled") - privilege: UserPrivilege = Field(default=UserPrivilege.USER, description="Permission level") - authenticationAuthority: AuthAuthority = Field(default=AuthAuthority.LOCAL, description="Primary authentication authority") - mandateId: Optional[str] = Field(None, description="ID of the mandate this user belongs to") + id: str = Field( + default_factory=lambda: str(uuid.uuid4()), + description="Unique ID of the user", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + username: str = Field( + description="Username for login", + frontend_type="text", + frontend_readonly=False, + frontend_required=True + ) + email: Optional[EmailStr] = Field( + None, + description="Email address of the user", + frontend_type="email", + frontend_readonly=False, + frontend_required=True + ) + fullName: Optional[str] = Field( + None, + description="Full name of the user", + frontend_type="text", + frontend_readonly=False, + frontend_required=False + ) + language: str = Field( + default="en", + description="Preferred language of the user", + frontend_type="select", + frontend_readonly=False, + frontend_required=True, + frontend_options=[ + {"value": "de", "label": {"en": "Deutsch", "fr": "Allemand"}}, + {"value": "en", "label": {"en": "English", "fr": "Anglais"}}, + {"value": "fr", "label": {"en": "Français", "fr": "Français"}}, + {"value": "it", "label": {"en": "Italiano", "fr": "Italien"}} + ] + ) + enabled: bool = Field( + default=True, + description="Indicates whether the user is enabled", + frontend_type="checkbox", + frontend_readonly=False, + frontend_required=False + ) + privilege: UserPrivilege = Field( + default=UserPrivilege.USER, + description="Permission level", + frontend_type="select", + frontend_readonly=False, + frontend_required=True, + frontend_options=[ + {"value": "user", "label": {"en": "User", "fr": "Utilisateur"}}, + {"value": "admin", "label": {"en": "Admin", "fr": "Administrateur"}}, + {"value": "sysadmin", "label": {"en": "SysAdmin", "fr": "Administrateur système"}} + ] + ) + authenticationAuthority: AuthAuthority = Field( + default=AuthAuthority.LOCAL, + description="Primary authentication authority", + frontend_type="select", + frontend_readonly=True, + frontend_required=False, + frontend_options=[ + {"value": "local", "label": {"en": "Local", "fr": "Local"}}, + {"value": "google", "label": {"en": "Google", "fr": "Google"}}, + {"value": "msft", "label": {"en": "Microsoft", "fr": "Microsoft"}} + ] + ) + mandateId: Optional[str] = Field( + None, + description="ID of the mandate this user belongs to", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) # Register labels for User register_model_labels( diff --git a/modules/interfaces/interfaceAppObjects.py b/modules/interfaces/interfaceAppObjects.py index 451335c4..a7cc91ed 100644 --- a/modules/interfaces/interfaceAppObjects.py +++ b/modules/interfaces/interfaceAppObjects.py @@ -284,15 +284,6 @@ class AppObjects: result = [] for conn_dict in connections: try: - # Convert string dates to datetime objects - for field in ['connectedAt', 'lastChecked', 'expiresAt']: - if field in conn_dict and conn_dict[field]: - try: - if isinstance(conn_dict[field], str): - conn_dict[field] = datetime.fromisoformat(conn_dict[field].replace('Z', '+00:00')) - except (ValueError, TypeError): - conn_dict[field] = None - # Create UserConnection object connection = UserConnection( id=conn_dict["id"], @@ -831,8 +822,8 @@ class AppObjects: logger.warning(f"No token found for connection: {connectionId}") return None - # Sort by creation date and get the latest - tokens.sort(key=lambda x: x.get("createdAt", ""), reverse=True) + # Sort by expiration date and get the latest (most recent expiration) + tokens.sort(key=lambda x: x.get("expiresAt", 0), reverse=True) latest_token = Token(**tokens[0]) # Check if token is expired @@ -905,6 +896,35 @@ class AppObjects: logger.error(f"Error deleting token for connection {connectionId}: {str(e)}") raise + def cleanupExpiredTokens(self) -> int: + """Clean up expired tokens for all connections, returns count of cleaned tokens""" + try: + from modules.shared.timezoneUtils import get_utc_timestamp + + current_time = get_utc_timestamp() + cleaned_count = 0 + + # Get all tokens + all_tokens = self.db.getRecordset("tokens", recordFilter={}) + + for token_data in all_tokens: + if token_data.get("expiresAt") and token_data.get("expiresAt") < current_time: + # Token is expired, delete it + self.db.recordDelete("tokens", token_data["id"]) + cleaned_count += 1 + logger.debug(f"Cleaned up expired token {token_data['id']} for connection {token_data.get('connectionId')}") + + # Clear cache to ensure fresh data + if cleaned_count > 0: + self._clearTableCache("tokens") + logger.info(f"Cleaned up {cleaned_count} expired tokens") + + return cleaned_count + + except Exception as e: + logger.error(f"Error cleaning up expired tokens: {str(e)}") + return 0 + def logout(self) -> None: """Logout current user - clear user context and tokens""" try: diff --git a/modules/interfaces/interfaceChatModel.py b/modules/interfaces/interfaceChatModel.py index 00966bc4..1e2363a5 100644 --- a/modules/interfaces/interfaceChatModel.py +++ b/modules/interfaces/interfaceChatModel.py @@ -465,17 +465,86 @@ register_model_labels( class ChatWorkflow(BaseModel, ModelMixin): """Data model for a chat workflow""" - id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Primary key") - mandateId: str = Field(description="ID of the mandate this workflow belongs to") - status: str = Field(description="Current status of the workflow") - name: Optional[str] = Field(None, description="Name of the workflow") - currentRound: int = Field(description="Current round number") - lastActivity: float = Field(default_factory=get_utc_timestamp, description="Timestamp of last activity (UTC timestamp in seconds)") - startedAt: float = Field(default_factory=get_utc_timestamp, description="When the workflow started (UTC timestamp in seconds)") - logs: List[ChatLog] = Field(default_factory=list, description="Workflow logs") - messages: List[ChatMessage] = Field(default_factory=list, description="Messages in the workflow") - stats: Optional[ChatStat] = Field(None, description="Workflow statistics") - tasks: List[TaskItem] = Field(default_factory=list, description="List of tasks in the workflow") + id: str = Field( + default_factory=lambda: str(uuid.uuid4()), + description="Primary key", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + mandateId: str = Field( + description="ID of the mandate this workflow belongs to", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + status: str = Field( + description="Current status of the workflow", + frontend_type="select", + frontend_readonly=False, + frontend_required=False, + frontend_options=[ + {"value": "running", "label": {"en": "Running", "fr": "En cours"}}, + {"value": "completed", "label": {"en": "Completed", "fr": "Terminé"}}, + {"value": "stopped", "label": {"en": "Stopped", "fr": "Arrêté"}}, + {"value": "error", "label": {"en": "Error", "fr": "Erreur"}} + ] + ) + name: Optional[str] = Field( + None, + description="Name of the workflow", + frontend_type="text", + frontend_readonly=False, + frontend_required=True + ) + currentRound: int = Field( + description="Current round number", + frontend_type="integer", + frontend_readonly=True, + frontend_required=False + ) + lastActivity: float = Field( + default_factory=get_utc_timestamp, + description="Timestamp of last activity (UTC timestamp in seconds)", + frontend_type="timestamp", + frontend_readonly=True, + frontend_required=False + ) + startedAt: float = Field( + default_factory=get_utc_timestamp, + description="When the workflow started (UTC timestamp in seconds)", + frontend_type="timestamp", + frontend_readonly=True, + frontend_required=False + ) + logs: List[ChatLog] = Field( + default_factory=list, + description="Workflow logs", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + messages: List[ChatMessage] = Field( + default_factory=list, + description="Messages in the workflow", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + stats: Optional[ChatStat] = Field( + None, + description="Workflow statistics", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + tasks: List[TaskItem] = Field( + default_factory=list, + description="List of tasks in the workflow", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) # Register labels for ChatWorkflow register_model_labels( diff --git a/modules/interfaces/interfaceChatObjects.py b/modules/interfaces/interfaceChatObjects.py index 847c150f..f5f27d3e 100644 --- a/modules/interfaces/interfaceChatObjects.py +++ b/modules/interfaces/interfaceChatObjects.py @@ -1335,7 +1335,7 @@ class ChatObjects: retryCount=createdAction.get("retryCount", 0), retryMax=createdAction.get("retryMax", 3), processingTime=createdAction.get("processingTime"), - timestamp=datetime.fromtimestamp(float(createdAction.get("timestamp", get_utc_timestamp()))), + timestamp=float(createdAction.get("timestamp", get_utc_timestamp())), result=createdAction.get("result"), resultDocuments=createdAction.get("resultDocuments", []) ) diff --git a/modules/interfaces/interfaceComponentModel.py b/modules/interfaces/interfaceComponentModel.py index b5b0b55d..3cc19dd8 100644 --- a/modules/interfaces/interfaceComponentModel.py +++ b/modules/interfaces/interfaceComponentModel.py @@ -16,13 +16,50 @@ from modules.shared.timezoneUtils import get_utc_timestamp class FileItem(BaseModel, ModelMixin): """Data model for a file item""" - id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Primary key") - mandateId: str = Field(description="ID of the mandate this file belongs to") - filename: str = Field(description="Name of the file") - mimeType: str = Field(description="MIME type of the file") - fileHash: str = Field(description="Hash of the file") - fileSize: int = Field(description="Size of the file in bytes") - creationDate: float = Field(default_factory=get_utc_timestamp, description="Date when the file was created (UTC timestamp in seconds)") + id: str = Field( + default_factory=lambda: str(uuid.uuid4()), + description="Primary key", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + mandateId: str = Field( + description="ID of the mandate this file belongs to", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + filename: str = Field( + description="Name of the file", + frontend_type="text", + frontend_readonly=False, + frontend_required=True + ) + mimeType: str = Field( + description="MIME type of the file", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + fileHash: str = Field( + description="Hash of the file", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + fileSize: int = Field( + description="Size of the file in bytes", + frontend_type="integer", + frontend_readonly=True, + frontend_required=False + ) + creationDate: float = Field( + default_factory=get_utc_timestamp, + description="Date when the file was created (UTC timestamp in seconds)", + frontend_type="timestamp", + frontend_readonly=True, + frontend_required=False + ) def to_dict(self) -> Dict[str, Any]: """Convert model to dictionary""" @@ -94,10 +131,31 @@ register_model_labels( class Prompt(BaseModel, ModelMixin): """Data model for a prompt""" - id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Primary key") - mandateId: str = Field(description="ID of the mandate this prompt belongs to") - content: str = Field(description="Content of the prompt") - name: str = Field(description="Name of the prompt") + id: str = Field( + default_factory=lambda: str(uuid.uuid4()), + description="Primary key", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + mandateId: str = Field( + description="ID of the mandate this prompt belongs to", + frontend_type="text", + frontend_readonly=True, + frontend_required=False + ) + content: str = Field( + description="Content of the prompt", + frontend_type="textarea", + frontend_readonly=False, + frontend_required=True + ) + name: str = Field( + description="Name of the prompt", + frontend_type="text", + frontend_readonly=False, + frontend_required=True + ) # Register labels for Prompt register_model_labels( diff --git a/modules/routes/routeSecurityGoogle.py b/modules/routes/routeSecurityGoogle.py index 1d7d47b7..bfa80f4e 100644 --- a/modules/routes/routeSecurityGoogle.py +++ b/modules/routes/routeSecurityGoogle.py @@ -512,7 +512,7 @@ async def refresh_token( appInterface.deleteTokenByConnectionId(google_connection.id) # Update the connection's expiration time - google_connection.expiresAt = datetime.fromtimestamp(refreshed_token.expiresAt) + google_connection.expiresAt = float(refreshed_token.expiresAt) google_connection.lastChecked = get_utc_timestamp() google_connection.status = ConnectionStatus.ACTIVE diff --git a/modules/routes/routeSecurityMsft.py b/modules/routes/routeSecurityMsft.py index 8737b9f7..90a957c9 100644 --- a/modules/routes/routeSecurityMsft.py +++ b/modules/routes/routeSecurityMsft.py @@ -443,6 +443,31 @@ async def logout( detail=f"Failed to logout: {str(e)}" ) +@router.post("/cleanup") +@limiter.limit("5/minute") +async def cleanup_expired_tokens( + request: Request, + currentUser: User = Depends(getCurrentUser) +) -> Dict[str, Any]: + """Clean up expired tokens for the current user""" + try: + appInterface = getInterface(currentUser) + + # Clean up expired tokens + cleaned_count = appInterface.cleanupExpiredTokens() + + return { + "message": f"Cleanup completed successfully", + "tokens_cleaned": cleaned_count + } + + except Exception as e: + logger.error(f"Error cleaning up expired tokens: {str(e)}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to cleanup expired tokens: {str(e)}" + ) + @router.post("/refresh") @limiter.limit("10/minute") async def refresh_token( @@ -473,7 +498,8 @@ async def refresh_token( logger.debug(f"Found Microsoft connection: {msft_connection.id}, status={msft_connection.status}") # Get the token for this specific connection using the new method - current_token = appInterface.getTokenForConnection(msft_connection.id, auto_refresh=False) + # Enable auto-refresh to handle expired tokens gracefully + current_token = appInterface.getTokenForConnection(msft_connection.id, auto_refresh=True) if not current_token: raise HTTPException( @@ -494,7 +520,7 @@ async def refresh_token( appInterface.deleteTokenByConnectionId(msft_connection.id) # Update the connection's expiration time - msft_connection.expiresAt = datetime.fromtimestamp(refreshed_token.expiresAt) + msft_connection.expiresAt = float(refreshed_token.expiresAt) msft_connection.lastChecked = get_utc_timestamp() msft_connection.status = ConnectionStatus.ACTIVE diff --git a/modules/shared/attributeUtils.py b/modules/shared/attributeUtils.py index a4624109..1b690383 100644 --- a/modules/shared/attributeUtils.py +++ b/modules/shared/attributeUtils.py @@ -16,7 +16,7 @@ class ModelMixin: """ Convert a Pydantic model to a dictionary. Handles both Pydantic v1 and v2. - Properly serializes datetime fields to ISO format strings. + All timestamp fields remain as float values. Returns: Dict[str, Any]: Dictionary representation of the model @@ -27,42 +27,10 @@ class ModelMixin: else: data: Dict[str, Any] = self.dict() # Pydantic v1 - # Convert datetime fields to ISO format strings - for key, value in data.items(): - if isinstance(value, datetime): - data[key] = value.isoformat() - elif isinstance(value, (int, float)) and self._is_timestamp_field(key): - # Handle timestamp fields based on field metadata - try: - data[key] = datetime.fromtimestamp(value).isoformat() - except (ValueError, TypeError): - # If conversion fails, keep the original value - pass - - return data - - def _is_timestamp_field(self, field_name: str) -> bool: - """ - Check if a field is a timestamp field based on field metadata. - Looks for 'UTC timestamp' in the field description. - """ - try: - # Get field info from Pydantic model - if hasattr(self, 'model_fields'): - # Pydantic v2 - field_info = self.model_fields.get(field_name) - if field_info and field_info.description: - return 'UTC timestamp' in field_info.description - elif hasattr(self, '__fields__'): - # Pydantic v1 - field_info = self.__fields__.get(field_name) - if field_info and field_info.field_info and field_info.field_info.description: - return 'UTC timestamp' in field_info.field_info.description - except Exception: - pass + # All fields (including timestamps) remain in their original format + # No conversions needed - timestamps are already float - # Fallback: return False for safety - return False + return data @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'ModelMixin': @@ -89,6 +57,12 @@ class AttributeDefinition(BaseModel, ModelMixin): options: Optional[List[Any]] = None validation: Optional[Dict[str, Any]] = None ui: Optional[Dict[str, Any]] = None + # New frontend metadata fields + readonly: bool = False + editable: bool = True + visible: bool = True + order: int = 0 + placeholder: Optional[str] = None # Global registry for model labels MODEL_LABELS: Dict[str, Dict[str, Dict[str, str]]] = {} @@ -194,30 +168,92 @@ def getModelAttributeDefinitions(modelClass: Type[BaseModel] = None, userLanguag if hasattr(modelClass, 'model_fields'): # Pydantic v2 fields = modelClass.model_fields for name, field in fields.items(): + # Extract frontend metadata from field info + field_info = field.field_info if hasattr(field, 'field_info') else None + # Check both direct attributes and extra field for frontend metadata + frontend_type = None + frontend_readonly = False + frontend_required = field.is_required() + frontend_options = None + + if field_info: + # Try direct attributes first + frontend_type = getattr(field_info, 'frontend_type', None) + frontend_readonly = getattr(field_info, 'frontend_readonly', False) + frontend_required = getattr(field_info, 'frontend_required', frontend_required) + frontend_options = getattr(field_info, 'frontend_options', None) + + # If not found, check extra field + if hasattr(field_info, 'extra') and field_info.extra: + if frontend_type is None: + frontend_type = field_info.extra.get('frontend_type') + if not frontend_readonly: + frontend_readonly = field_info.extra.get('frontend_readonly', False) + if frontend_required == field.is_required(): # Only override if we didn't get it from direct attribute + frontend_required = field_info.extra.get('frontend_required', frontend_required) + if frontend_options is None: + frontend_options = field_info.extra.get('frontend_options') + + # Use frontend type if available, otherwise fall back to Python type + field_type = frontend_type if frontend_type else (field.annotation.__name__ if hasattr(field.annotation, "__name__") else str(field.annotation)) + attributes.append({ "name": name, - "type": field.annotation.__name__ if hasattr(field.annotation, "__name__") else str(field.annotation), - "required": field.is_required() if hasattr(field, "is_required") else True, + "type": field_type, + "required": frontend_required, "description": field.description if hasattr(field, "description") else "", "label": labels.get(name, name), "placeholder": f"Please enter {labels.get(name, name)}", - "editable": True, + "editable": not frontend_readonly, "visible": True, - "order": len(attributes) + "order": len(attributes), + "readonly": frontend_readonly, + "options": frontend_options }) else: # Pydantic v1 fields = modelClass.__fields__ for name, field in fields.items(): + # Extract frontend metadata from field info + field_info = field.field_info if hasattr(field, 'field_info') else None + # Check both direct attributes and extra field for frontend metadata + frontend_type = None + frontend_readonly = False + frontend_required = field.required + frontend_options = None + + if field_info: + # Try direct attributes first + frontend_type = getattr(field_info, 'frontend_type', None) + frontend_readonly = getattr(field_info, 'frontend_readonly', False) + frontend_required = getattr(field_info, 'frontend_required', frontend_required) + frontend_options = getattr(field_info, 'frontend_options', None) + + # If not found, check extra field + if hasattr(field_info, 'extra') and field_info.extra: + if frontend_type is None: + frontend_type = field_info.extra.get('frontend_type') + if not frontend_readonly: + frontend_readonly = field_info.extra.get('frontend_readonly', False) + if frontend_required == field.required: # Only override if we didn't get it from direct attribute + frontend_required = field_info.extra.get('frontend_required', frontend_required) + if frontend_options is None: + frontend_options = field_info.extra.get('frontend_options') + + # Use frontend type if available, otherwise fall back to Python type + field_type = frontend_type if frontend_type else (field.type_.__name__ if hasattr(field.type_, "__name__") else str(field.type_)) + attributes.append({ "name": name, - "type": field.type_.__name__ if hasattr(field.type_, "__name__") else str(field.type_), - "required": field.required, + "type": field_type, + "required": frontend_required, "description": field.field_info.description if hasattr(field.field_info, "description") else "", "label": labels.get(name, name), "placeholder": f"Please enter {labels.get(name, name)}", - "editable": True, + "editable": not frontend_readonly, "visible": True, - "order": len(attributes) + "order": len(attributes), + "readonly": frontend_readonly, + "options": frontend_options }) return { diff --git a/notes/readme.md b/notes/readme.md index 47f21832..14233338 100644 --- a/notes/readme.md +++ b/notes/readme.md @@ -92,6 +92,7 @@ git remote set-url origin https://valueon@github.com/valueonag/gateway git remote set-url origin https://valueon@github.com/valueonag/frontend_agents git remote set-url origin https://valueon@github.com/valueonag/wiki git remote set-url origin https://valueon@github.com/valueonag/customer-svbe +git remote set-url origin https://valueon@github.com/valueonag/customer-althaus ### git delete workflow runs (cleanup)