From 6b2626244584e3b6ff5bdc4539bf61e22621994f Mon Sep 17 00:00:00 2001 From: ValueOn AG Date: Tue, 11 Nov 2025 23:08:45 +0100 Subject: [PATCH] feat: sysadmin and admin can manage automation events --- modules/interfaces/interfaceDbChatAccess.py | 18 +++- modules/interfaces/interfaceDbChatObjects.py | 96 ++++++++++++++++++-- modules/routes/routeDataAutomation.py | 36 +++++--- 3 files changed, 122 insertions(+), 28 deletions(-) diff --git a/modules/interfaces/interfaceDbChatAccess.py b/modules/interfaces/interfaceDbChatAccess.py index 4662484a..37e96d84 100644 --- a/modules/interfaces/interfaceDbChatAccess.py +++ b/modules/interfaces/interfaceDbChatAccess.py @@ -42,11 +42,19 @@ class ChatAccess: # Apply filtering based on privilege if table_name == "AutomationDefinition": - # Users see only their own automation definitions - filtered_records = [ - r for r in recordset - if r.get("mandateId","-") == self.mandateId and r.get("_createdBy") == self.userId - ] + # Filter automations based on user privilege + if userPrivilege == UserPrivilege.SYSADMIN: + # System admins see all automations + filtered_records = recordset + elif userPrivilege == UserPrivilege.ADMIN: + # Admins see all automations in their mandate + filtered_records = [r for r in recordset if r.get("mandateId","-") == self.mandateId] + else: + # Regular users see only their own automations + filtered_records = [ + r for r in recordset + if r.get("mandateId","-") == self.mandateId and r.get("_createdBy") == self.userId + ] elif userPrivilege == UserPrivilege.SYSADMIN: filtered_records = recordset # System admins see all records elif userPrivilege == UserPrivilege.ADMIN: diff --git a/modules/interfaces/interfaceDbChatObjects.py b/modules/interfaces/interfaceDbChatObjects.py index 8bb86fb1..db2c5d59 100644 --- a/modules/interfaces/interfaceDbChatObjects.py +++ b/modules/interfaces/interfaceDbChatObjects.py @@ -185,14 +185,26 @@ class ChatObjects: # First apply access control filteredRecords = self.access.uam(model_class, recordset) - # Then filter out database-specific fields - cleanedRecords = [] - for record in filteredRecords: - # Create a new dict with only non-database fields - cleanedRecord = {k: v for k, v in record.items() if not k.startswith('_')} - cleanedRecords.append(cleanedRecord) - - return cleanedRecords + # For AutomationDefinition, keep _createdBy and mandateId for enrichment purposes + # Other fields starting with _ are filtered out as they're database-specific + if model_class.__name__ == "AutomationDefinition": + # Keep _createdBy and mandateId for enrichment, filter out other _ fields + cleanedRecords = [] + for record in filteredRecords: + cleanedRecord = {} + for k, v in record.items(): + # Keep _createdBy and mandateId, filter out other _ fields + if k == "_createdBy" or k == "mandateId" or not k.startswith('_'): + cleanedRecord[k] = v + cleanedRecords.append(cleanedRecord) + return cleanedRecords + else: + # For other models, filter out all database-specific fields + cleanedRecords = [] + for record in filteredRecords: + cleanedRecord = {k: v for k, v in record.items() if not k.startswith('_')} + cleanedRecords.append(cleanedRecord) + return cleanedRecords def _canModify(self, model_class: type, recordId: Optional[str] = None) -> bool: """Delegate to access control module.""" @@ -1221,6 +1233,69 @@ class ChatObjects: eventId = automation.get("eventId") return "Running" if eventId else "Idle" + def _enrichAutomationsWithUserAndMandate(self, automations: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """ + Batch enrich automations with user names and mandate names for display. + Uses AppObjects interface to fetch users and mandates with proper access control. + """ + if not automations: + return automations + + from modules.interfaces.interfaceDbAppObjects import getInterface as getAppInterface + + # Collect all unique user IDs and mandate IDs + userIds = set() + mandateIds = set() + + for automation in automations: + createdBy = automation.get("_createdBy") + if createdBy: + userIds.add(createdBy) + + mandateId = automation.get("mandateId") + if mandateId: + mandateIds.add(mandateId) + + # Use AppObjects interface to fetch users (respects access control) + appInterface = getAppInterface(self.currentUser) + usersMap = {} + if userIds: + for user_id in userIds: + user = appInterface.getUser(user_id) + if user: + usersMap[user_id] = user.username or user.email or user_id + + # Use AppObjects interface to fetch mandates (respects access control) + mandatesMap = {} + if mandateIds: + for mandate_id in mandateIds: + mandate = appInterface.getMandate(mandate_id) + if mandate: + mandatesMap[mandate_id] = mandate.name or mandate_id + + # Enrich each automation with the fetched data + for automation in automations: + createdBy = automation.get("_createdBy") + if createdBy: + automation["_createdByUserName"] = usersMap.get(createdBy, createdBy) + else: + automation["_createdByUserName"] = "-" + + mandateId = automation.get("mandateId") + if mandateId: + automation["mandateName"] = mandatesMap.get(mandateId, mandateId) + else: + automation["mandateName"] = "-" + + return automations + + def _enrichAutomationWithUserAndMandate(self, automation: Dict[str, Any]) -> Dict[str, Any]: + """ + Enrich a single automation with user name and mandate name for display. + For multiple automations, use _enrichAutomationsWithUserAndMandate for better performance. + """ + return self._enrichAutomationsWithUserAndMandate([automation])[0] + def getAllAutomationDefinitions(self, pagination: Optional[PaginationParams] = None) -> Union[List[Dict[str, Any]], PaginatedResult]: """ Returns automation definitions based on user access level. @@ -1237,6 +1312,9 @@ class ChatObjects: if automation.get("executionLogs") is None: automation["executionLogs"] = [] + # Batch enrich with user and mandate names + self._enrichAutomationsWithUserAndMandate(filteredAutomations) + # If no pagination requested, return all items if pagination is None: return filteredAutomations @@ -1278,6 +1356,8 @@ class ChatObjects: # Ensure executionLogs is always a list, not None if automation.get("executionLogs") is None: automation["executionLogs"] = [] + # Enrich with user and mandate names + self._enrichAutomationWithUserAndMandate(automation) return automation except Exception as e: logger.error(f"Error getting automation definition: {str(e)}") diff --git a/modules/routes/routeDataAutomation.py b/modules/routes/routeDataAutomation.py index 7e268c0b..903d0d53 100644 --- a/modules/routes/routeDataAutomation.py +++ b/modules/routes/routeDataAutomation.py @@ -66,23 +66,29 @@ async def get_automations( # If pagination was requested, result is PaginatedResult # If no pagination, result is List[Dict] + # Note: Using JSONResponse to bypass Pydantic validation which would filter out _createdBy + # The enriched fields (_createdByUserName, mandateName) are not in the Pydantic model + from fastapi.responses import JSONResponse + if paginationParams: - return PaginatedResponse( - items=result.items, - pagination=PaginationMetadata( - currentPage=paginationParams.page, - pageSize=paginationParams.pageSize, - totalItems=result.totalItems, - totalPages=result.totalPages, - sort=paginationParams.sort, - filters=paginationParams.filters - ) - ) + response_data = { + "items": result.items, + "pagination": { + "currentPage": paginationParams.page, + "pageSize": paginationParams.pageSize, + "totalItems": result.totalItems, + "totalPages": result.totalPages, + "sort": paginationParams.sort, + "filters": paginationParams.filters + } + } else: - return PaginatedResponse( - items=result, - pagination=None - ) + response_data = { + "items": result, + "pagination": None + } + + return JSONResponse(content=response_data) except HTTPException: raise except Exception as e: