(session: SessionDep, editor: UserEditor, trans: Trans)
| 211 | resource_id_expr="editor.id" |
| 212 | )) |
| 213 | async def update(session: SessionDep, editor: UserEditor, trans: Trans): |
| 214 | user_model: UserModel = get_db_user(session = session, user_id = editor.id) |
| 215 | if not user_model: |
| 216 | raise Exception(f"User with id [{editor.id}] not found!") |
| 217 | if editor.account != user_model.account: |
| 218 | raise Exception(f"account cannot be changed!") |
| 219 | """ if editor.email != user_model.email and check_email_exists(session=session, email=editor.email): |
| 220 | raise Exception(trans('i18n_exist', msg = f"{trans('i18n_user.email')} [{editor.email}]")) """ |
| 221 | if not check_email_format(editor.email): |
| 222 | raise Exception(trans('i18n_format_invalid', key = f"{trans('i18n_user.email')} [{editor.email}]")) |
| 223 | origin_oid: int = user_model.oid |
| 224 | |
| 225 | uws_list_stmt = select(UserWsModel).where(UserWsModel.uid == editor.id) |
| 226 | uws_list = session.exec(uws_list_stmt).all() |
| 227 | |
| 228 | existing_oids = {uws.oid for uws in uws_list} |
| 229 | new_oid_set = set(editor.oid_list) if editor.oid_list else set() |
| 230 | oids_to_remove = existing_oids - new_oid_set |
| 231 | oids_to_add = new_oid_set - existing_oids |
| 232 | |
| 233 | if oids_to_remove: |
| 234 | del_stmt = sqlmodel_delete(UserWsModel).where(UserWsModel.uid == editor.id, UserWsModel.oid.in_(oids_to_remove)) |
| 235 | session.exec(del_stmt) |
| 236 | |
| 237 | data = editor.model_dump(exclude_unset=True) |
| 238 | user_model.sqlmodel_update(data) |
| 239 | |
| 240 | user_model.oid = 0 |
| 241 | if editor.oid_list: |
| 242 | user_model.oid = origin_oid if origin_oid in editor.oid_list else editor.oid_list[0] |
| 243 | if oids_to_add: |
| 244 | db_uws_model_list = [ |
| 245 | UserWsModel.model_validate({ |
| 246 | "oid": oid, |
| 247 | "uid": user_model.id, |
| 248 | "weight": 0 |
| 249 | }) |
| 250 | for oid in oids_to_add |
| 251 | ] |
| 252 | session.add_all(db_uws_model_list) |
| 253 | session.add(user_model) |
| 254 | |
| 255 | @router.delete("/{id}", summary=f"{PLACEHOLDER_PREFIX}user_del_api", description=f"{PLACEHOLDER_PREFIX}user_del_api") |
| 256 | @require_permissions(permission=SqlbotPermission(role=['admin'])) |
no test coverage detected