Update user information. Updates user fields with validation for unique constraints on email and username. Only provided fields are updated. Args: user_id: ID of the user to update. user_update: Fields to update with new values. db: Datab
(self, user_id: int, user_update: UserUpdate, db: AsyncSession)
| 225 | return user |
| 226 | |
| 227 | async def update(self, user_id: int, user_update: UserUpdate, db: AsyncSession) -> dict[str, Any]: |
| 228 | """Update user information. |
| 229 | |
| 230 | Updates user fields with validation for unique constraints on email |
| 231 | and username. Only provided fields are updated. |
| 232 | |
| 233 | Args: |
| 234 | user_id: ID of the user to update. |
| 235 | user_update: Fields to update with new values. |
| 236 | db: Database session for the operation. |
| 237 | |
| 238 | Returns: |
| 239 | Updated user data dictionary. |
| 240 | |
| 241 | Raises: |
| 242 | UserNotFoundError: If the user doesn't exist. |
| 243 | UserExistsError: If email or username conflicts with existing users. |
| 244 | |
| 245 | Note: |
| 246 | Validates uniqueness when updating email or username. |
| 247 | Only non-deleted users can be updated. |
| 248 | |
| 249 | Example: |
| 250 | ```python |
| 251 | update_data = UserUpdate( |
| 252 | email="newemail@example.com", |
| 253 | first_name="John" |
| 254 | ) |
| 255 | updated_user = await service.update(123, update_data, db) |
| 256 | ``` |
| 257 | """ |
| 258 | existing_user = await crud_users.get(db=db, id=user_id, is_deleted=False) |
| 259 | if not existing_user: |
| 260 | raise UserNotFoundError(f"User with ID {user_id} not found") |
| 261 | |
| 262 | update_data = user_update.model_dump(exclude_unset=True) |
| 263 | |
| 264 | if "email" in update_data and update_data["email"] != existing_user["email"]: |
| 265 | email_exists = await crud_users.exists(db=db, email=update_data["email"]) |
| 266 | if email_exists: |
| 267 | raise UserExistsError("Email already registered") |
| 268 | |
| 269 | if "username" in update_data and update_data["username"] != existing_user["username"]: |
| 270 | username_exists = await crud_users.exists(db=db, username=update_data["username"]) |
| 271 | if username_exists: |
| 272 | raise UserExistsError("Username already taken") |
| 273 | |
| 274 | updated_user = await crud_users.update( |
| 275 | db=db, object=user_update, id=user_id, return_columns=list(UserSchema.model_fields.keys()) |
| 276 | ) |
| 277 | if not updated_user: |
| 278 | raise UserNotFoundError(f"User with ID {user_id} not found") |
| 279 | return updated_user |
| 280 | |
| 281 | async def check_update_permission(self, requester_user: dict[str, Any], target_username: str) -> bool: |
| 282 | """Check if user has permission to update another user. |