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

mandate invitation and notification system

Browse source
This commit is contained in:
ValueOn AG 2026-01-26 01:29:17 +01:00
parent e737bf5cdb
commit a0304c6d78
9 changed files with 1072 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
app.py
View file

@ -492,6 +492,9 @@ app.include_router(featuresAdminRouter)
from modules.routes.routeInvitations import router as invitationsRouter
app.include_router(invitationsRouter)
from modules.routes.routeNotifications import router as notificationsRouter
app.include_router(notificationsRouter)
from modules.routes.routeAdminRbacExport import router as rbacAdminExportRouter
app.include_router(rbacAdminExportRouter)

48
modules/connectors/connectorDbPostgre.py
View file

@ -40,6 +40,34 @@ class SystemTable(BaseModel):
)
def _isJsonbType(fieldType) -> bool:
"""Check if a type should be stored as JSONB in PostgreSQL."""
# Direct dict or list
if fieldType == dict or fieldType == list:
return True
# Generic List[X] or Dict[X, Y]
origin = get_origin(fieldType)
if origin in (dict, list):
return True
# Direct Pydantic BaseModel subclass
if isinstance(fieldType, type) and issubclass(fieldType, BaseModel):
return True
# Optional[X] - check the inner type
if origin is Union:
args = get_args(fieldType)
for arg in args:
if arg is type(None):
continue
# Recursively check the inner type
if _isJsonbType(arg):
return True
return False
def _get_model_fields(model_class) -> Dict[str, str]:
"""Get all fields from Pydantic model and map to SQL types."""
# Pydantic v2
@ -52,20 +80,7 @@ def _get_model_fields(model_class) -> Dict[str, str]:
# Check for JSONB fields (Dict, List, or complex types)
# Purely type-based detection - no hardcoded field names
if (
field_type == dict
or field_type == list
or (
hasattr(field_type, "__origin__")
and field_type.__origin__ in (dict, list)
)
# Check if field type is directly a Pydantic BaseModel subclass (for nested models like TextMultilingual)
or (isinstance(field_type, type) and issubclass(field_type, BaseModel))
# Check if field type is Optional[BaseModel] (Union with None)
or (hasattr(field_type, "__origin__") and get_origin(field_type) is Union
and any(isinstance(arg, type) and issubclass(arg, BaseModel)
for arg in get_args(field_type) if arg is not type(None)))
):
if _isJsonbType(field_type):
fields[field_name] = "JSONB"
# Simple type mapping
elif field_type in (str, type(None)) or (
@ -970,7 +985,10 @@ class DatabaseConnector:
record["id"] = str(uuid.uuid4())
# Save record
self._saveRecord(model_class, record["id"], record)
success = self._saveRecord(model_class, record["id"], record)
if not success:
table = model_class.__name__
raise ValueError(f"Failed to save record {record['id']} to table {table}")
# Check if this is the first record in the table and register as initial ID
table = model_class.__name__

17
modules/datamodels/datamodelInvitation.py
View file

@ -46,9 +46,13 @@ class Invitation(BaseModel):
)
# Einladungs-Details
targetUsername: str = Field(
description="Username of the invited user (must match on acceptance)",
json_schema_extra={"frontend_type": "text", "frontend_readonly": False, "frontend_required": True}
)
email: Optional[str] = Field(
default=None,
description="Target email address (optional, for tracking)",
description="Email address to send invitation link (optional)",
json_schema_extra={"frontend_type": "email", "frontend_readonly": False, "frontend_required": False}
)
createdBy: str = Field(
@ -82,6 +86,13 @@ class Invitation(BaseModel):
json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False}
)
# Email-Status
emailSent: bool = Field(
default=False,
description="Whether the invitation email was successfully sent",
json_schema_extra={"frontend_type": "checkbox", "frontend_readonly": True, "frontend_required": False}
)
# Einschränkungen
maxUses: int = Field(
default=1,
@ -107,13 +118,15 @@ registerModelLabels(
"mandateId": {"en": "Mandate", "de": "Mandant", "fr": "Mandat"},
"featureInstanceId": {"en": "Feature Instance", "de": "Feature-Instanz", "fr": "Instance"},
"roleIds": {"en": "Roles", "de": "Rollen", "fr": "Rôles"},
"email": {"en": "Email", "de": "E-Mail", "fr": "Email"},
"targetUsername": {"en": "Target Username", "de": "Ziel-Benutzername", "fr": "Nom d'utilisateur cible"},
"email": {"en": "Email (optional)", "de": "E-Mail (optional)", "fr": "Email (optionnel)"},
"createdBy": {"en": "Created By", "de": "Erstellt von", "fr": "Créé par"},
"createdAt": {"en": "Created At", "de": "Erstellt am", "fr": "Créé le"},
"expiresAt": {"en": "Expires At", "de": "Gültig bis", "fr": "Expire le"},
"usedBy": {"en": "Used By", "de": "Verwendet von", "fr": "Utilisé par"},
"usedAt": {"en": "Used At", "de": "Verwendet am", "fr": "Utilisé le"},
"revokedAt": {"en": "Revoked At", "de": "Widerrufen am", "fr": "Révoqué le"},
"emailSent": {"en": "Email Sent", "de": "E-Mail gesendet", "fr": "Email envoyé"},
"maxUses": {"en": "Max Uses", "de": "Max. Verwendungen", "fr": "Utilisations max"},
"currentUses": {"en": "Current Uses", "de": "Aktuelle Verwendungen", "fr": "Utilisations actuelles"},
},

209
modules/datamodels/datamodelNotification.py Normal file
View file

@ -0,0 +1,209 @@
# Copyright (c) 2025 Patrick Motsch
# All rights reserved.
"""
Notification model for in-app notifications.
Supports actionable notifications (e.g., invitation accept/decline).
"""
import uuid
from typing import Optional, List
from enum import Enum
from pydantic import BaseModel, Field, ConfigDict
from modules.shared.attributeUtils import registerModelLabels
from modules.shared.timeUtils import getUtcTimestamp
class NotificationType(str, Enum):
"""Types of notifications"""
INVITATION = "invitation" # Einladung zu Mandat/Feature
SYSTEM = "system" # System-Nachrichten
WORKFLOW = "workflow" # Workflow-Status Updates
MENTION = "mention" # Erwähnung in Chat/Kommentar
class NotificationStatus(str, Enum):
"""Status of a notification"""
UNREAD = "unread" # Noch nicht gelesen
READ = "read" # Gelesen
ACTIONED = "actioned" # Aktion wurde durchgeführt
DISMISSED = "dismissed" # Verworfen/Geschlossen
class NotificationAction(BaseModel):
"""Possible action for a notification"""
actionId: str = Field(
description="Unique identifier for the action (e.g., 'accept', 'decline')"
)
label: str = Field(
description="Display label for the action button"
)
style: str = Field(
default="default",
description="Button style: 'primary', 'danger', 'default'"
)
class UserNotification(BaseModel):
"""
In-app notification for a user.
Supports actionable notifications with accept/decline buttons.
"""
id: str = Field(
default_factory=lambda: str(uuid.uuid4()),
description="Unique ID of the notification",
json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False}
)
userId: str = Field(
description="Target user ID for this notification",
json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": True}
)
# Notification type and status
type: NotificationType = Field(
default=NotificationType.SYSTEM,
description="Type of notification",
json_schema_extra={
"frontend_type": "select",
"frontend_readonly": True,
"frontend_required": True,
"frontend_options": [
{"value": "invitation", "label": {"en": "Invitation", "de": "Einladung"}},
{"value": "system", "label": {"en": "System", "de": "System"}},
{"value": "workflow", "label": {"en": "Workflow", "de": "Workflow"}},
{"value": "mention", "label": {"en": "Mention", "de": "Erwähnung"}}
]
}
)
status: NotificationStatus = Field(
default=NotificationStatus.UNREAD,
description="Current status of the notification",
json_schema_extra={
"frontend_type": "select",
"frontend_readonly": True,
"frontend_required": False,
"frontend_options": [
{"value": "unread", "label": {"en": "Unread", "de": "Ungelesen"}},
{"value": "read", "label": {"en": "Read", "de": "Gelesen"}},
{"value": "actioned", "label": {"en": "Actioned", "de": "Bearbeitet"}},
{"value": "dismissed", "label": {"en": "Dismissed", "de": "Verworfen"}}
]
}
)
# Content
title: str = Field(
description="Notification title",
json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": True}
)
message: str = Field(
description="Notification message/body",
json_schema_extra={"frontend_type": "textarea", "frontend_readonly": True, "frontend_required": True}
)
icon: Optional[str] = Field(
default=None,
description="Optional icon identifier (e.g., 'mail', 'warning', 'info')",
json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False}
)
# Reference to triggering object (for actionable notifications)
referenceType: Optional[str] = Field(
default=None,
description="Type of referenced object (e.g., 'Invitation', 'Workflow')",
json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False}
)
referenceId: Optional[str] = Field(
default=None,
description="ID of referenced object",
json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False}
)
# Actions (for actionable notifications like invitations)
actions: Optional[List[NotificationAction]] = Field(
default=None,
description="List of possible actions for this notification",
json_schema_extra={"frontend_type": "json", "frontend_readonly": True, "frontend_required": False}
)
# Action result (when user takes action)
actionTaken: Optional[str] = Field(
default=None,
description="Which action was taken (actionId)",
json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False}
)
actionResult: Optional[str] = Field(
default=None,
description="Result message from the action",
json_schema_extra={"frontend_type": "textarea", "frontend_readonly": True, "frontend_required": False}
)
# Timestamps
createdAt: float = Field(
default_factory=getUtcTimestamp,
description="When the notification was created (UTC timestamp)",
json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False}
)
readAt: Optional[float] = Field(
default=None,
description="When the notification was read (UTC timestamp)",
json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False}
)
actionedAt: Optional[float] = Field(
default=None,
description="When action was taken (UTC timestamp)",
json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False}
)
expiresAt: Optional[float] = Field(
default=None,
description="When the notification expires (optional, UTC timestamp)",
json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False}
)
model_config = ConfigDict(use_enum_values=True)
registerModelLabels(
"UserNotification",
{"en": "Notification", "de": "Benachrichtigung", "fr": "Notification"},
{
"id": {"en": "ID", "de": "ID", "fr": "ID"},
"userId": {"en": "User", "de": "Benutzer", "fr": "Utilisateur"},
"type": {"en": "Type", "de": "Typ", "fr": "Type"},
"status": {"en": "Status", "de": "Status", "fr": "Statut"},
"title": {"en": "Title", "de": "Titel", "fr": "Titre"},
"message": {"en": "Message", "de": "Nachricht", "fr": "Message"},
"icon": {"en": "Icon", "de": "Symbol", "fr": "Icône"},
"referenceType": {"en": "Reference Type", "de": "Referenz-Typ", "fr": "Type de référence"},
"referenceId": {"en": "Reference ID", "de": "Referenz-ID", "fr": "ID de référence"},
"actions": {"en": "Actions", "de": "Aktionen", "fr": "Actions"},
"actionTaken": {"en": "Action Taken", "de": "Durchgeführte Aktion", "fr": "Action effectuée"},
"actionResult": {"en": "Action Result", "de": "Aktions-Ergebnis", "fr": "Résultat de l'action"},
"createdAt": {"en": "Created At", "de": "Erstellt am", "fr": "Créé le"},
"readAt": {"en": "Read At", "de": "Gelesen am", "fr": "Lu le"},
"actionedAt": {"en": "Actioned At", "de": "Bearbeitet am", "fr": "Traité le"},
"expiresAt": {"en": "Expires At", "de": "Gültig bis", "fr": "Expire le"},
},
)
registerModelLabels(
"NotificationType",
{"en": "Notification Type", "de": "Benachrichtigungs-Typ", "fr": "Type de notification"},
{
"invitation": {"en": "Invitation", "de": "Einladung", "fr": "Invitation"},
"system": {"en": "System", "de": "System", "fr": "Système"},
"workflow": {"en": "Workflow", "de": "Workflow", "fr": "Workflow"},
"mention": {"en": "Mention", "de": "Erwähnung", "fr": "Mention"},
},
)
registerModelLabels(
"NotificationStatus",
{"en": "Notification Status", "de": "Benachrichtigungs-Status", "fr": "Statut de notification"},
{
"unread": {"en": "Unread", "de": "Ungelesen", "fr": "Non lu"},
"read": {"en": "Read", "de": "Gelesen", "fr": "Lu"},
"actioned": {"en": "Actioned", "de": "Bearbeitet", "fr": "Traité"},
"dismissed": {"en": "Dismissed", "de": "Verworfen", "fr": "Rejeté"},
},
)

