def update_user()

in microservices/user_management/src/routes/user.py [0:0]

• 103 lines of code
• 32 McCabe index (conditional complexity)

def update_user(user_id: str,
                  input_user: UpdateUserModel,
                  request: Request,
                  update_inspace_user: Optional[bool] = False):
  """Update a user with the user_id passed in the request body

  ### Args:
      input_user (UserModel): Required body of the user

  ### Raises:
      ResourceNotFoundException: If the user does not exist
      Exception: 500 Internal Server Error if something went wrong

  ### Returns:
      UpdateUserResponseModel: User Object
  """
  try:
    if not is_inspace_enabled(update_inspace_user):
      raise ValidationError("Inspace is disabled, please make "
                            "'update_inspace_user' as false in "
                            "the API request body")
    existing_user = User.find_by_uuid(user_id)

    input_user_dict = {**input_user.dict(exclude_unset=True)}
    user_fields = existing_user.get_fields()

    # update learner or staff based on user_type
    update_dict = {}
    agent_dict = {}
    first_name = input_user_dict.get("first_name")
    last_name = input_user_dict.get("last_name")

    if input_user_dict.get("email"):
      updated_email = input_user_dict.get("email")
      _ = User.find_by_email(updated_email)
      if _ is not None:
        raise ConflictError(
          f"User with the given email: {updated_email} "
          "already exists")

      if user_fields["user_type"] == "learner":
        update_dict["email_address"] = input_user_dict.get("email")
      elif user_fields["user_type"] in STAFF_USERS:
        update_dict["email"] = input_user_dict.get("email")

    if input_user_dict.get("user_groups"):
      for uuid in input_user_dict["user_groups"]:
        user_group = UserGroup.find_by_uuid(uuid)
        if user_group.is_immutable and \
          user_fields["user_type"] != user_group.name:
          raise ValidationError((f"User of user type {user_fields['user_type']}"
                                 f" cannot be assigned to User Group "
                                 f"{user_group.name} with uuid {uuid}"))

    if first_name:
      update_dict["first_name"] = first_name
      agent_dict["name"] = first_name
    else:
      agent_dict["name"] = user_fields["first_name"]
    if last_name:
      update_dict["last_name"] = last_name
      agent_dict["name"] += " " + last_name
    else:
      agent_dict["name"] += " " + user_fields["last_name"]

    if update_dict:
      headers = {"Authorization": request.headers.get("Authorization")}
      if user_fields["user_type"] == "learner":  # update learner
        learner_id = user_fields["user_type_ref"]
        update_learner(headers, learner_id, learner_dict=update_dict)
      elif user_fields["user_type"] in STAFF_USERS:  # update staff
        staff_id = user_fields["user_type_ref"]
        update_staff(staff_id, input_staff=UpdateStaffModel(**update_dict))

    for key, value in input_user_dict.items():
      if key == "user_groups" and user_fields.get("status") == "inactive":
        value = []
      elif key == "user_groups":
        value = CollectionHandler.update_existing_references(
            user_id, "user_groups", "users",
            input_user_dict.get("user_groups", []),
            user_fields.get("user_groups", []))

      if value is not None:
        user_fields[key] = value

    for key, value in user_fields.items():
      setattr(existing_user, key, value)

    existing_user.update()
    user_fields = existing_user.get_fields(reformat_datetime=True)

    response_msg = "Successfully updated the user"

    if first_name or last_name:
      # update agent
      agent_id = get_agent(headers, user_id)
      update_agent(headers, agent_id, agent_dict)

      if update_inspace_user:
        if existing_user.inspace_user["is_inspace_user"] is True:
          # update Inspace User
          update_payload = {}
          if first_name is not None:
            update_payload["firstName"] = first_name
          if last_name is not None:
            update_payload["lastName"] = last_name

          is_update_successful = update_inspace_user_helper(existing_user,
                                                            update_payload)

          if is_update_successful is True:
            response_msg = "Successfully updated the user and corresponding"
            response_msg += " Inspace user"
          else:
            response_msg = "Successfully updated the user but corresponding"
            response_msg += " Inspace user couldn't be updated"
    return {
      "success": True,
      "message": response_msg,
      "data": user_fields
    }

  except ResourceNotFoundException as e:
    Logger.error(e)
    Logger.error(traceback.print_exc())
    raise ResourceNotFound(str(e)) from e
  except ConflictError as e:
    Logger.error(e)
    Logger.error(traceback.print_exc())
    raise Conflict(str(e)) from e
  except Exception as e:
    Logger.error(e)
    Logger.error(traceback.print_exc())
    raise InternalServerError(str(e)) from e