PowerOn/platform-core
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Pydantic FK als Single Source of Truth
All checks were successful
Deploy Plattform-Core / test (push) Successful in 44s
Details
Deploy Plattform-Core / deploy (push) Successful in 4s
Details

Browse source
This commit is contained in:
ValueOn AG 2026-05-25 15:14:05 +02:00
parent c2443a7781
commit 9719a22581
8 changed files with 728 additions and 89 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
modules/datamodels/datamodelKnowledge.py
View file

@ -98,7 +98,7 @@ class FileContentIndex(PowerOnModel):
connectionId: Optional[str] = Field(
default=None,
description="UserConnection ID if this index entry originates from an external connector",
json_schema_extra={"label": "Connection-ID"},
json_schema_extra={"label": "Connection-ID", "fk_target": {"db": "poweron_app", "table": "UserConnection", "labelField": "authority"}},
)
neutralizationStatus: Optional[str] = Field(
default=None,
@ -122,7 +122,7 @@ class ContentChunk(PowerOnModel):
)
contentObjectId: str = Field(
description="Reference to the content object within FileContentIndex",
json_schema_extra={"label": "Inhaltsobjekt-ID"},
json_schema_extra={"label": "Inhaltsobjekt-ID", "fk_target": {"db": "poweron_knowledge", "table": "FileContentIndex", "labelField": "fileName"}},
)
fileId: str = Field(
description="FK to the source file",
@ -177,7 +177,7 @@ class RoundMemory(PowerOnModel):
)
workflowId: str = Field(
description="FK to the workflow",
json_schema_extra={"label": "Workflow-ID"},
json_schema_extra={"label": "Workflow-ID", "fk_target": {"db": "poweron_chat", "table": "ChatWorkflow", "labelField": "name"}},
)
roundNumber: int = Field(
default=0,

160
modules/features/commcoach/datamodelCommcoach.py
View file

@ -74,9 +74,18 @@ class CoachingScoreTrend(str, Enum):
class TrainingModule(PowerOnModel):
"""A training module representing a topic the user is working on."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
userId: str = Field(description="Owner user ID (strict ownership)")
mandateId: str = Field(description="Mandate ID")
instanceId: str = Field(description="Feature instance ID")
userId: str = Field(
description="Owner user ID (strict ownership)",
json_schema_extra={"label": "Benutzer", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}},
)
mandateId: str = Field(
description="Mandate ID",
json_schema_extra={"label": "Mandant", "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"}},
)
instanceId: str = Field(
description="Feature instance ID",
json_schema_extra={"label": "Feature-Instanz", "fk_target": {"db": "poweron_app", "table": "FeatureInstance", "labelField": "label"}},
)
title: str = Field(description="Module title, e.g. 'Conflict with team lead'")
description: Optional[str] = Field(default=None, description="Short description")
moduleType: TrainingModuleType = Field(default=TrainingModuleType.COACHING)
@ -84,7 +93,10 @@ class TrainingModule(PowerOnModel):
goals: Optional[str] = Field(default=None, description="Free-text goal description")
insights: Optional[str] = Field(default=None, description="JSON array of AI insights [{id, text, sessionId, createdAt}]")
metadata: Optional[str] = Field(default=None, description="JSON object with flexible metadata")
personaId: Optional[str] = Field(default=None, description="Default persona for sessions")
personaId: Optional[str] = Field(
default=None, description="Default persona for sessions",
json_schema_extra={"label": "Persona", "fk_target": {"db": "poweron_commcoach", "table": "CoachingPersona", "labelField": "label"}},
)
kpiTargets: Optional[str] = Field(default=None, description="JSON object with structured KPI targets")
sessionCount: int = Field(default=0)
taskCount: int = Field(default=0)
@ -96,12 +108,27 @@ class TrainingModule(PowerOnModel):
class CoachingSession(PowerOnModel):
"""A single coaching conversation session within a module."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
moduleId: str = Field(description="FK to TrainingModule")
userId: str = Field(description="Owner user ID")
mandateId: str = Field(description="Mandate ID")
instanceId: str = Field(description="Feature instance ID")
moduleId: str = Field(
description="FK to TrainingModule",
json_schema_extra={"label": "Trainingsmodul", "fk_target": {"db": "poweron_commcoach", "table": "TrainingModule", "labelField": "title"}},
)
userId: str = Field(
description="Owner user ID",
json_schema_extra={"label": "Benutzer", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}},
)
mandateId: str = Field(
description="Mandate ID",
json_schema_extra={"label": "Mandant", "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"}},
)
instanceId: str = Field(
description="Feature instance ID",
json_schema_extra={"label": "Feature-Instanz", "fk_target": {"db": "poweron_app", "table": "FeatureInstance", "labelField": "label"}},
)
status: CoachingSessionStatus = Field(default=CoachingSessionStatus.ACTIVE)
personaId: Optional[str] = Field(default=None, description="FK to CoachingPersona (Iteration 2)")
personaId: Optional[str] = Field(
default=None, description="FK to CoachingPersona",
json_schema_extra={"label": "Persona", "fk_target": {"db": "poweron_commcoach", "table": "CoachingPersona", "labelField": "label"}},
)
summary: Optional[str] = Field(default=None, description="AI-generated session summary")
coachNotes: Optional[str] = Field(default=None, description="JSON: AI internal notes for continuity")
compressedHistorySummary: Optional[str] = Field(default=None, description="AI summary of older messages for long sessions")
@ -118,9 +145,18 @@ class CoachingSession(PowerOnModel):
class CoachingMessage(PowerOnModel):
"""A single message in a coaching session."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
sessionId: str = Field(description="FK to CoachingSession")
moduleId: str = Field(description="FK to TrainingModule")
userId: str = Field(description="Owner user ID")
sessionId: str = Field(
description="FK to CoachingSession",
json_schema_extra={"label": "Session", "fk_target": {"db": "poweron_commcoach", "table": "CoachingSession", "labelField": None}},
)
moduleId: str = Field(
description="FK to TrainingModule",
json_schema_extra={"label": "Trainingsmodul", "fk_target": {"db": "poweron_commcoach", "table": "TrainingModule", "labelField": "title"}},
)
userId: str = Field(
description="Owner user ID",
json_schema_extra={"label": "Benutzer", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}},
)
role: CoachingMessageRole = Field(description="Message author role")
content: str = Field(description="Message content (Markdown)")
contentType: CoachingMessageContentType = Field(default=CoachingMessageContentType.TEXT)
@ -131,10 +167,22 @@ class CoachingMessage(PowerOnModel):
class CoachingTask(PowerOnModel):
"""A task/checklist item assigned within a training module."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
moduleId: str = Field(description="FK to TrainingModule")
sessionId: Optional[str] = Field(default=None, description="FK to originating session")
userId: str = Field(description="Owner user ID")
mandateId: str = Field(description="Mandate ID")
moduleId: str = Field(
description="FK to TrainingModule",
json_schema_extra={"label": "Trainingsmodul", "fk_target": {"db": "poweron_commcoach", "table": "TrainingModule", "labelField": "title"}},
)
sessionId: Optional[str] = Field(
default=None, description="FK to originating session",
json_schema_extra={"label": "Session", "fk_target": {"db": "poweron_commcoach", "table": "CoachingSession", "labelField": None}},
)
userId: str = Field(
description="Owner user ID",
json_schema_extra={"label": "Benutzer", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}},
)
mandateId: str = Field(
description="Mandate ID",
json_schema_extra={"label": "Mandant", "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"}},
)
title: str = Field(description="Task title")
description: Optional[str] = Field(default=None)
status: CoachingTaskStatus = Field(default=CoachingTaskStatus.OPEN)
@ -146,10 +194,22 @@ class CoachingTask(PowerOnModel):
class CoachingScore(PowerOnModel):
"""A competence score for a dimension, recorded after a session."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
moduleId: str = Field(description="FK to TrainingModule")
sessionId: str = Field(description="FK to CoachingSession")
userId: str = Field(description="Owner user ID")
mandateId: str = Field(description="Mandate ID")
moduleId: str = Field(
description="FK to TrainingModule",
json_schema_extra={"label": "Trainingsmodul", "fk_target": {"db": "poweron_commcoach", "table": "TrainingModule", "labelField": "title"}},
)
sessionId: str = Field(
description="FK to CoachingSession",
json_schema_extra={"label": "Session", "fk_target": {"db": "poweron_commcoach", "table": "CoachingSession", "labelField": None}},
)
userId: str = Field(
description="Owner user ID",
json_schema_extra={"label": "Benutzer", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}},
)
mandateId: str = Field(
description="Mandate ID",
json_schema_extra={"label": "Mandant", "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"}},
)
dimension: str = Field(description="e.g. empathy, clarity, assertiveness, listening")
score: float = Field(ge=0.0, le=100.0)
trend: CoachingScoreTrend = Field(default=CoachingScoreTrend.STABLE)
@ -159,9 +219,18 @@ class CoachingScore(PowerOnModel):
class CoachingUserProfile(PowerOnModel):
"""Per-user coaching profile and preferences."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
userId: str = Field(description="Owner user ID")
mandateId: str = Field(description="Mandate ID")
instanceId: str = Field(description="Feature instance ID")
userId: str = Field(
description="Owner user ID",
json_schema_extra={"label": "Benutzer", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}},
)
mandateId: str = Field(
description="Mandate ID",
json_schema_extra={"label": "Mandant", "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"}},
)
instanceId: str = Field(
description="Feature instance ID",
json_schema_extra={"label": "Feature-Instanz", "fk_target": {"db": "poweron_app", "table": "FeatureInstance", "labelField": "label"}},
)
dailyReminderTime: Optional[str] = Field(default=None, description="HH:MM format")
dailyReminderEnabled: bool = Field(default=False)
emailSummaryEnabled: bool = Field(default=True)
@ -179,9 +248,18 @@ class CoachingUserProfile(PowerOnModel):
class CoachingPersona(PowerOnModel):
"""A roleplay persona for coaching sessions."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
userId: str = Field(description="Owner user ID ('system' for builtins)")
mandateId: Optional[str] = Field(default=None)
instanceId: Optional[str] = Field(default=None)
userId: str = Field(
description="Owner user ID ('system' for builtins)",
json_schema_extra={"label": "Benutzer", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username", "softFk": True}},
)
mandateId: Optional[str] = Field(
default=None,
json_schema_extra={"label": "Mandant", "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"}},
)
instanceId: Optional[str] = Field(
default=None,
json_schema_extra={"label": "Feature-Instanz", "fk_target": {"db": "poweron_app", "table": "FeatureInstance", "labelField": "label"}},
)
key: str = Field(description="Unique key, e.g. 'critical_cfo_f'")
label: str = Field(description="Display label, e.g. 'Kritische CFO'")
description: str = Field(description="Detailed role description for the AI")
@ -198,9 +276,18 @@ class CoachingPersona(PowerOnModel):
class ModulePersonaMapping(PowerOnModel):
"""Maps which personas are available for a specific training module."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
moduleId: str = Field(description="FK to TrainingModule")
personaId: str = Field(description="FK to CoachingPersona")
instanceId: str = Field(description="Feature instance ID")
moduleId: str = Field(
description="FK to TrainingModule",
json_schema_extra={"label": "Trainingsmodul", "fk_target": {"db": "poweron_commcoach", "table": "TrainingModule", "labelField": "title"}},
)
personaId: str = Field(
description="FK to CoachingPersona",
json_schema_extra={"label": "Persona", "fk_target": {"db": "poweron_commcoach", "table": "CoachingPersona", "labelField": "label"}},
)
instanceId: str = Field(
description="Feature instance ID",
json_schema_extra={"label": "Feature-Instanz", "fk_target": {"db": "poweron_app", "table": "FeatureInstance", "labelField": "label"}},
)
class SetModulePersonasRequest(BaseModel):
@ -214,9 +301,18 @@ class SetModulePersonasRequest(BaseModel):
class CoachingBadge(PowerOnModel):
"""An achievement badge awarded to a user."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
userId: str = Field(description="Owner user ID")
mandateId: str = Field(description="Mandate ID")
instanceId: str = Field(description="Feature instance ID")
userId: str = Field(
description="Owner user ID",
json_schema_extra={"label": "Benutzer", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}},
)
mandateId: str = Field(
description="Mandate ID",
json_schema_extra={"label": "Mandant", "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"}},
)
instanceId: str = Field(
description="Feature instance ID",
json_schema_extra={"label": "Feature-Instanz", "fk_target": {"db": "poweron_app", "table": "FeatureInstance", "labelField": "label"}},
)
badgeKey: str = Field(description="Badge identifier, e.g. 'streak_7'")
awardedAt: Optional[float] = Field(default=None, json_schema_extra={"frontend_type": "timestamp"})

16
modules/features/realEstate/datamodelFeatureRealEstate.py
View file

@ -265,15 +265,19 @@ class Kanton(PowerOnModel):
)
mandateId: str = Field(
description="ID of the mandate",
frontend_type="text",
frontend_readonly=True,
frontend_required=False,
json_schema_extra={
"frontend_type": "text", "frontend_readonly": True, "frontend_required": False,
"label": "Mandant",
"fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"},
},
)
featureInstanceId: str = Field(
description="ID of the feature instance",
frontend_type="text",
frontend_readonly=True,
frontend_required=False,
json_schema_extra={
"frontend_type": "text", "frontend_readonly": True, "frontend_required": False,
"label": "Feature-Instanz",
"fk_target": {"db": "poweron_app", "table": "FeatureInstance", "labelField": "label"},
},
)
label: str = Field(
description="Canton name (e.g. 'Zürich')",

104
modules/features/teamsbot/datamodelTeamsbot.py
View file

@ -102,12 +102,24 @@ class TeamsbotModuleStatus(str, Enum):
class TeamsbotMeetingModule(PowerOnModel):
"""A meeting module groups related sessions (e.g. 'Weekly Standup')."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Module ID")
instanceId: str = Field(description="Feature instance ID (FK)")
mandateId: str = Field(description="Mandate ID (FK)")
ownerUserId: str = Field(description="Owner user ID")
instanceId: str = Field(
description="Feature instance ID",
json_schema_extra={"label": "Feature-Instanz", "fk_target": {"db": "poweron_app", "table": "FeatureInstance", "labelField": "label"}},
)
mandateId: str = Field(
description="Mandate ID",
json_schema_extra={"label": "Mandant", "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"}},
)
ownerUserId: str = Field(
description="Owner user ID",
json_schema_extra={"label": "Besitzer", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}},
)
title: str = Field(description="Module title, e.g. 'Weekly Standup'")
seriesType: TeamsbotSeriesType = Field(default=TeamsbotSeriesType.ADHOC)
defaultBotId: Optional[str] = Field(default=None, description="FK to TeamsbotSystemBot")
defaultBotId: Optional[str] = Field(
default=None, description="FK to TeamsbotSystemBot",
json_schema_extra={"label": "Standard-Bot", "fk_target": {"db": "poweron_teamsbot", "table": "TeamsbotSystemBot", "labelField": "name"}},
)
defaultDirectorPrompts: Optional[str] = Field(default=None, description="JSON list of default director prompts")
goals: Optional[str] = Field(default=None, description="Free-text goals")
kpiTargets: Optional[str] = Field(default=None, description="JSON object with structured KPI targets")
@ -120,8 +132,8 @@ class TeamsbotMeetingModule(PowerOnModel):
description="Default display name for the bot when starting a session from this module",
)
defaultAvatarFileId: Optional[str] = Field(
default=None,
description="FileItem ID for the default avatar image/video shown in the meeting",
default=None, description="FileItem ID for the default avatar image/video shown in the meeting",
json_schema_extra={"label": "Avatar-Datei", "fk_target": {"db": "poweron_management", "table": "FileItem", "labelField": "fileName"}},
)
status: TeamsbotModuleStatus = Field(default=TeamsbotModuleStatus.ACTIVE)
@ -129,15 +141,27 @@ class TeamsbotMeetingModule(PowerOnModel):
class TeamsbotSession(PowerOnModel):
"""A Teams Bot meeting session."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Session ID")
instanceId: str = Field(description="Feature instance ID (FK)")
mandateId: str = Field(description="Mandate ID (FK)")
moduleId: Optional[str] = Field(default=None, description="FK to TeamsbotMeetingModule (nullable during transition)")
instanceId: str = Field(
description="Feature instance ID",
json_schema_extra={"label": "Feature-Instanz", "fk_target": {"db": "poweron_app", "table": "FeatureInstance", "labelField": "label"}},
)
mandateId: str = Field(
description="Mandate ID",
json_schema_extra={"label": "Mandant", "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"}},
)
moduleId: Optional[str] = Field(
default=None, description="FK to TeamsbotMeetingModule",
json_schema_extra={"label": "Meeting-Modul", "fk_target": {"db": "poweron_teamsbot", "table": "TeamsbotMeetingModule", "labelField": "title"}},
)
meetingLink: str = Field(description="Teams meeting join link")
botName: str = Field(default="AI Assistant", description="Display name of the bot in the meeting")
status: TeamsbotSessionStatus = Field(default=TeamsbotSessionStatus.PENDING, description="Current session status")
startedAt: Optional[float] = Field(default=None, description="UTC unix timestamp when session started", json_schema_extra={"frontend_type": "timestamp"})
endedAt: Optional[float] = Field(default=None, description="UTC unix timestamp when session ended", json_schema_extra={"frontend_type": "timestamp"})
startedByUserId: str = Field(description="User ID who started the session")
startedByUserId: str = Field(
description="User ID who started the session",
json_schema_extra={"label": "Gestartet von", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}},
)
bridgeSessionId: Optional[str] = Field(default=None, description="Session ID on the .NET Media Bridge")
meetingChatId: Optional[str] = Field(default=None, description="Teams meeting chat ID for Graph API messages")
sessionContext: Optional[str] = Field(default=None, description="Custom context/knowledge provided by the user for this session")
@ -150,7 +174,10 @@ class TeamsbotSession(PowerOnModel):
class TeamsbotTranscript(PowerOnModel):
"""A single transcript segment from the meeting."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Transcript segment ID")
sessionId: str = Field(description="Session ID (FK)")
sessionId: str = Field(
description="FK to TeamsbotSession",
json_schema_extra={"label": "Session", "fk_target": {"db": "poweron_teamsbot", "table": "TeamsbotSession", "labelField": "botName"}},
)
speaker: Optional[str] = Field(default=None, description="Speaker name or identifier")
text: str = Field(description="Transcribed text")
timestamp: float = Field(description="UTC unix timestamp of the speech segment", json_schema_extra={"frontend_type": "timestamp"})
@ -163,12 +190,18 @@ class TeamsbotTranscript(PowerOnModel):
class TeamsbotBotResponse(PowerOnModel):
"""A bot response generated during a meeting session."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Response ID")
sessionId: str = Field(description="Session ID (FK)")
sessionId: str = Field(
description="FK to TeamsbotSession",
json_schema_extra={"label": "Session", "fk_target": {"db": "poweron_teamsbot", "table": "TeamsbotSession", "labelField": "botName"}},
)
responseText: str = Field(description="The bot's response text")
responseType: TeamsbotResponseType = Field(default=TeamsbotResponseType.BOTH, description="How the response was delivered")
detectedIntent: TeamsbotDetectedIntent = Field(default=TeamsbotDetectedIntent.NONE, description="What triggered the response")
reasoning: Optional[str] = Field(default=None, description="AI reasoning for why it responded")
triggeredByTranscriptId: Optional[str] = Field(default=None, description="Transcript segment that triggered this response")
triggeredByTranscriptId: Optional[str] = Field(
default=None, description="Transcript segment that triggered this response",
json_schema_extra={"label": "Ausgelöst durch", "fk_target": {"db": "poweron_teamsbot", "table": "TeamsbotTranscript", "labelField": None}},
)
modelName: Optional[str] = Field(default=None, description="AI model used for this response")
processingTime: float = Field(default=0.0, description="Processing time in seconds")
priceCHF: float = Field(default=0.0, description="Cost of this AI call in CHF")
@ -184,7 +217,10 @@ class TeamsbotSystemBot(PowerOnModel):
Credentials are stored encrypted in the database, NOT in the UI-visible config.
Only mandate admins can manage system bots."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="System bot ID")
mandateId: str = Field(description="Mandate ID (FK) - bots are scoped to mandates")
mandateId: str = Field(
description="Mandate ID - bots are scoped to mandates",
json_schema_extra={"label": "Mandant", "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"}},
)
name: str = Field(description="Display name (e.g. 'Nyla Larsson')")
email: str = Field(description="Microsoft account email")
encryptedPassword: str = Field(description="Encrypted Microsoft account password")
@ -200,8 +236,14 @@ class TeamsbotUserAccount(PowerOnModel):
Each user can store their own MS credentials per mandate.
Password is encrypted; on login only MFA confirmation is needed."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Record ID")
userId: str = Field(description="Poweron user ID (FK)")
mandateId: str = Field(description="Mandate ID (FK)")
userId: str = Field(
description="Poweron user ID",
json_schema_extra={"label": "Benutzer", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}},
)
mandateId: str = Field(
description="Mandate ID",
json_schema_extra={"label": "Mandant", "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"}},
)
email: str = Field(description="Microsoft account email")
encryptedPassword: str = Field(description="Encrypted Microsoft account password")
displayName: Optional[str] = Field(default=None, description="Display name derived from MS account")
@ -216,8 +258,14 @@ class TeamsbotUserSettings(PowerOnModel):
Each user has their own settings per feature instance.
These override the instance-level defaults (TeamsbotConfig)."""
id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Settings ID")
userId: str = Field(description="User ID (FK)")
instanceId: str = Field(description="Feature instance ID (FK)")
userId: str = Field(
description="User ID",
json_schema_extra={"label": "Benutzer", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}},
)
instanceId: str = Field(
description="Feature instance ID",
json_schema_extra={"label": "Feature-Instanz", "fk_target": {"db": "poweron_app", "table": "FeatureInstance", "labelField": "label"}},
)
botName: Optional[str] = Field(default=None, description="Bot display name override")
aiSystemPrompt: Optional[str] = Field(default=None, description="AI system prompt override")
responseMode: Optional[str] = Field(default=None, description="Response mode override: auto, manual, transcribeOnly")
@ -229,7 +277,10 @@ class TeamsbotUserSettings(PowerOnModel):
triggerCooldownSeconds: Optional[int] = Field(default=None, description="Trigger cooldown override")
contextWindowSegments: Optional[int] = Field(default=None, description="Context window override")
debugMode: Optional[bool] = Field(default=None, description="Debug mode override")
avatarFileId: Optional[str] = Field(default=None, description="FileItem ID for bot avatar image/video override")
avatarFileId: Optional[str] = Field(
default=None, description="FileItem ID for bot avatar image/video override",
json_schema_extra={"label": "Avatar-Datei", "fk_target": {"db": "poweron_management", "table": "FileItem", "labelField": "fileName"}},
)
# ============================================================================
@ -382,9 +433,18 @@ class TeamsbotDirectorPrompt(PowerOnModel):
meeting participants.
"""
id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="Director prompt ID")
sessionId: str = Field(description="Teams Bot session ID (FK)")
instanceId: str = Field(description="Feature instance ID (FK)")
operatorUserId: str = Field(description="User ID of the operator who issued the prompt")
sessionId: str = Field(
description="FK to TeamsbotSession",
json_schema_extra={"label": "Session", "fk_target": {"db": "poweron_teamsbot", "table": "TeamsbotSession", "labelField": "botName"}},
)
instanceId: str = Field(
description="Feature instance ID",
json_schema_extra={"label": "Feature-Instanz", "fk_target": {"db": "poweron_app", "table": "FeatureInstance", "labelField": "label"}},
)
operatorUserId: str = Field(
description="User ID of the operator who issued the prompt",
json_schema_extra={"label": "Operator", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}},
)
text: str = Field(description="The director instruction text", max_length=DIRECTOR_PROMPT_TEXT_LIMIT)
mode: TeamsbotDirectorPromptMode = Field(default=TeamsbotDirectorPromptMode.ONE_SHOT, description="oneShot or persistent")
fileIds: List[str] = Field(default_factory=list, description="UDB-selected file/object IDs to attach as RAG context")

61
modules/routes/routeAdminDatabaseHealth.py
View file

@ -20,6 +20,8 @@ from modules.system.databaseHealth import (
OrphanCleanupRefused,
_cleanAllOrphans,
_cleanOrphans,
_discoverLegacyTables,
_dropLegacyTable,
_getTableStats,
_isUserIdFk,
_listOrphans,
@ -209,6 +211,65 @@ def postDatabaseOrphansCleanAll(
return {"results": results, "skipped": skipped, "errored": errored, "deleted": deletedTotal}
# ---------------------------------------------------------------------------
# Legacy Tables (tables without Pydantic model)
# ---------------------------------------------------------------------------
class LegacyTableDropRequest(BaseModel):
"""Body for dropping a legacy table."""
db: str = Field(..., description="Database name")
table: str = Field(..., description="Table name to drop")
@router.get("/legacy-tables")
@limiter.limit("10/minute")
def getLegacyTables(
request: Request,
db: Optional[str] = None,
currentUser: User = Depends(requireSysAdmin),
) -> Dict[str, Any]:
"""List tables that exist in the database but have no Pydantic model.
Optional ``db`` filter to scope to a single database.
"""
tables = _discoverLegacyTables(dbFilter=db)
totalRows = sum(t["rowCount"] for t in tables)
totalSize = sum(t["sizeBytes"] for t in tables)
return {
"legacyTables": tables,
"totalCount": len(tables),
"totalRows": totalRows,
"totalSizeBytes": totalSize,
}
@router.post("/legacy-tables/drop")
@limiter.limit("10/minute")
def postLegacyTableDrop(
request: Request,
body: LegacyTableDropRequest,
currentUser: User = Depends(requireSysAdmin),
) -> Dict[str, Any]:
"""Drop a legacy table (CASCADE). Refuses if the table is model-backed."""
try:
result = _dropLegacyTable(body.db, body.table)
except ValueError as e:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(e),
) from e
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Drop failed: {e}",
) from e
logger.info(
"SysAdmin legacy-table drop: user=%s db=%s table=%s rows=%s",
currentUser.username, body.db, body.table, result.get("rowCount"),
)
return result
# ---------------------------------------------------------------------------
# Migration (Backup / Restore)
# ---------------------------------------------------------------------------

99
modules/system/databaseHealth.py
View file

@ -790,3 +790,102 @@ def _jsonSafe(v):
except Exception:
return repr(v)
return str(v)
# ---------------------------------------------------------------------------
# Legacy table discovery + drop
# ---------------------------------------------------------------------------
def _discoverLegacyTables(dbFilter: Optional[str] = None) -> List[dict]:
"""Find tables that exist in the DB but have no entry in MODEL_REGISTRY.
Returns a list of dicts: {db, table, rowCount, sizeBytes}.
"""
from modules.datamodels.datamodelBase import MODEL_REGISTRY
from modules.shared.fkRegistry import _buildTableToDbMap
tableToDb = _buildTableToDbMap()
registeredDbs = getRegisteredDatabases()
results: List[dict] = []
for dbName in sorted(registeredDbs.keys()):
if dbFilter and dbName != dbFilter:
continue
try:
conn = _getConnection(dbName)
except Exception as e:
logger.warning("Legacy scan: cannot connect to %s: %s", dbName, e)
continue
try:
with conn.cursor() as cur:
cur.execute("""
SELECT c.relname AS table_name,
c.reltuples::bigint AS row_estimate,
pg_total_relation_size(c.oid) AS size_bytes
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname = 'public'
AND c.relkind = 'r'
AND c.relname NOT LIKE '\\_%'
ORDER BY c.relname
""")
for row in cur.fetchall():
tblName = row["table_name"]
inRegistry = (
tblName in MODEL_REGISTRY
and tableToDb.get(tblName) == dbName
)
if not inRegistry:
results.append({
"db": dbName,
"table": tblName,
"rowCount": max(0, int(row["row_estimate"])),
"sizeBytes": int(row["size_bytes"]),
})
finally:
conn.close()
return results
def _dropLegacyTable(dbName: str, tableName: str) -> dict:
"""Drop a single legacy table after verifying it is NOT in MODEL_REGISTRY.
Returns {db, table, dropped, rowCount}.
Raises ValueError if the table is model-backed (safety guard).
"""
from modules.datamodels.datamodelBase import MODEL_REGISTRY
from modules.shared.fkRegistry import _buildTableToDbMap
tableToDb = _buildTableToDbMap()
inRegistry = (
tableName in MODEL_REGISTRY
and tableToDb.get(tableName) == dbName
)
if inRegistry:
raise ValueError(
f"Table '{dbName}.{tableName}' is backed by a Pydantic model and cannot be dropped via legacy cleanup."
)
conn = _getConnection(dbName)
try:
with conn.cursor() as cur:
cur.execute("""
SELECT reltuples::bigint AS row_estimate
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname = 'public' AND c.relname = %s
""", (tableName,))
row = cur.fetchone()
rowCount = max(0, int(row["row_estimate"])) if row else 0
cur.execute(f'DROP TABLE IF EXISTS "{tableName}" CASCADE')
conn.commit()
logger.info("Dropped legacy table %s.%s (%d rows)", dbName, tableName, rowCount)
return {"db": dbName, "table": tableName, "dropped": True, "rowCount": rowCount}
except Exception as e:
conn.rollback()
logger.error("Failed to drop legacy table %s.%s: %s", dbName, tableName, e)
raise
finally:
conn.close()

63
modules/system/databaseMigration.py
View file

@ -21,6 +21,8 @@ import psycopg2.extras
from modules.shared.configuration import APP_CONFIG
from modules.shared.dbRegistry import getRegisteredDatabases
from modules.shared.fkRegistry import _buildTableToDbMap, getFkRelationships
from modules.datamodels.datamodelBase import MODEL_REGISTRY
from modules.system.databaseHealth import _getConnection, _jsonSafe
logger = logging.getLogger(__name__)
@ -116,13 +118,33 @@ def _exportDatabases(databases: List[str]) -> dict:
return exportData
def _getModelTablesForDb(dbName: str, physicalTables: List[str]) -> List[str]:
"""Return only those physical tables that have a matching Pydantic model
registered in MODEL_REGISTRY and mapped to *dbName*.
Tables without a Pydantic class (legacy / orphan tables) are excluded
from export so the backup contains only model-backed data.
"""
tableToDb = _buildTableToDbMap()
return sorted(
t for t in physicalTables
if t in MODEL_REGISTRY and tableToDb.get(t) == dbName
)
def _exportSingleDb(dbName: str) -> dict:
conn = _getConnection(dbName)
excluded = _EXCLUDED_TABLES.get(dbName, set())
try:
tables = _listTables(conn)
allTables = _listTables(conn)
modelTables = _getModelTablesForDb(dbName, allTables)
skippedLegacy = set(allTables) - set(modelTables) - excluded - {_SYSTEM_TABLE}
if skippedLegacy:
logger.info("Export %s: skipping %d legacy tables without model: %s",
dbName, len(skippedLegacy), sorted(skippedLegacy))
dbPayload: dict = {"tables": {}, "summary": {}, "tableCount": 0, "totalRecords": 0}
for tbl in tables:
for tbl in modelTables:
if tbl in excluded:
logger.info("Export: skipping excluded table %s.%s", dbName, tbl)
continue
@ -635,35 +657,24 @@ def _createTableFromExport(conn, tableName: str, rows: List[dict]) -> None:
logger.info("Created table %s with %d columns", tableName, len(allKeys))
def _getTableImportOrder(conn, tableNames: List[str]) -> List[str]:
def _getTableImportOrder(conn, tableNames: List[str], dbName: str = "") -> List[str]:
"""Sort tables by FK dependencies (parents first) using topological sort.
Queries ``information_schema`` for FK relationships, builds a dependency
graph, and returns the tables in an order that satisfies referential
integrity: parent tables before child tables.
Uses Pydantic ``fk_target`` metadata from ``fkRegistry`` as the single
source of truth (works for ALL databases, not just those with SQL FKs).
Only *intra-DB* dependencies are considered; cross-DB FKs (e.g. to
``poweron_app.Mandate``) are handled by importing databases in order.
"""
tableSet = set(tableNames)
with conn.cursor() as cur:
cur.execute("""
SELECT DISTINCT
tc.table_name AS child_table,
ccu.table_name AS parent_table
FROM information_schema.table_constraints tc
JOIN information_schema.constraint_column_usage ccu
ON ccu.constraint_name = tc.constraint_name
AND ccu.table_schema = tc.table_schema
WHERE tc.constraint_type = 'FOREIGN KEY'
AND tc.table_schema = 'public'
AND tc.table_name != ccu.table_name
""")
fks = cur.fetchall()
allRels = getFkRelationships()
deps: Dict[str, Set[str]] = {t: set() for t in tableNames}
for fk in fks:
child = fk["child_table"]
parent = fk["parent_table"]
if child in tableSet and parent in tableSet:
for rel in allRels:
if rel.sourceDb != dbName or rel.targetDb != dbName:
continue
child = rel.sourceTable
parent = rel.targetTable
if child in tableSet and parent in tableSet and child != parent:
deps[child].add(parent)
inDegree = {t: len(deps[t]) for t in tableNames}
@ -746,7 +757,7 @@ def _importSingleDb(payload: dict, dbName: str, mode: str, protectedIds: List[st
if t not in excluded
and isinstance(tables.get(t), list)
and t in existingTables]
importOrder = _getTableImportOrder(conn, importable)
importOrder = _getTableImportOrder(conn, importable, dbName)
logger.info("Import order for %s: %s", dbName, importOrder)

308
scripts/exportDbSchemaFromModels.py Normal file
View file

@ -0,0 +1,308 @@
"""Export the database schema from Pydantic MODEL_REGISTRY + fk_target metadata.
Usage (run from gateway/):
python scripts/exportDbSchemaFromModels.py
python scripts/exportDbSchemaFromModels.py --validate
python scripts/exportDbSchemaFromModels.py --output ../wiki/b-reference/database-schema.md
The Pydantic classes are the single source of truth. The optional --validate
flag cross-checks against the live database and reports mismatches.
"""
import argparse
import importlib
import os
import sys
from collections import defaultdict
from datetime import datetime
from typing import Dict, List, Optional, Tuple
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
def _getArgs():
p = argparse.ArgumentParser(description="Export DB schema from Pydantic models")
p.add_argument("--output", default="../wiki/b-reference/database-schema.md")
p.add_argument("--validate", action="store_true",
help="Cross-check against live DB and report mismatches")
return p.parse_args()
def _loadAllModels():
"""Import all datamodel and interface modules to populate MODEL_REGISTRY + dbRegistry."""
for root, _dirs, files in os.walk("modules"):
for f in files:
if not f.endswith(".py") or f.startswith("__"):
continue
isDatamodel = f.startswith("datamodel")
isInterface = f.startswith("interface") and ("Db" in f or "Feature" in f)
if not isDatamodel and not isInterface:
continue
modPath = os.path.join(root, f).replace(os.sep, ".").replace(".py", "")
try:
importlib.import_module(modPath)
except Exception:
pass
def _buildCompleteTableToDbMap() -> Dict[str, str]:
"""Build tableName -> dbName by querying every registered DB's catalog.
More reliable than fkRegistry._buildTableToDbMap() for the schema script
because it catches ALL tables, not just FK targets.
"""
from modules.shared.dbRegistry import getRegisteredDatabases
from modules.system.databaseHealth import _getConnection
mapping: Dict[str, str] = {}
for dbName in getRegisteredDatabases():
try:
conn = _getConnection(dbName)
try:
with conn.cursor() as cur:
cur.execute("""
SELECT table_name FROM information_schema.tables
WHERE table_schema = 'public' AND table_type = 'BASE TABLE'
AND table_name NOT LIKE '\\_%'
""")
for row in cur.fetchall():
tbl = row["table_name"] if isinstance(row, dict) else row[0]
if tbl not in mapping:
mapping[tbl] = dbName
finally:
conn.close()
except Exception as e:
print(f" Warning: could not query {dbName}: {e}")
return mapping
def _buildSchema() -> Tuple[Dict[str, List[dict]], Dict[str, str]]:
"""Build {dbName: [tableInfo, ...]} from MODEL_REGISTRY + fk_target.
Returns (schema, tableToDb).
"""
from modules.datamodels.datamodelBase import MODEL_REGISTRY
tableToDb = _buildCompleteTableToDbMap()
schema: Dict[str, List[dict]] = defaultdict(list)
for tableName, modelCls in sorted(MODEL_REGISTRY.items()):
dbName = tableToDb.get(tableName)
if not dbName:
continue
fields = []
fkRefs = []
pkField = None
for fieldName, fieldInfo in modelCls.model_fields.items():
annotation = modelCls.__annotations__.get(fieldName)
typeName = _resolveTypeName(annotation)
isOptional = typeName.startswith("Optional[")
extra = fieldInfo.json_schema_extra or {}
fkTarget = extra.get("fk_target")
if fieldName == "id":
pkField = {"name": fieldName, "type": typeName}
continue
if fkTarget:
fkRefs.append({
"column": fieldName,
"targetDb": fkTarget.get("db", ""),
"targetTable": fkTarget.get("table", ""),
"targetColumn": fkTarget.get("column", "id"),
"labelField": fkTarget.get("labelField"),
"softFk": fkTarget.get("softFk", False),
})
fields.append({
"name": fieldName,
"type": typeName,
"optional": isOptional,
"description": fieldInfo.description or "",
})
schema[dbName].append({
"tableName": tableName,
"pk": pkField,
"fields": fields,
"fks": fkRefs,
"modelClass": f"{modelCls.__module__}.{modelCls.__name__}",
})
return dict(schema), tableToDb
def _resolveTypeName(annotation) -> str:
"""Best-effort stringification of a type annotation."""
if annotation is None:
return "Any"
origin = getattr(annotation, "__origin__", None)
if origin is not None:
args = getattr(annotation, "__args__", ())
if str(origin) == "typing.Union" or getattr(origin, "__name__", "") == "Union":
nonNone = [a for a in args if a is not type(None)]
if len(nonNone) == 1:
return f"Optional[{_resolveTypeName(nonNone[0])}]"
return f"Union[{', '.join(_resolveTypeName(a) for a in args)}]"
argStr = ", ".join(_resolveTypeName(a) for a in args)
name = getattr(origin, "__name__", str(origin))
return f"{name}[{argStr}]" if argStr else name
return getattr(annotation, "__name__", str(annotation))
def _renderMarkdown(schema: Dict[str, List[dict]]) -> str:
"""Render the schema as markdown."""
from modules.shared.dbRegistry import getRegisteredDatabases
registeredDbs = getRegisteredDatabases()
now = datetime.now().strftime("%Y-%m-%d %H:%M")
totalTables = sum(len(tables) for tables in schema.values())
totalFks = sum(len(t["fks"]) for tables in schema.values() for t in tables)
lines = [
"# PowerOn Database Schema\n",
f"> **Generated from**: Pydantic MODEL_REGISTRY + fk_target",
f"> **Date**: {now}",
f"> **Registered databases**: {len(registeredDbs)}",
f"> **Tables**: {totalTables}",
f"> **FK relationships**: {totalFks}\n",
"---\n",
]
for dbName in sorted(schema.keys()):
tables = schema[dbName]
lines.append(f"## {dbName}\n")
for tbl in sorted(tables, key=lambda t: t["tableName"]):
lines.append(f"### {tbl['tableName']}\n")
if tbl["pk"]:
lines.append(f"- **PK**: `{tbl['pk']['name']}` ({tbl['pk']['type']})")
for fk in tbl["fks"]:
crossDb = ""
if fk["targetDb"] != dbName:
crossDb = f" [cross-db: {fk['targetDb']}]"
soft = " **(soft)**" if fk["softFk"] else ""
lines.append(
f"- **FK**: `{fk['column']}` -> `{fk['targetTable']}.{fk['targetColumn']}`{crossDb}{soft}"
)
nonFkFields = []
fkCols = {fk["column"] for fk in tbl["fks"]}
for f in tbl["fields"]:
if f["name"] in fkCols or f["name"].startswith("sys"):
continue
opt = " (optional)" if f["optional"] else ""
nonFkFields.append(f"`{f['name']}` {f['type']}{opt}")
if nonFkFields:
lines.append(f"- **Fields**: {', '.join(nonFkFields)}")
lines.append("")
return "\n".join(lines)
def _validateAgainstLiveDb(schema: Dict[str, List[dict]], tableToDb: Dict[str, str]) -> List[str]:
"""Compare Pydantic schema against live PostgreSQL and return mismatch warnings."""
from modules.shared.configuration import APP_CONFIG
import psycopg2
import psycopg2.extras
host = APP_CONFIG.get("DB_HOST", "localhost")
port = int(APP_CONFIG.get("DB_PORT", 5432))
user = APP_CONFIG.get("DB_USER", "poweron_dev")
password = APP_CONFIG.get("DB_PASSWORD_SECRET")
if not password:
return ["ERROR: DB_PASSWORD_SECRET not available for validation"]
warnings = []
for dbName, tables in sorted(schema.items()):
try:
conn = psycopg2.connect(
host=host, port=port, user=user, password=password,
database=dbName, client_encoding="utf8",
cursor_factory=psycopg2.extras.RealDictCursor,
)
except Exception as e:
warnings.append(f" {dbName}: connection failed ({e})")
continue
try:
with conn.cursor() as cur:
cur.execute("""
SELECT table_name FROM information_schema.tables
WHERE table_schema = 'public' AND table_type = 'BASE TABLE'
""")
liveTables = {row["table_name"] for row in cur.fetchall()}
for tbl in tables:
name = tbl["tableName"]
if name not in liveTables:
warnings.append(f" {dbName}.{name}: model exists but NO table in DB")
continue
with conn.cursor() as cur:
cur.execute("""
SELECT column_name FROM information_schema.columns
WHERE table_schema = 'public' AND table_name = %s
""", (name,))
liveCols = {row["column_name"] for row in cur.fetchall()}
modelCols = {"id"} | {f["name"] for f in tbl["fields"]}
missingInDb = modelCols - liveCols
legacyAuditCols = {
"_createdAt", "_createdBy", "_modifiedAt", "_modifiedBy",
"sysCreatedAt", "sysCreatedBy", "sysModifiedAt", "sysModifiedBy",
"createdAt", "updatedAt", "creationDate", "lastModified",
}
extraInDb = liveCols - modelCols - legacyAuditCols
if missingInDb:
warnings.append(f" {dbName}.{name}: columns in model but not in DB: {sorted(missingInDb)}")
if extraInDb:
warnings.append(f" {dbName}.{name}: columns in DB but not in model: {sorted(extraInDb)}")
modelTableNames = {t["tableName"] for t in tables}
for lt in sorted(liveTables):
if lt not in modelTableNames and not lt.startswith("_"):
warnings.append(f" {dbName}.{lt}: table in DB but no Pydantic model (legacy?)")
finally:
conn.close()
return warnings
def main():
args = _getArgs()
_loadAllModels()
print("Building schema from MODEL_REGISTRY...")
schema, tableToDb = _buildSchema()
totalTables = sum(len(t) for t in schema.values())
totalFks = sum(len(t["fks"]) for tables in schema.values() for t in tables)
print(f" {len(schema)} databases, {totalTables} tables, {totalFks} FK relationships")
md = _renderMarkdown(schema)
with open(args.output, "w", encoding="utf-8") as f:
f.write(md)
print(f"\nSchema written to {args.output}")
if args.validate:
print("\nValidating against live database...")
warnings = _validateAgainstLiveDb(schema, tableToDb)
if warnings:
print(f"\n{len(warnings)} mismatches found:")
for w in warnings:
print(w)
else:
print(" No mismatches - live DB matches Pydantic models perfectly.")
if __name__ == "__main__":
main()