53
modules/routes/routeAdminFeatures.py
View file

@ -21,6 +21,7 @@ from modules.datamodels.datamodelUam import User, UserInDB
from modules.datamodels.datamodelFeatures import Feature, FeatureInstance
from modules.interfaces.interfaceDbApp import getRootInterface
from modules.interfaces.interfaceFeatures import getFeatureInterface
from modules.security.rbacCatalog import getCatalogService
logger = logging.getLogger(__name__)
@ -72,15 +73,16 @@ async def list_features(
"""
List all available features.
Returns global feature definitions that can be activated for mandates.
Returns global feature definitions from the RBAC Catalog.
Features are automatically registered at startup from feature containers.
Any authenticated user can see available features.
"""
try:
rootInterface = getRootInterface()
featureInterface = getFeatureInterface(rootInterface.db)
features = featureInterface.getAllFeatures()
return [f.model_dump() for f in features]
# Features come from the RBAC Catalog (registered at startup from feature containers)
# NOT from the database - features are code-defined, not user-created
catalogService = getCatalogService()
features = catalogService.getFeatureDefinitions()
return features
except Exception as e:
logger.error(f"Error listing features: {e}")
@ -153,14 +155,15 @@ async def get_my_feature_instances(
"features": []
}
# Get feature info
# Get feature info from catalog (features are code-defined)
featureKey = f"{mandateId}_{instance.featureCode}"
if featureKey not in featuresMap:
feature = featureInterface.getFeature(instance.featureCode)
catalogService = getCatalogService()
featureDef = catalogService.getFeatureDefinition(instance.featureCode)
featuresMap[featureKey] = {
"code": instance.featureCode,
"label": feature.label if feature and hasattr(feature, 'label') else {"de": instance.featureCode, "en": instance.featureCode},
"icon": feature.icon if feature and hasattr(feature, 'icon') else "folder",
"label": featureDef.get("label", {"de": instance.featureCode, "en": instance.featureCode}) if featureDef else {"de": instance.featureCode, "en": instance.featureCode},
"icon": featureDef.get("icon", "folder") if featureDef else "folder",
"instances": [],
"_mandateId": mandateId # Temporary for grouping
}
@ -376,8 +379,9 @@ async def create_feature(
rootInterface = getRootInterface()
featureInterface = getFeatureInterface(rootInterface.db)
# Check if feature already exists
existing = featureInterface.getFeature(code)
# Check if feature already exists in catalog (features are code-defined)
catalogService = getCatalogService()
existing = catalogService.getFeatureDefinition(code)
if existing:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
@ -525,9 +529,10 @@ async def create_feature_instance(
rootInterface = getRootInterface()
featureInterface = getFeatureInterface(rootInterface.db)
# Verify feature exists
feature = featureInterface.getFeature(data.featureCode)
if not feature:
# Verify feature exists in catalog (features are code-defined, not DB-stored)
catalogService = getCatalogService()
featureDef = catalogService.getFeatureDefinition(data.featureCode)
if not featureDef:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Feature '{data.featureCode}' not found"
@ -818,9 +823,10 @@ async def create_template_role(
rootInterface = getRootInterface()
featureInterface = getFeatureInterface(rootInterface.db)
# Verify feature exists
feature = featureInterface.getFeature(featureCode)
if not feature:
# Verify feature exists in catalog (features are code-defined)
catalogService = getCatalogService()
featureDef = catalogService.getFeatureDefinition(featureCode)
if not featureDef:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Feature '{featureCode}' not found"
@ -1331,17 +1337,16 @@ async def get_feature(
featureCode: Feature code (e.g., 'trustee', 'chatbot')
"""
try:
rootInterface = getRootInterface()
featureInterface = getFeatureInterface(rootInterface.db)
feature = featureInterface.getFeature(featureCode)
if not feature:
# Features come from the RBAC Catalog (code-defined, not DB-stored)
catalogService = getCatalogService()
featureDef = catalogService.getFeatureDefinition(featureCode)
if not featureDef:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Feature '{featureCode}' not found"
)
return feature.model_dump()
return featureDef
except HTTPException:
raise

152
modules/routes/routeInvitations.py
View file

@ -37,7 +37,8 @@ router = APIRouter(
class InvitationCreate(BaseModel):
"""Request model for creating an invitation"""
email: Optional[str] = Field(None, description="Target email address (optional)")
targetUsername: str = Field(..., description="Username of the user to invite (must match on acceptance)")
email: Optional[str] = Field(None, description="Email address to send invitation link (optional)")
roleIds: List[str] = Field(..., description="Role IDs to assign to the invited user")
featureInstanceId: Optional[str] = Field(None, description="Optional feature instance access")
expiresInHours: int = Field(
@ -61,6 +62,7 @@ class InvitationResponse(BaseModel):
mandateId: str
featureInstanceId: Optional[str]
roleIds: List[str]
targetUsername: str
email: Optional[str]
createdBy: str
createdAt: float
@ -71,6 +73,7 @@ class InvitationResponse(BaseModel):
maxUses: int
currentUses: int
inviteUrl: str # Full URL for the invitation
emailSent: bool = False # Whether invitation email was sent
class InvitationValidation(BaseModel):
@ -78,8 +81,11 @@ class InvitationValidation(BaseModel):
valid: bool
reason: Optional[str]
mandateId: Optional[str]
mandateName: Optional[str] = None
featureInstanceId: Optional[str]
roleIds: List[str]
roleLabels: List[str] = []
targetUsername: Optional[str] = None
# =============================================================================
@ -118,6 +124,11 @@ async def create_invitation(
try:
rootInterface = getRootInterface()
# Note: targetUsername does NOT need to exist yet!
# The invitation can be for a user who will register later.
# When they register with this username (or accept the invitation),
# they will get the assigned roles.
# Validate role IDs exist and belong to this mandate or are global
for roleId in data.roleIds:
from modules.datamodels.datamodelRbac import Role
@ -164,6 +175,7 @@ async def create_invitation(
mandateId=str(context.mandateId),
featureInstanceId=data.featureInstanceId,
roleIds=data.roleIds,
targetUsername=data.targetUsername,
email=data.email,
createdBy=str(context.user.id),
expiresAt=expiresAt,
@ -179,9 +191,98 @@ async def create_invitation(
frontendUrl = APP_CONFIG.get("APP_FRONTEND_URL", "http://localhost:8080")
inviteUrl = f"{frontendUrl}/invite/{invitation.token}"
# Send email if email address is provided
emailSent = False
if data.email:
try:
from modules.connectors.connectorMessagingEmail import ConnectorMessagingEmail
from modules.datamodels.datamodelUam import Mandate
# Get mandate name for the email
mandateRecords = rootInterface.db.getRecordset(
Mandate,
recordFilter={"id": str(context.mandateId)}
)
mandateName = mandateRecords[0].get("name", "PowerOn") if mandateRecords else "PowerOn"
emailConnector = ConnectorMessagingEmail()
emailSubject = f"Einladung zu {mandateName}"
emailBody = f"""
<html>
<body style="font-family: Arial, sans-serif; line-height: 1.6; color: #333;">
<h2>Sie wurden eingeladen!</h2>
<p>Hallo <strong>{data.targetUsername}</strong>,</p>
<p>Sie wurden eingeladen, dem Mandanten <strong>{mandateName}</strong> beizutreten.</p>
<p>Klicken Sie auf den folgenden Link, um die Einladung anzunehmen:</p>
<p style="margin: 20px 0;">
<a href="{inviteUrl}" style="background-color: #007bff; color: white; padding: 12px 24px; text-decoration: none; border-radius: 4px;">
Einladung annehmen
</a>
</p>
<p style="color: #666; font-size: 0.9em;">
Oder kopieren Sie diesen Link in Ihren Browser:<br>
<code>{inviteUrl}</code>
</p>
<p style="color: #666; font-size: 0.9em;">
Diese Einladung ist {data.expiresInHours} Stunden gültig.
</p>
<hr style="border: none; border-top: 1px solid #eee; margin: 20px 0;">
<p style="color: #999; font-size: 0.8em;">
Diese E-Mail wurde automatisch von PowerOn gesendet.
</p>
</body>
</html>
"""
emailConnector.send(
recipient=data.email,
subject=emailSubject,
message=emailBody
)
emailSent = True
logger.info(f"Invitation email sent to {data.email} for user {data.targetUsername}")
except Exception as emailError:
logger.warning(f"Failed to send invitation email to {data.email}: {emailError}")
# Don't fail the invitation creation if email fails
# Update the invitation record with emailSent status
if emailSent:
rootInterface.db.recordModify(
Invitation,
createdRecord.get("id"),
{"emailSent": True}
)
createdRecord["emailSent"] = True
# If the target user already exists, create an in-app notification
try:
existingUser = rootInterface.getUserByUsername(data.targetUsername)
if existingUser:
from modules.routes.routeNotifications import createInvitationNotification
from modules.datamodels.datamodelUam import Mandate
# Get mandate name for notification
mandateRecords = rootInterface.db.getRecordset(
Mandate,
recordFilter={"id": str(context.mandateId)}
)
mandateName = mandateRecords[0].get("mandateLabel", "PowerOn") if mandateRecords else "PowerOn"
inviterName = context.user.fullName or context.user.username
createInvitationNotification(
userId=str(existingUser.id),
invitationId=str(createdRecord.get("id")),
mandateName=mandateName,
inviterName=inviterName
)
logger.info(f"Created notification for existing user {data.targetUsername}")
except Exception as notifError:
logger.warning(f"Failed to create notification for user {data.targetUsername}: {notifError}")
# Don't fail the invitation if notification fails
logger.info(
f"User {context.user.id} created invitation for mandate {context.mandateId}, "
f"expires in {data.expiresInHours}h"
f"User {context.user.id} created invitation for user {data.targetUsername} "
f"to mandate {context.mandateId}, expires in {data.expiresInHours}h"
)
return InvitationResponse(
@ -190,6 +291,7 @@ async def create_invitation(
mandateId=str(createdRecord.get("mandateId")),
featureInstanceId=createdRecord.get("featureInstanceId"),
roleIds=createdRecord.get("roleIds", []),
targetUsername=createdRecord.get("targetUsername"),
email=createdRecord.get("email"),
createdBy=str(createdRecord.get("createdBy")),
createdAt=createdRecord.get("createdAt"),
@ -199,7 +301,8 @@ async def create_invitation(
revokedAt=createdRecord.get("revokedAt"),
maxUses=createdRecord.get("maxUses", 1),
currentUses=createdRecord.get("currentUses", 0),
inviteUrl=inviteUrl
inviteUrl=inviteUrl,
emailSent=emailSent
)
except HTTPException:
@ -441,12 +544,38 @@ async def validate_invitation(
roleIds=[]
)
# Get additional info for display
mandateId = invitation.get("mandateId")
mandateName = None
roleLabels = []
targetUsername = invitation.get("targetUsername")
# Get mandate name
from modules.datamodels.datamodelUam import Mandate
mandateRecords = rootInterface.db.getRecordset(
Mandate,
recordFilter={"id": mandateId}
)
if mandateRecords:
mandateName = mandateRecords[0].get("name")
# Get role names
roleIds = invitation.get("roleIds", [])
from modules.datamodels.datamodelRbac import Role
for roleId in roleIds:
roleRecords = rootInterface.db.getRecordset(Role, recordFilter={"id": roleId})
if roleRecords:
roleLabels.append(roleRecords[0].get("roleLabel", roleId))
return InvitationValidation(
valid=True,
reason=None,
mandateId=invitation.get("mandateId"),
mandateId=mandateId,
mandateName=mandateName,
featureInstanceId=invitation.get("featureInstanceId"),
roleIds=invitation.get("roleIds", [])
roleIds=roleIds,
roleLabels=roleLabels,
targetUsername=targetUsername
)
except Exception as e:
@ -513,6 +642,17 @@ async def accept_invitation(
detail="Invitation has reached maximum uses"
)
# Validate username matches - the invitation is bound to a specific user
targetUsername = invitation.get("targetUsername")
if targetUsername and currentUser.username != targetUsername:
logger.warning(
f"User {currentUser.username} tried to accept invitation meant for {targetUsername}"
)
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Diese Einladung ist für Benutzer '{targetUsername}' bestimmt"
)
mandateId = invitation.get("mandateId")
roleIds = invitation.get("roleIds", [])
featureInstanceId = invitation.get("featureInstanceId")

575
modules/routes/routeNotifications.py Normal file
View file

@ -0,0 +1,575 @@
# Copyright (c) 2025 Patrick Motsch
# All rights reserved.
"""
Notification routes for in-app notifications.
Provides user-specific notification inbox with support for actionable notifications.
"""
from fastapi import APIRouter, HTTPException, Depends, Request
from typing import List, Dict, Any, Optional
from fastapi import status
import logging
from pydantic import BaseModel, Field
from modules.auth import limiter, getCurrentUser
from modules.datamodels.datamodelUam import User
from modules.datamodels.datamodelNotification import (
UserNotification,
NotificationType,
NotificationStatus,
NotificationAction
)
from modules.interfaces.interfaceDbApp import getRootInterface
from modules.shared.timeUtils import getUtcTimestamp
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/api/notifications",
tags=["Notifications"],
responses={404: {"description": "Not found"}}
)
# =============================================================================
# Request/Response Models
# =============================================================================
class NotificationActionRequest(BaseModel):
"""Request model for executing a notification action"""
actionId: str = Field(..., description="ID of the action to execute (e.g., 'accept', 'decline')")
class UnreadCountResponse(BaseModel):
"""Response model for unread count"""
count: int
# =============================================================================
# Helper Functions
# =============================================================================
def _createNotification(
userId: str,
notificationType: NotificationType,
title: str,
message: str,
referenceType: Optional[str] = None,
referenceId: Optional[str] = None,
actions: Optional[List[NotificationAction]] = None,
icon: Optional[str] = None,
expiresAt: Optional[float] = None
) -> UserNotification:
"""
Create a notification for a user.
This is a helper function that can be imported by other modules.
"""
rootInterface = getRootInterface()
notification = UserNotification(
userId=userId,
type=notificationType,
title=title,
message=message,
referenceType=referenceType,
referenceId=referenceId,
actions=actions,
icon=icon,
expiresAt=expiresAt
)
# Store in database
rootInterface.db.recordCreate(
model_class=UserNotification,
record=notification.model_dump()
)
logger.info(f"Created notification {notification.id} for user {userId}: {title}")
return notification
def createInvitationNotification(
userId: str,
invitationId: str,
mandateName: str,
inviterName: str
) -> UserNotification:
"""
Create a notification for a pending invitation.
Called when an invitation is created for an existing user.
"""
return _createNotification(
userId=userId,
notificationType=NotificationType.INVITATION,
title="Neue Einladung",
message=f"{inviterName} hat Sie zu '{mandateName}' eingeladen.",
referenceType="Invitation",
referenceId=invitationId,
icon="mail",
actions=[
NotificationAction(actionId="accept", label="Annehmen", style="primary"),
NotificationAction(actionId="decline", label="Ablehnen", style="danger")
]
)
# =============================================================================
# API Endpoints
# =============================================================================
@router.get("", response_model=List[Dict[str, Any]])
@limiter.limit("60/minute")
async def getNotifications(
request: Request,
currentUser: User = Depends(getCurrentUser),
status: Optional[str] = None,
type: Optional[str] = None,
limit: int = 50
) -> List[Dict[str, Any]]:
"""
Get all notifications for the current user.
Optionally filter by status (unread, read, actioned, dismissed) or type.
"""
try:
rootInterface = getRootInterface()
# Build filter
recordFilter = {"userId": str(currentUser.id)}
if status:
recordFilter["status"] = status
if type:
recordFilter["type"] = type
# Get notifications
notifications = rootInterface.db.getRecordset(
model_class=UserNotification,
recordFilter=recordFilter
)
# Sort by creation date (newest first) and limit
notifications = sorted(notifications, key=lambda x: x.get("createdAt", 0), reverse=True)
if limit:
notifications = notifications[:limit]
return notifications
except Exception as e:
logger.error(f"Error getting notifications: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get notifications: {str(e)}"
)
@router.get("/unread-count", response_model=UnreadCountResponse)
@limiter.limit("120/minute")
async def getUnreadCount(
request: Request,
currentUser: User = Depends(getCurrentUser)
) -> UnreadCountResponse:
"""
Get the count of unread notifications for the current user.
Used for the notification badge in the header.
"""
try:
rootInterface = getRootInterface()
notifications = rootInterface.db.getRecordset(
model_class=UserNotification,
recordFilter={
"userId": str(currentUser.id),
"status": NotificationStatus.UNREAD.value
}
)
return UnreadCountResponse(count=len(notifications))
except Exception as e:
logger.error(f"Error getting unread count: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to get unread count: {str(e)}"
)
@router.put("/{notificationId}/read", response_model=Dict[str, Any])
@limiter.limit("60/minute")
async def markAsRead(
request: Request,
notificationId: str,
currentUser: User = Depends(getCurrentUser)
) -> Dict[str, Any]:
"""
Mark a notification as read.
"""
try:
rootInterface = getRootInterface()
# Get the notification
notifications = rootInterface.db.getRecordset(
model_class=UserNotification,
recordFilter={"id": notificationId}
)
if not notifications:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Notification not found"
)
notification = notifications[0]
# Verify ownership
if notification.get("userId") != currentUser.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to access this notification"
)
# Update status
rootInterface.db.recordModify(
model_class=UserNotification,
recordId=notificationId,
record={
"status": NotificationStatus.READ.value,
"readAt": getUtcTimestamp()
}
)
return {"message": "Notification marked as read", "id": notificationId}
except HTTPException:
raise
except Exception as e:
logger.error(f"Error marking notification as read: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to mark notification as read: {str(e)}"
)
@router.put("/mark-all-read", response_model=Dict[str, Any])
@limiter.limit("10/minute")
async def markAllAsRead(
request: Request,
currentUser: User = Depends(getCurrentUser)
) -> Dict[str, Any]:
"""
Mark all notifications as read for the current user.
"""
try:
rootInterface = getRootInterface()
# Get all unread notifications
notifications = rootInterface.db.getRecordset(
model_class=UserNotification,
recordFilter={
"userId": currentUser.id,
"status": NotificationStatus.UNREAD.value
}
)
currentTime = getUtcTimestamp()
updatedCount = 0
for notification in notifications:
rootInterface.db.recordModify(
model_class=UserNotification,
recordId=notification.get("id"),
record={
"status": NotificationStatus.READ.value,
"readAt": currentTime
}
)
updatedCount += 1
return {"message": f"Marked {updatedCount} notifications as read", "count": updatedCount}
except Exception as e:
logger.error(f"Error marking all notifications as read: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to mark notifications as read: {str(e)}"
)
@router.post("/{notificationId}/action", response_model=Dict[str, Any])
@limiter.limit("30/minute")
async def executeAction(
request: Request,
notificationId: str,
actionRequest: NotificationActionRequest,
currentUser: User = Depends(getCurrentUser)
) -> Dict[str, Any]:
"""
Execute an action on a notification (e.g., accept/decline invitation).
"""
try:
rootInterface = getRootInterface()
# Get the notification
notifications = rootInterface.db.getRecordset(
model_class=UserNotification,
recordFilter={"id": notificationId}
)
if not notifications:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Notification not found"
)
notification = notifications[0]
# Verify ownership
if notification.get("userId") != currentUser.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to access this notification"
)
# Check if already actioned
if notification.get("status") == NotificationStatus.ACTIONED.value:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Notification has already been actioned"
)
# Validate action exists
actions = notification.get("actions", [])
validActionIds = [a.get("actionId") if isinstance(a, dict) else a.actionId for a in (actions or [])]
if actionRequest.actionId not in validActionIds:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Invalid action. Valid actions: {validActionIds}"
)
# Execute action based on notification type
actionResult = None
if notification.get("type") == NotificationType.INVITATION.value:
actionResult = await _handleInvitationAction(
notification=notification,
actionId=actionRequest.actionId,
currentUser=currentUser,
rootInterface=rootInterface
)
else:
# Generic action handling
actionResult = f"Action '{actionRequest.actionId}' executed"
# Update notification status
rootInterface.db.recordModify(
model_class=UserNotification,
recordId=notificationId,
record={
"status": NotificationStatus.ACTIONED.value,
"actionTaken": actionRequest.actionId,
"actionResult": actionResult,
"actionedAt": getUtcTimestamp()
}
)
return {
"message": actionResult,
"action": actionRequest.actionId,
"notificationId": notificationId
}
except HTTPException:
raise
except Exception as e:
logger.error(f"Error executing notification action: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to execute action: {str(e)}"
)
async def _handleInvitationAction(
notification: Dict[str, Any],
actionId: str,
currentUser: User,
rootInterface
) -> str:
"""Handle accept/decline actions for invitation notifications."""
from modules.datamodels.datamodelInvitation import Invitation
from modules.datamodels.datamodelUam import Mandate
from modules.datamodels.datamodelMembership import UserMandate
invitationId = notification.get("referenceId")
if not invitationId:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="No invitation reference found"
)
# Get the invitation
invitations = rootInterface.db.getRecordset(
model_class=Invitation,
recordFilter={"id": invitationId}
)
if not invitations:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Invitation not found"
)
invitation = invitations[0]
# Verify username matches
if invitation.get("targetUsername") != currentUser.username:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="This invitation is for a different user"
)
# Check if invitation is still valid
currentTime = getUtcTimestamp()
if invitation.get("expiresAt", 0) < currentTime:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invitation has expired"
)
if invitation.get("revokedAt"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invitation has been revoked"
)
if invitation.get("currentUses", 0) >= invitation.get("maxUses", 1):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invitation has reached maximum uses"
)
if actionId == "accept":
# Accept the invitation - assign roles and mandate access
mandateId = invitation.get("mandateId")
roleIds = invitation.get("roleIds", [])
# Get mandate name for result message
mandates = rootInterface.db.getRecordset(
model_class=Mandate,
recordFilter={"id": mandateId}
)
mandateName = mandates[0].get("mandateLabel", mandateId) if mandates else mandateId
# Check if user already has this mandate
existingMemberships = rootInterface.db.getRecordset(
model_class=UserMandate,
recordFilter={
"userId": currentUser.id,
"mandateId": mandateId
}
)
if existingMemberships:
# Update existing membership with new roles
existingMembership = existingMemberships[0]
existingRoles = existingMembership.get("roleIds", [])
mergedRoles = list(set(existingRoles + roleIds))
rootInterface.db.recordModify(
model_class=UserMandate,
recordId=existingMembership.get("id"),
record={"roleIds": mergedRoles}
)
logger.info(f"Updated UserMandate for user {currentUser.id} in mandate {mandateId}")
else:
# Create new user-mandate relationship
userMandate = UserMandate(
userId=currentUser.id,
mandateId=mandateId,
roleIds=roleIds
)
rootInterface.db.recordCreate(
model_class=UserMandate,
record=userMandate.model_dump()
)
logger.info(f"Created UserMandate for user {currentUser.id} in mandate {mandateId}")
# Mark invitation as used
rootInterface.db.recordModify(
model_class=Invitation,
recordId=invitationId,
record={
"usedBy": currentUser.id,
"usedAt": currentTime,
"currentUses": invitation.get("currentUses", 0) + 1
}
)
logger.info(f"User {currentUser.id} accepted invitation {invitationId} for mandate {mandateId}")
return f"Einladung angenommen. Sie haben jetzt Zugang zu '{mandateName}'."
elif actionId == "decline":
# Decline the invitation
# We don't revoke it, just mark the notification as declined
logger.info(f"User {currentUser.id} declined invitation {invitationId}")
return "Einladung abgelehnt."
else:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Unknown action: {actionId}"
)
@router.delete("/{notificationId}", response_model=Dict[str, Any])
@limiter.limit("30/minute")
async def deleteNotification(
request: Request,
notificationId: str,
currentUser: User = Depends(getCurrentUser)
) -> Dict[str, Any]:
"""
Delete/dismiss a notification.
"""
try:
rootInterface = getRootInterface()
# Get the notification
notifications = rootInterface.db.getRecordset(
model_class=UserNotification,
recordFilter={"id": notificationId}
)
if not notifications:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Notification not found"
)
notification = notifications[0]
# Verify ownership
if notification.get("userId") != currentUser.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not authorized to delete this notification"
)
# Mark as dismissed (soft delete)
rootInterface.db.recordModify(
model_class=UserNotification,
recordId=notificationId,
record={
"status": NotificationStatus.DISMISSED.value
}
)
return {"message": "Notification dismissed", "id": notificationId}
except HTTPException:
raise
except Exception as e:
logger.error(f"Error deleting notification: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to delete notification: {str(e)}"
)

47
modules/routes/routeSecurityLocal.py
View file

@ -20,6 +20,7 @@ from modules.interfaces.interfaceDbApp import getInterface, getRootInterface
from modules.datamodels.datamodelUam import User, UserInDB, AuthAuthority, Mandate
from modules.datamodels.datamodelSecurity import Token
from modules.shared.configuration import APP_CONFIG
from modules.shared.timeUtils import getUtcTimestamp
# Configure logger
logger = logging.getLogger(__name__)
@ -322,6 +323,52 @@ Falls Sie sich nicht registriert haben, können Sie diese E-Mail ignorieren."""
logger.error(f"Error sending registration email: {str(emailErr)}")
# Don't fail registration if email fails - user can request reset later
# Check for pending invitations and create notifications
try:
from modules.datamodels.datamodelInvitation import Invitation
from modules.routes.routeNotifications import createInvitationNotification
from modules.datamodels.datamodelUam import Mandate
currentTime = getUtcTimestamp()
pendingInvitations = appInterface.db.getRecordset(
model_class=Invitation,
recordFilter={"targetUsername": userData.username}
)
for invitation in pendingInvitations:
# Skip expired, revoked, or fully used invitations
if invitation.get("expiresAt", 0) < currentTime:
continue
if invitation.get("revokedAt"):
continue
if invitation.get("currentUses", 0) >= invitation.get("maxUses", 1):
continue
# Get mandate name for notification
mandateId = invitation.get("mandateId")
mandateRecords = appInterface.db.getRecordset(
Mandate,
recordFilter={"id": mandateId}
)
mandateName = mandateRecords[0].get("mandateLabel", "PowerOn") if mandateRecords else "PowerOn"
# Get inviter name
inviterId = invitation.get("createdBy")
inviter = appInterface.getUserById(inviterId) if inviterId else None
inviterName = (inviter.fullName or inviter.username) if inviter else "PowerOn"
createInvitationNotification(
userId=str(user.id),
invitationId=str(invitation.get("id")),
mandateName=mandateName,
inviterName=inviterName
)
logger.info(f"Created notification for new user {userData.username} for invitation {invitation.get('id')}")
except Exception as notifErr:
logger.warning(f"Failed to create notifications for pending invitations: {notifErr}")
# Don't fail registration if notification creation fails
return {
"message": "Registrierung erfolgreich! Bitte prüfen Sie Ihre E-Mail für den Link zum Setzen Ihres Passworts."
}

16
modules/system/registry.py
View file

@ -111,7 +111,7 @@ def loadFeatureMainModules() -> Dict[str, Any]:
def registerAllFeaturesInCatalog(catalogService) -> Dict[str, bool]:
"""
Register all features' RBAC objects in the catalog.
Also registers system-level RBAC objects.
Also registers system-level RBAC objects and feature definitions.
"""
results = {}
@ -132,6 +132,20 @@ def registerAllFeaturesInCatalog(catalogService) -> Dict[str, bool]:
mainModules = loadFeatureMainModules()
for featureName, module in mainModules.items():
# Register feature definition in catalog (for /api/features/ endpoint)
if hasattr(module, "getFeatureDefinition"):
try:
featureDef = module.getFeatureDefinition()
catalogService.registerFeatureDefinition(
featureCode=featureDef.get("code", featureName),
label=featureDef.get("label", {"en": featureName, "de": featureName}),
icon=featureDef.get("icon", "mdi-puzzle")
)
logger.info(f"Registered feature definition: {featureDef.get('code', featureName)}")
except Exception as e:
logger.error(f"Error registering feature definition for {featureName}: {e}")
# Register RBAC objects (UI, RESOURCE, DATA)
if hasattr(module, "registerFeature"):
try:
success = module.registerFeature(catalogService)