422 lines
15 KiB
Python
422 lines
15 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
System-level Workflow Dashboard API.
|
|
|
|
Provides cross-feature, cross-mandate access to workflow runs AND workflows
|
|
with RBAC scoping: user sees own runs/workflows, mandate admin sees mandate
|
|
runs/workflows, sysadmin sees all.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import math
|
|
from typing import Optional, List
|
|
from fastapi import APIRouter, Depends, Request, Query, Path, HTTPException
|
|
from slowapi import Limiter
|
|
from slowapi.util import get_remote_address
|
|
|
|
from modules.auth.authentication import getRequestContext, RequestContext
|
|
from modules.interfaces.interfaceDbApp import getRootInterface
|
|
from modules.connectors.connectorDbPostgre import DatabaseConnector
|
|
from modules.shared.configuration import APP_CONFIG
|
|
from modules.datamodels.datamodelPagination import PaginationParams
|
|
from modules.datamodels.datamodelFeatures import FeatureInstance
|
|
from modules.datamodels.datamodelUam import Mandate
|
|
from modules.features.graphicalEditor.datamodelFeatureGraphicalEditor import (
|
|
AutoRun, AutoStepLog, AutoWorkflow, AutoTask,
|
|
)
|
|
from modules.shared.i18nRegistry import apiRouteContext
|
|
|
|
routeApiMsg = apiRouteContext("routeWorkflowDashboard")
|
|
|
|
logger = logging.getLogger(__name__)
|
|
limiter = Limiter(key_func=get_remote_address)
|
|
|
|
router = APIRouter(prefix="/api/system/workflow-runs", tags=["WorkflowDashboard"])
|
|
|
|
_GREENFIELD_DB = "poweron_graphicaleditor"
|
|
|
|
|
|
def _getDb() -> DatabaseConnector:
|
|
return DatabaseConnector(
|
|
dbHost=APP_CONFIG.get("DB_HOST", "localhost"),
|
|
dbDatabase=_GREENFIELD_DB,
|
|
dbUser=APP_CONFIG.get("DB_USER"),
|
|
dbPassword=APP_CONFIG.get("DB_PASSWORD_SECRET") or APP_CONFIG.get("DB_PASSWORD"),
|
|
dbPort=int(APP_CONFIG.get("DB_PORT", 5432)),
|
|
userId=None,
|
|
)
|
|
|
|
|
|
def _getUserMandateIds(userId: str) -> list[str]:
|
|
"""Get mandate IDs the user is a member of."""
|
|
rootIface = getRootInterface()
|
|
memberships = rootIface.getUserMandates(userId)
|
|
return [um.mandateId for um in memberships if um.mandateId and um.enabled]
|
|
|
|
|
|
def _getAdminMandateIds(userId: str, mandateIds: list) -> list:
|
|
"""Batch-check which mandates the user is admin for (2 SQL queries total)."""
|
|
if not mandateIds:
|
|
return []
|
|
rootIface = getRootInterface()
|
|
from modules.datamodels.datamodelMembership import UserMandateRole
|
|
allRoles = rootIface.db.getRecordset(UserMandateRole, recordFilter={
|
|
"userId": userId, "mandateId": mandateIds,
|
|
})
|
|
if not allRoles:
|
|
return []
|
|
|
|
roleIds = set()
|
|
roleToMandate: dict = {}
|
|
for r in allRoles:
|
|
row = r if isinstance(r, dict) else r.__dict__
|
|
rid = row.get("roleId")
|
|
mid = row.get("mandateId")
|
|
if rid:
|
|
roleIds.add(rid)
|
|
roleToMandate.setdefault(rid, set()).add(mid)
|
|
|
|
if not roleIds:
|
|
return []
|
|
|
|
from modules.datamodels.datamodelRbac import MandateRole
|
|
roleRecords = rootIface.db.getRecordset(MandateRole, recordFilter={"id": list(roleIds)})
|
|
adminMandates: set = set()
|
|
for role in (roleRecords or []):
|
|
row = role if isinstance(role, dict) else role.__dict__
|
|
if row.get("isAdmin"):
|
|
rid = row.get("id")
|
|
if rid and rid in roleToMandate:
|
|
adminMandates.update(roleToMandate[rid])
|
|
|
|
return [mid for mid in mandateIds if mid in adminMandates]
|
|
|
|
|
|
def _isUserMandateAdmin(userId: str, mandateId: str) -> bool:
|
|
"""Check if user is admin for a specific mandate."""
|
|
adminIds = _getAdminMandateIds(userId, [mandateId])
|
|
return mandateId in adminIds
|
|
|
|
|
|
def _scopedRunFilter(context: RequestContext) -> Optional[dict]:
|
|
"""
|
|
Build a DB filter dict based on RBAC:
|
|
- sysadmin: None (no filter)
|
|
- mandate admin: mandateId IN user's mandates
|
|
- normal user: ownerId = userId
|
|
"""
|
|
if context.hasSysAdminRole:
|
|
return None
|
|
|
|
userId = str(context.user.id) if context.user else None
|
|
if not userId:
|
|
return {"ownerId": "__impossible__"}
|
|
|
|
mandateIds = _getUserMandateIds(userId)
|
|
adminMandateIds = _getAdminMandateIds(userId, mandateIds)
|
|
|
|
if adminMandateIds:
|
|
return {"mandateId": adminMandateIds}
|
|
|
|
return {"ownerId": userId}
|
|
|
|
|
|
def _scopedWorkflowFilter(context: RequestContext) -> Optional[dict]:
|
|
"""
|
|
Build a DB filter for AutoWorkflow based on RBAC:
|
|
- sysadmin: None (no filter, sees all)
|
|
- normal user: mandateId IN user's mandates
|
|
"""
|
|
if context.hasSysAdminRole:
|
|
return None
|
|
|
|
userId = str(context.user.id) if context.user else None
|
|
if not userId:
|
|
return {"mandateId": "__impossible__"}
|
|
|
|
mandateIds = _getUserMandateIds(userId)
|
|
if mandateIds:
|
|
return {"mandateId": mandateIds}
|
|
|
|
return {"mandateId": "__impossible__"}
|
|
|
|
|
|
def _getManagementDb() -> DatabaseConnector:
|
|
"""Get connector to the management DB for Mandate/FeatureInstance lookups."""
|
|
return DatabaseConnector(
|
|
dbHost=APP_CONFIG.get("DB_HOST", "localhost"),
|
|
dbDatabase=APP_CONFIG.get("DB_NAME", "poweron_management"),
|
|
dbUser=APP_CONFIG.get("DB_USER"),
|
|
dbPassword=APP_CONFIG.get("DB_PASSWORD_SECRET") or APP_CONFIG.get("DB_PASSWORD"),
|
|
dbPort=int(APP_CONFIG.get("DB_PORT", 5432)),
|
|
userId=None,
|
|
)
|
|
|
|
|
|
@router.get("")
|
|
@limiter.limit("60/minute")
|
|
def get_workflow_runs(
|
|
request: Request,
|
|
limit: int = Query(50, ge=1, le=200),
|
|
offset: int = Query(0, ge=0),
|
|
status: Optional[str] = Query(None, description="Filter by status"),
|
|
mandateId: Optional[str] = Query(None, description="Filter by mandate"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""List workflow runs with RBAC scoping (SQL-paginated)."""
|
|
db = _getDb()
|
|
if not db._ensureTableExists(AutoRun):
|
|
return {"runs": [], "total": 0, "limit": limit, "offset": offset}
|
|
|
|
baseFilter = _scopedRunFilter(context)
|
|
recordFilter = dict(baseFilter) if baseFilter else {}
|
|
|
|
if status:
|
|
recordFilter["status"] = status
|
|
if mandateId:
|
|
recordFilter["mandateId"] = mandateId
|
|
|
|
page = (offset // limit) + 1 if limit > 0 else 1
|
|
pagination = PaginationParams(
|
|
page=page,
|
|
pageSize=limit,
|
|
sort=[{"field": "sysCreatedAt", "direction": "desc"}],
|
|
)
|
|
|
|
result = db.getRecordsetPaginated(
|
|
AutoRun,
|
|
pagination=pagination,
|
|
recordFilter=recordFilter if recordFilter else None,
|
|
)
|
|
pageRuns = result.get("items", []) if isinstance(result, dict) else result.items
|
|
total = result.get("totalItems", 0) if isinstance(result, dict) else result.totalItems
|
|
|
|
wfIds = list({r.get("workflowId") for r in pageRuns if r.get("workflowId")})
|
|
wfLabelMap = {}
|
|
if wfIds and db._ensureTableExists(AutoWorkflow):
|
|
wfs = db.getRecordset(AutoWorkflow, recordFilter={"id": wfIds})
|
|
for wf in (wfs or []):
|
|
wfLabelMap[wf.get("id")] = wf.get("label") or wf.get("id")
|
|
|
|
runs = []
|
|
for r in pageRuns:
|
|
row = dict(r)
|
|
row["workflowLabel"] = wfLabelMap.get(row.get("workflowId"), row.get("workflowId") or "—")
|
|
runs.append(row)
|
|
|
|
return {"runs": runs, "total": total, "limit": limit, "offset": offset}
|
|
|
|
|
|
@router.get("/metrics")
|
|
@limiter.limit("60/minute")
|
|
def get_workflow_metrics(
|
|
request: Request,
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Aggregated metrics across all accessible workflow runs (SQL COUNT)."""
|
|
db = _getDb()
|
|
if not db._ensureTableExists(AutoRun):
|
|
return {"totalRuns": 0, "runsByStatus": {}, "totalTokens": 0, "totalCredits": 0}
|
|
|
|
baseFilter = _scopedRunFilter(context)
|
|
|
|
countPagination = PaginationParams(page=1, pageSize=1)
|
|
countResult = db.getRecordsetPaginated(AutoRun, pagination=countPagination, recordFilter=baseFilter)
|
|
totalRuns = countResult.get("totalItems", 0) if isinstance(countResult, dict) else countResult.totalItems
|
|
|
|
statusValues = db.getDistinctColumnValues(AutoRun, "status", recordFilter=baseFilter)
|
|
|
|
runsByStatus = {}
|
|
for sv in statusValues:
|
|
statusFilter = dict(baseFilter) if baseFilter else {}
|
|
statusFilter["status"] = sv
|
|
sr = db.getRecordsetPaginated(AutoRun, pagination=PaginationParams(page=1, pageSize=1), recordFilter=statusFilter)
|
|
runsByStatus[sv] = sr.get("totalItems", 0) if isinstance(sr, dict) else sr.totalItems
|
|
|
|
totalTokens = 0
|
|
totalCredits = 0.0
|
|
if totalRuns > 0 and totalRuns <= 10000:
|
|
allRuns = db.getRecordset(AutoRun, recordFilter=baseFilter, fieldFilter=["costTokens", "costCredits"]) or []
|
|
for r in allRuns:
|
|
totalTokens += r.get("costTokens", 0) or 0
|
|
totalCredits += r.get("costCredits", 0.0) or 0.0
|
|
|
|
workflowCount = 0
|
|
activeWorkflows = 0
|
|
if db._ensureTableExists(AutoWorkflow):
|
|
wfFilter: dict = {"isTemplate": False}
|
|
if not context.hasSysAdminRole:
|
|
userId = str(context.user.id) if context.user else None
|
|
mandateIds = _getUserMandateIds(userId) if userId else []
|
|
if mandateIds:
|
|
wfFilter["mandateId"] = mandateIds
|
|
else:
|
|
wfFilter["mandateId"] = "__impossible__"
|
|
|
|
wfCount = db.getRecordsetPaginated(AutoWorkflow, pagination=PaginationParams(page=1, pageSize=1), recordFilter=wfFilter)
|
|
workflowCount = wfCount.get("totalItems", 0) if isinstance(wfCount, dict) else wfCount.totalItems
|
|
|
|
activeFilter = dict(wfFilter)
|
|
activeFilter["active"] = True
|
|
activeCount = db.getRecordsetPaginated(AutoWorkflow, pagination=PaginationParams(page=1, pageSize=1), recordFilter=activeFilter)
|
|
activeWorkflows = activeCount.get("totalItems", 0) if isinstance(activeCount, dict) else activeCount.totalItems
|
|
|
|
return {
|
|
"totalRuns": totalRuns,
|
|
"runsByStatus": runsByStatus,
|
|
"totalTokens": totalTokens,
|
|
"totalCredits": round(totalCredits, 4),
|
|
"workflowCount": workflowCount,
|
|
"activeWorkflows": activeWorkflows,
|
|
}
|
|
|
|
|
|
@router.get("/{runId}/steps")
|
|
@limiter.limit("60/minute")
|
|
def get_run_steps(
|
|
request: Request,
|
|
runId: str = Path(..., description="Run ID"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""Get step logs for a specific run (with access check)."""
|
|
db = _getDb()
|
|
if not db._ensureTableExists(AutoRun):
|
|
raise HTTPException(status_code=404, detail=routeApiMsg("Run not found"))
|
|
|
|
runs = db.getRecordset(AutoRun, recordFilter={"id": runId})
|
|
if not runs:
|
|
raise HTTPException(status_code=404, detail=routeApiMsg("Run not found"))
|
|
run = dict(runs[0])
|
|
|
|
if not context.hasSysAdminRole:
|
|
userId = str(context.user.id) if context.user else None
|
|
runOwner = run.get("ownerId")
|
|
runMandate = run.get("mandateId")
|
|
|
|
if runOwner == userId:
|
|
pass
|
|
elif runMandate and userId and _isUserMandateAdmin(userId, runMandate):
|
|
pass
|
|
else:
|
|
raise HTTPException(status_code=403, detail=routeApiMsg("Access denied"))
|
|
|
|
if not db._ensureTableExists(AutoStepLog):
|
|
return {"steps": []}
|
|
|
|
records = db.getRecordset(AutoStepLog, recordFilter={"runId": runId})
|
|
steps = [dict(r) for r in records] if records else []
|
|
steps.sort(key=lambda s: s.get("startedAt") or 0)
|
|
return {"steps": steps}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# System-level Workflow listing (all workflows the user can see via RBAC)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.get("/workflows")
|
|
@limiter.limit("60/minute")
|
|
def get_system_workflows(
|
|
request: Request,
|
|
active: Optional[bool] = Query(None, description="Filter by active status"),
|
|
mandateId: Optional[str] = Query(None, description="Filter by mandate"),
|
|
pagination: Optional[str] = Query(None, description="JSON-encoded PaginationParams"),
|
|
context: RequestContext = Depends(getRequestContext),
|
|
) -> dict:
|
|
"""List all workflows the user has access to (RBAC-scoped, cross-instance)."""
|
|
db = _getDb()
|
|
if not db._ensureTableExists(AutoWorkflow):
|
|
return {"items": [], "pagination": {"currentPage": 1, "pageSize": 25, "totalItems": 0, "totalPages": 0}}
|
|
|
|
baseFilter = _scopedWorkflowFilter(context)
|
|
recordFilter = dict(baseFilter) if baseFilter else {}
|
|
recordFilter["isTemplate"] = False
|
|
|
|
if active is not None:
|
|
recordFilter["active"] = active
|
|
if mandateId:
|
|
recordFilter["mandateId"] = mandateId
|
|
|
|
paginationParams = None
|
|
if pagination:
|
|
try:
|
|
paginationParams = PaginationParams(**json.loads(pagination))
|
|
except Exception:
|
|
pass
|
|
|
|
if not paginationParams:
|
|
paginationParams = PaginationParams(
|
|
page=1,
|
|
pageSize=25,
|
|
sort=[{"field": "sysCreatedAt", "direction": "desc"}],
|
|
)
|
|
|
|
result = db.getRecordsetPaginated(
|
|
AutoWorkflow,
|
|
pagination=paginationParams,
|
|
recordFilter=recordFilter if recordFilter else None,
|
|
)
|
|
pageItems = result.get("items", []) if isinstance(result, dict) else result.items
|
|
totalItems = result.get("totalItems", 0) if isinstance(result, dict) else result.totalItems
|
|
totalPages = result.get("totalPages", 0) if isinstance(result, dict) else result.totalPages
|
|
|
|
mandateIds = list({w.get("mandateId") for w in pageItems if w.get("mandateId")})
|
|
instanceIds = list({w.get("featureInstanceId") for w in pageItems if w.get("featureInstanceId")})
|
|
|
|
mandateLabelMap: dict = {}
|
|
instanceLabelMap: dict = {}
|
|
try:
|
|
mgmtDb = _getManagementDb()
|
|
if mandateIds and mgmtDb._ensureTableExists(Mandate):
|
|
mandates = mgmtDb.getRecordset(Mandate, recordFilter={"id": mandateIds})
|
|
for m in (mandates or []):
|
|
row = dict(m)
|
|
mandateLabelMap[row.get("id")] = row.get("label") or row.get("name") or row.get("id")
|
|
if instanceIds and mgmtDb._ensureTableExists(FeatureInstance):
|
|
instances = mgmtDb.getRecordset(FeatureInstance, recordFilter={"id": instanceIds})
|
|
for fi in (instances or []):
|
|
row = dict(fi)
|
|
instanceLabelMap[row.get("id")] = row.get("label") or row.get("id")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to enrich workflow labels: {e}")
|
|
|
|
userId = str(context.user.id) if context.user else None
|
|
adminMandateIds = []
|
|
if userId and not context.hasSysAdminRole:
|
|
userMandateIds = _getUserMandateIds(userId)
|
|
adminMandateIds = _getAdminMandateIds(userId, userMandateIds)
|
|
|
|
items = []
|
|
for w in pageItems:
|
|
row = dict(w)
|
|
wMandateId = row.get("mandateId")
|
|
row["mandateLabel"] = mandateLabelMap.get(wMandateId, wMandateId or "—")
|
|
row["instanceLabel"] = instanceLabelMap.get(row.get("featureInstanceId"), row.get("featureInstanceId") or "—")
|
|
|
|
if context.hasSysAdminRole:
|
|
row["canEdit"] = True
|
|
row["canDelete"] = True
|
|
row["canExecute"] = True
|
|
elif wMandateId and wMandateId in adminMandateIds:
|
|
row["canEdit"] = True
|
|
row["canDelete"] = True
|
|
row["canExecute"] = True
|
|
else:
|
|
row["canEdit"] = False
|
|
row["canDelete"] = False
|
|
row["canExecute"] = False
|
|
|
|
row.pop("graph", None)
|
|
|
|
items.append(row)
|
|
|
|
return {
|
|
"items": items,
|
|
"pagination": {
|
|
"currentPage": paginationParams.page,
|
|
"pageSize": paginationParams.pageSize,
|
|
"totalItems": totalItems,
|
|
"totalPages": totalPages,
|
|
},
|
|
}
|