PowerOn/gateway
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #67 from valueonag/feat/automation-center

Browse source 
feat: sysadmin and admin can manage automation events
This commit is contained in:
ValueOn AG 2025-11-11 23:17:17 +01:00 committed by GitHub
commit b603399690
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 122 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
modules/interfaces/interfaceDbChatAccess.py
View file

@ -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:

96
modules/interfaces/interfaceDbChatObjects.py
View file

@ -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)}")

36
modules/routes/routeDataAutomation.py
View file

@ -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: