Initialize the root user if no users exist.
(self, user_id: str)
| 128 | return self.SessionLocal() |
| 129 | |
| 130 | def _init_root_user(self, user_id: str) -> None: |
| 131 | """Initialize the root user if no users exist.""" |
| 132 | session = self._get_session() |
| 133 | try: |
| 134 | # Check if any users exist |
| 135 | user_count = session.query(User).count() |
| 136 | if user_count == 0: |
| 137 | root_user = User(user_id=user_id, user_name=user_id, role=UserRole.ROOT) |
| 138 | session.add(root_user) |
| 139 | session.commit() |
| 140 | logger.info("Root user created successfully") |
| 141 | else: |
| 142 | self.create_user(user_name=user_id, user_id=user_id, role=UserRole.ROOT) |
| 143 | except Exception as e: |
| 144 | session.rollback() |
| 145 | logger.error(f"Failed to create {user_id} user: {e}") |
| 146 | finally: |
| 147 | session.close() |
| 148 | |
| 149 | def create_user( |
| 150 | self, user_name: str, role: UserRole = UserRole.USER, user_id: str | None = None |
no test coverage detected