17 KiB
Automation Templates – Gateway Implementation Concept
Basis: doc_automation_templates_db_and_editor_concept.md
⚠️ STATUS: BEREITS IMPLEMENTIERT
Das Backend ist vollständig implementiert! Folgende Komponenten existieren bereits:
| Komponente | Status | Datei |
|---|---|---|
| AutomationTemplate Model | ✅ Vorhanden | datamodelFeatureAutomation.py:49 |
| RBAC Namespace | ✅ Vorhanden | interfaceRbac.py:69 |
| API Routes (CRUD) | ✅ Vorhanden | routeFeatureAutomation.py:391-607 |
| Interface Methods | ✅ Vorhanden | interfaceFeatureAutomation.py:411-575 |
| Navigation Entry | ✅ Vorhanden | mainSystem.py:96-103 |
Was noch fehlt:
- Bootstrap-Funktion für initiale Template-Migration (optional)
Actions-Katalog: ✅ Bereits vorhanden unter GET /api/automations/actions (Zeile 293)
1. Bestehendes Datenmodell: AutomationTemplate
1.1 Datei: gateway/modules/features/automation/datamodelFeatureAutomation.py
Neues Modell AutomationTemplate hinzufügen (neben bestehendem AutomationDefinition):
from modules.datamodels.datamodelUtils import TextMultilingual
class AutomationTemplate(BaseModel):
"""Automation-Vorlage ohne scharfe Placeholder-Werte."""
id: str = Field(
default_factory=lambda: str(uuid.uuid4()),
description="Primary key",
json_schema_extra={"frontend_type": "text", "frontend_readonly": True}
)
label: TextMultilingual = Field(
description="Template name (multilingual)",
json_schema_extra={"frontend_type": "multilingual", "frontend_required": True}
)
overview: Optional[TextMultilingual] = Field(
None,
description="Short description (multilingual)",
json_schema_extra={"frontend_type": "multilingual", "frontend_required": False}
)
template: str = Field(
description="JSON workflow structure with {{KEY:...}} placeholders",
json_schema_extra={"frontend_type": "textarea", "frontend_required": True}
)
# System fields (_createdAt, _createdBy, etc.) werden automatisch vom DB-Connector gesetzt
registerModelLabels(
"AutomationTemplate",
{"en": "Automation Template", "de": "Automation-Vorlage", "fr": "Modèle d'automatisation"},
{
"id": {"en": "ID", "de": "ID", "fr": "ID"},
"label": {"en": "Label", "de": "Bezeichnung", "fr": "Libellé"},
"overview": {"en": "Overview", "de": "Übersicht", "fr": "Aperçu"},
"template": {"en": "Template", "de": "Vorlage", "fr": "Modèle"},
},
)
1.2 Namespace & Tabelle
- Namespace:
data.automation(wie AutomationDefinition) - Tabelle:
AutomationTemplate - Registrierung in
interfaceRbac.py(TABLE_NAMESPACE_MAP):
"AutomationTemplate": "automation",
2. RBAC für AutomationTemplate
2.1 Bootstrap-Regeln (interfaceBootstrap.py)
In _createTableSpecificRules() hinzufügen:
# AutomationTemplate: MY-level (user-owned), like AutomationDefinition
for roleId in [adminId, userId]:
if roleId:
tableRules.append(AccessRule(
roleId=roleId,
context=AccessRuleContext.DATA,
item="data.automation.AutomationTemplate",
view=True,
read=AccessLevel.MY,
create=AccessLevel.MY,
update=AccessLevel.MY,
delete=AccessLevel.MY,
))
if viewerId:
tableRules.append(AccessRule(
roleId=viewerId,
context=AccessRuleContext.DATA,
item="data.automation.AutomationTemplate",
view=True,
read=AccessLevel.MY,
create=AccessLevel.NONE,
update=AccessLevel.NONE,
delete=AccessLevel.NONE,
))
In _ensureDataContextRules() (Zeile ~845) ergänzen:
"data.automation.AutomationTemplate",
2.2 UI-Regeln
In mainAutomation.py bereits vorhanden:
{"objectKey": "ui.feature.automation.templates", ...}
Sichtbarkeit gemäss RBAC (alle mit Berechtigung; nicht nur SysAdmin).
2.3 Navigation (mainSystem.py)
In NAVIGATION_SECTIONS unter dem "workflows" Abschnitt hinzufügen:
{
"id": "automation-templates",
"objectKey": "ui.system.automation-templates",
"label": {"en": "Templates", "de": "Vorlagen", "fr": "Modèles"},
"icon": "FaFileAlt",
"path": "/workflows/automation-templates",
"order": 35, # Nach automations (30)
"public": True,
},
WICHTIG: Das objectKey Format für Navigation ist ui.system.xxx, nicht page.system.xxx!
3. API Routes für AutomationTemplate
3.1 Neue Route-Datei oder erweitern: routeFeatureAutomation.py
Endpoints unter /api/automation-templates (oder /api/automations/templates erweitern):
from modules.features.automation.datamodelFeatureAutomation import AutomationTemplate
# GET /api/automation-templates - Liste (RBAC-gefiltert)
@router.get("/templates", response_model=PaginatedResponse[AutomationTemplate])
async def get_templates(
request: Request,
pagination: Optional[str] = Query(None),
context: RequestContext = Depends(getRequestContext)
):
"""Get automation templates, filtered by RBAC (MY = own templates)."""
chatInterface = getChatInterface(context.user, ...)
result = chatInterface.getAllAutomationTemplates(pagination=paginationParams)
return JSONResponse(content=result)
# GET /api/automation-templates/{id}
@router.get("/templates/{templateId}", response_model=AutomationTemplate)
async def get_template(templateId: str, context: RequestContext = Depends(getRequestContext)):
chatInterface = getChatInterface(context.user, ...)
template = chatInterface.getAutomationTemplate(templateId)
if not template:
raise HTTPException(404, "Template not found")
return template
# POST /api/automation-templates
@router.post("/templates", response_model=AutomationTemplate)
async def create_template(
request: Request,
templateData: Dict[str, Any] = Body(...),
context: RequestContext = Depends(getRequestContext)
):
chatInterface = getChatInterface(context.user, ...)
return chatInterface.createAutomationTemplate(templateData)
# PUT /api/automation-templates/{id}
@router.put("/templates/{templateId}", response_model=AutomationTemplate)
async def update_template(
templateId: str,
templateData: Dict[str, Any] = Body(...),
context: RequestContext = Depends(getRequestContext)
):
chatInterface = getChatInterface(context.user, ...)
return chatInterface.updateAutomationTemplate(templateId, templateData)
# DELETE /api/automation-templates/{id}
@router.delete("/templates/{templateId}")
async def delete_template(templateId: str, context: RequestContext = Depends(getRequestContext)):
chatInterface = getChatInterface(context.user, ...)
success = chatInterface.deleteAutomationTemplate(templateId)
if not success:
raise HTTPException(404, "Template not found or no permission")
return {"success": True}
3.2 Interface-Methoden (interfaceDbChat.py)
Analog zu getAllAutomationDefinitions, createAutomationDefinition, etc.:
def getAllAutomationTemplates(self, pagination=None) -> Union[List[Dict], PaginatedResult]:
"""Returns templates filtered by RBAC (MY = own templates)."""
filteredTemplates = getRecordsetWithRBAC(self.db, AutomationTemplate, self.currentUser)
# ... pagination, enrichment
return filteredTemplates
def getAutomationTemplate(self, templateId: str) -> Optional[AutomationTemplate]:
filtered = getRecordsetWithRBAC(self.db, AutomationTemplate, self.currentUser, recordFilter={"id": templateId})
return AutomationTemplate(**filtered[0]) if filtered else None
def createAutomationTemplate(self, templateData: Dict) -> AutomationTemplate:
if not self.checkRbacPermission(AutomationTemplate, "create"):
raise ValueError("No permission to create template")
simpleFields, _ = self._separateObjectFields(AutomationTemplate, templateData)
created = self.db.recordCreate(AutomationTemplate, simpleFields)
return AutomationTemplate(**created)
def updateAutomationTemplate(self, templateId: str, templateData: Dict) -> AutomationTemplate:
existing = self.getAutomationTemplate(templateId)
if not existing:
raise ValueError("Template not found")
if not self.checkRbacPermission(AutomationTemplate, "update", templateId):
raise ValueError("No permission to update")
simpleFields, _ = self._separateObjectFields(AutomationTemplate, templateData)
updated = self.db.recordModify(AutomationTemplate, templateId, simpleFields)
return AutomationTemplate(**updated)
def deleteAutomationTemplate(self, templateId: str) -> bool:
existing = self.getAutomationTemplate(templateId)
if not existing:
return False
if not self.checkRbacPermission(AutomationTemplate, "delete", templateId):
raise ValueError("No permission to delete")
self.db.recordDelete(AutomationTemplate, templateId)
return True
4. Bootstrap: Template-Seed
4.1 Neue Funktion in interfaceBootstrap.py
def initAutomationTemplates(db: DatabaseConnector) -> None:
"""
Seed initial automation templates from subAutomationTemplates.py.
Only runs if no templates exist yet (bootstrap).
Creates templates with _createdBy = admin user (SysAdmin privilege).
"""
from modules.features.automation.subAutomationTemplates import AUTOMATION_TEMPLATES
from modules.features.automation.datamodelFeatureAutomation import AutomationTemplate
# Check if templates already exist
existing = db.getRecordset(AutomationTemplate)
if existing:
logger.info(f"Automation templates already seeded ({len(existing)} templates)")
return
# Get admin user ID for _createdBy
adminUsers = db.getRecordset(UserInDB, {"email": APP_CONFIG.ADMIN_EMAIL})
adminUserId = adminUsers[0]["id"] if adminUsers else None
templates = AUTOMATION_TEMPLATES.get("sets", [])
for i, templateSet in enumerate(templates):
templateContent = templateSet.get("template", {})
overview = templateContent.get("overview", f"Template {i+1}")
# Create multilingual label from overview (German as primary since current templates are German)
label = {"en": overview, "de": overview}
# Create template WITHOUT parameters (no sharp values)
templateData = {
"label": label,
"overview": {"en": overview, "de": overview},
"template": json.dumps(templateContent), # Only template JSON with {{KEY:...}}
}
# Set _createdBy to admin for bootstrap
if adminUserId:
templateData["_createdBy"] = adminUserId
db.recordCreate(AutomationTemplate, templateData)
logger.info(f"Created automation template: {overview}")
logger.info(f"Seeded {len(templates)} automation templates")
4.2 In initBootstrap() aufrufen
def initBootstrap(db: DatabaseConnector) -> None:
# ... existing code ...
# Seed automation templates (after admin user exists)
initAutomationTemplates(db)
logger.info("System bootstrap completed")
5. Actions-Katalog Endpoint
5.1 Neuer Endpoint: GET /api/automations/actions
from modules.workflows.processing.shared.methodDiscovery import discoverMethods, methods
@router.get("/actions")
async def get_available_actions(
request: Request,
context: RequestContext = Depends(getRequestContext)
):
"""
Get available workflow actions filtered by RBAC.
Returns action definitions with parameters and example JSON snippets.
"""
# Ensure methods are discovered
if not methods:
# Need a serviceCenter with current user for RBAC filtering
# This requires a lightweight serviceCenter or direct method iteration
pass
actionsList = []
for methodName, methodInfo in methods.items():
# Skip duplicate short names (e.g., "ai" and "AiMethod" are same)
if methodName != methodName.lower():
continue
methodInstance = methodInfo.get("instance")
if not methodInstance:
continue
for actionName, actionDef in methodInstance._actions.items():
actionId = actionDef.actionId
# RBAC check: user needs view permission on this action (RESOURCE context)
permissions = context.rbac.getUserPermissions(
user=context.user,
context=AccessRuleContext.RESOURCE,
item=actionId
)
if not permissions.view:
continue
# Build action info from WorkflowActionDefinition
actionInfo = {
"method": methodName,
"action": actionName,
"actionId": actionId,
"description": actionDef.description,
"category": actionDef.category,
"parameters": []
}
# Add parameters from WorkflowActionParameter
for paramName, paramDef in actionDef.parameters.items():
actionInfo["parameters"].append({
"name": paramName,
"type": paramDef.type,
"frontendType": paramDef.frontendType.value if paramDef.frontendType else "text",
"required": paramDef.required,
"default": paramDef.default,
"description": paramDef.description,
"frontendOptions": paramDef.frontendOptions,
})
# Build example JSON snippet for copy/paste
exampleParams = {}
for paramName, paramDef in actionDef.parameters.items():
if paramDef.required:
exampleParams[paramName] = f"{{{{KEY:{paramName}}}}}"
else:
exampleParams[paramName] = paramDef.default or f"{{{{KEY:{paramName}}}}}"
actionInfo["exampleJson"] = {
"execMethod": methodName,
"execAction": actionName,
"execParameters": exampleParams,
"execResultLabel": f"{methodName}_{actionName}_result"
}
actionsList.append(actionInfo)
return JSONResponse(content={"actions": actionsList})
6. Label in User-Sprache bei Definition-Erstellung
6.1 Bei "Aus Template erstellen" (Frontend ruft Backend)
Wenn das Frontend eine neue AutomationDefinition aus einem Template erstellt, sendet es:
template(JSON von AutomationTemplate.template)label(aus AutomationTemplate.label in User-Sprache extrahiert)
Backend-seitig (falls Backend den Label-Extract macht):
def createAutomationFromTemplate(self, templateId: str, userLanguage: str = "en") -> AutomationDefinition:
"""Create a new AutomationDefinition from a template, label in user's language."""
template = self.getAutomationTemplate(templateId)
if not template:
raise ValueError("Template not found")
# Extract label in user's language
labelMulti = template.label # TextMultilingual object
if hasattr(labelMulti, 'get_text'):
label = labelMulti.get_text(userLanguage)
elif isinstance(labelMulti, dict):
label = labelMulti.get(userLanguage) or labelMulti.get("en", "New Automation")
else:
label = str(labelMulti)
# Create definition with template content
definitionData = {
"label": label,
"template": template.template, # Copy template JSON
"placeholders": {}, # Empty - user fills in later
"schedule": "0 22 * * *", # Default schedule
"active": False,
}
return self.createAutomationDefinition(definitionData)
Alternative: Frontend extrahiert Label selbst und sendet direkt an createAutomationDefinition.
7. Zusammenfassung der Änderungen
| Datei | Änderung |
|---|---|
datamodelFeatureAutomation.py |
Neues Modell AutomationTemplate mit TextMultilingual für label/overview |
interfaceRbac.py |
TABLE_NAMESPACE_MAP erweitern: "AutomationTemplate": "automation" |
interfaceBootstrap.py |
RBAC-Regeln für AutomationTemplate (MY); Bootstrap-Seed initAutomationTemplates() |
interfaceDbChat.py |
CRUD-Methoden für AutomationTemplate (analog AutomationDefinition) |
routeFeatureAutomation.py |
Endpoints: GET/POST/PUT/DELETE /api/automation-templates/*, GET /api/automations/actions |
mainAutomation.py |
UI-Object ui.feature.automation.templates bereits vorhanden |
8. Abhängigkeiten & Reihenfolge
- Datamodel erstellen (AutomationTemplate)
- RBAC in interfaceRbac.py und interfaceBootstrap.py
- Interface-Methoden in interfaceDbChat.py
- Routes in routeFeatureAutomation.py
- Bootstrap-Seed in interfaceBootstrap.py (nach DB-Migration)
- Actions-Endpoint als letzte Erweiterung