# Copyright (c) 2025 Patrick Motsch # All rights reserved. """ Root access management for system-level operations. Provides secure access to root user and DbApp database connector. Bootstrap is guaranteed by app.py lifespan before any access. """ import logging from modules.connectors.connectorDbPostgre import DatabaseConnector from modules.datamodels.datamodelUam import User, UserInDB from modules.shared.configuration import APP_CONFIG logger = logging.getLogger(__name__) _rootDbAppConnector = None _rootUser = None def getRootDbAppConnector() -> DatabaseConnector: """ Returns a DatabaseConnector instance for the DbApp database. This is used for accessing system tables like AccessRule. """ global _rootDbAppConnector if _rootDbAppConnector is None: _rootDbAppConnector = DatabaseConnector( dbHost=APP_CONFIG.get("DB_HOST"), dbDatabase="poweron_app", dbUser=APP_CONFIG.get("DB_USER"), dbPassword=APP_CONFIG.get("DB_PASSWORD_SECRET"), dbPort=int(APP_CONFIG.get("DB_PORT", 5432)), userId=None # No user context for root connector ) _rootDbAppConnector.initDbSystem() return _rootDbAppConnector def getRootUser() -> User: """ Returns the root user (initial user from database). Used for system-level operations that require root privileges. Raises RuntimeError if no user exists (bootstrap incomplete). """ global _rootUser if _rootUser is None: dbApp = getRootDbAppConnector() initialUserId = dbApp.getInitialId(UserInDB) if not initialUserId: raise RuntimeError( "No root user found - bootstrap incomplete. " "Ensure app.py lifespan runs initBootstrap before any service access." ) users = dbApp.getRecordset(UserInDB, recordFilter={"id": initialUserId}) if not users: raise RuntimeError("Initial user not found in database") user_data = users[0] _rootUser = User(**user_data) return _rootUser