Create a new user. Args: user_name (str): Name of the user. role (UserRole): Role of the user. user_id (str, optional): Custom user ID. If None, generates UUID. Returns: str: The created user ID. Raises: ValueErro
(
self, user_name: str, role: UserRole = UserRole.USER, user_id: str | None = None
)
| 147 | session.close() |
| 148 | |
| 149 | def create_user( |
| 150 | self, user_name: str, role: UserRole = UserRole.USER, user_id: str | None = None |
| 151 | ) -> str: |
| 152 | """Create a new user. |
| 153 | |
| 154 | Args: |
| 155 | user_name (str): Name of the user. |
| 156 | role (UserRole): Role of the user. |
| 157 | user_id (str, optional): Custom user ID. If None, generates UUID. |
| 158 | |
| 159 | Returns: |
| 160 | str: The created user ID. |
| 161 | |
| 162 | Raises: |
| 163 | ValueError: If user_name already exists. |
| 164 | """ |
| 165 | session = self._get_session() |
| 166 | try: |
| 167 | # Check if user_name already exists |
| 168 | existing_user = session.query(User).filter(User.user_name == user_name).first() |
| 169 | if existing_user: |
| 170 | logger.info(f"User with name '{user_name}' already exists") |
| 171 | return existing_user.user_id |
| 172 | user = User(user_name=user_name, role=role, user_id=user_id or str(uuid.uuid4())) |
| 173 | session.add(user) |
| 174 | session.commit() |
| 175 | logger.info(f"User '{user_name}' created with ID: {user.user_id}") |
| 176 | return user.user_id |
| 177 | except IntegrityError: |
| 178 | session.rollback() |
| 179 | logger.info(f"failed to create user with name '{user_name}' already exists") |
| 180 | except Exception as e: |
| 181 | session.rollback() |
| 182 | logger.error(f"Error creating user: {e}") |
| 183 | raise |
| 184 | finally: |
| 185 | session.close() |
| 186 | |
| 187 | def get_user(self, user_id: str) -> User | None: |
| 188 | """Get user by ID. |
no test coverage detected