♻️ Refactor user update (#689)

backend/app/api/routes/users.py

@@ -80,6 +80,12 @@ def update_user_me(
     Update own user.
     """
 
+    if user_in.email:
+        existing_user = crud.get_user_by_email(session=session, email=user_in.email)
+        if existing_user:
+            raise HTTPException(
+                status_code=409, detail="User with this email already exists"
+            )
     user_data = user_in.model_dump(exclude_unset=True)
     current_user.sqlmodel_update(user_data)
     session.add(current_user)
@@ -170,12 +176,20 @@ def update_user(
     Update a user.
     """
 
-    db_user = crud.update_user(session=session, user_id=user_id, user_in=user_in)
-    if db_user is None:
+    db_user = session.get(User, user_id)
+    if not db_user:
         raise HTTPException(
             status_code=404,
             detail="The user with this id does not exist in the system",
         )
+    if user_in.email:
+        existing_user = crud.get_user_by_email(session=session, email=user_in.email)
+        if existing_user:
+            raise HTTPException(
+                status_code=409, detail="User with this email already exists"
+            )
+
+    db_user = crud.update_user(session=session, db_user=db_user, user_in=user_in)
     return db_user
 
 

backend/app/crud.py

@@ -16,10 +16,7 @@ def create_user(*, session: Session, user_create: UserCreate) -> User:
     return db_obj
 
 
-def update_user(*, session: Session, user_id: int, user_in: UserUpdate) -> Any:
-    db_user = session.get(User, user_id)
-    if not db_user:
-        return None
+def update_user(*, session: Session, db_user: User, user_in: UserUpdate) -> Any:
     user_data = user_in.model_dump(exclude_unset=True)
     extra_data = {}
     if "password" in user_data:

backend/app/tests/api/routes/test_users.py

@@ -170,7 +170,7 @@ def test_update_user_me(
     client: TestClient, normal_user_token_headers: dict[str, str], db: Session
 ) -> None:
     full_name = "Updated Name"
-    email = "updated email"
+    email = random_email()
     data = {"full_name": full_name, "email": email}
     r = client.patch(
         f"{settings.API_V1_STR}/users/me",
@@ -228,6 +228,24 @@ def test_update_password_me_incorrect_password(
     assert updated_user["detail"] == "Incorrect password"
 
 
+def test_update_user_me_email_exists(
+    client: TestClient, normal_user_token_headers: dict[str, str], db: Session
+) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    user = crud.create_user(session=db, user_create=user_in)
+
+    data = {"email": user.email}
+    r = client.patch(
+        f"{settings.API_V1_STR}/users/me",
+        headers=normal_user_token_headers,
+        json=data,
+    )
+    assert r.status_code == 409
+    assert r.json()["detail"] == "User with this email already exists"
+
+
 def test_update_password_me_same_password_error(
     client: TestClient, superuser_token_headers: dict[str, str], db: Session
 ) -> None:
@@ -330,6 +348,29 @@ def test_update_user_not_exists(
     assert r.json()["detail"] == "The user with this id does not exist in the system"
 
 
+def test_update_user_email_exists(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    user = crud.create_user(session=db, user_create=user_in)
+
+    username2 = random_email()
+    password2 = random_lower_string()
+    user_in2 = UserCreate(email=username2, password=password2)
+    user2 = crud.create_user(session=db, user_create=user_in2)
+
+    data = {"email": user2.email}
+    r = client.patch(
+        f"{settings.API_V1_STR}/users/{user.id}",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert r.status_code == 409
+    assert r.json()["detail"] == "User with this email already exists"
+
+
 def test_delete_user_super_user(
     client: TestClient, superuser_token_headers: dict[str, str], db: Session
 ) -> None:

backend/app/tests/crud/test_user.py

@@ -84,7 +84,7 @@ def test_update_user(db: Session) -> None:
     new_password = random_lower_string()
     user_in_update = UserUpdate(password=new_password, is_superuser=True)
     if user.id is not None:
-        crud.update_user(session=db, user_id=user.id, user_in=user_in_update)
+        crud.update_user(session=db, db_user=user, user_in=user_in_update)
     user_2 = db.get(User, user.id)
     assert user_2
     assert user.email == user_2.email

backend/app/tests/utils/user.py

@@ -44,6 +44,6 @@ def authentication_token_from_email(
         user_in_update = UserUpdate(password=password)
         if not user.id:
             raise Exception("User id not set")
-        user = crud.update_user(session=db, user_id=user.id, user_in=user_in_update)
+        user = crud.update_user(session=db, db_user=user, user_in=user_in_update)
 
     return user_authentication_headers(client=client, email=email, password=password)