Fix #13: add /me update route

docs/usage/routes.md

@@ -104,3 +104,28 @@ Return the current authenticated active user.
 
 !!! fail "`401 Unauthorized`"
     Missing token or inactive user.
+
+### `PATCH /me`
+
+Update the current authenticated active user.
+
+!!! abstract "Payload"
+    ```json
+    {
+        "email": "king.arthur@tintagel.bt",
+        "password": "merlin"
+    }
+    ```
+
+!!! success "`200 OK`"
+    ```json
+    {
+        "id": "57cbb51a-ab71-4009-8802-3f54b4f2e23",
+        "email": "king.arthur@tintagel.bt",
+        "is_active": true,
+        "is_superuser": false
+    }
+    ```
+
+!!! fail "`401 Unauthorized`"
+    Missing token or inactive user.

fastapi_users/models.py

@@ -18,6 +18,11 @@ class BaseUser(BaseModel):
     def default_id(cls, v):
         return v or str(uuid.uuid4())
 
+    def create_update_dict(self):
+        return self.dict(
+            skip_defaults=True, exclude={"id", "is_superuser", "is_active"}
+        )
+
 
 class BaseUserCreate(BaseUser):
     email: EmailStr
@@ -25,7 +30,7 @@ class BaseUserCreate(BaseUser):
 
 
 class BaseUserUpdate(BaseUser):
-    pass
+    password: Optional[str]
 
 
 class BaseUserDB(BaseUser):

fastapi_users/router.py

@@ -45,8 +45,7 @@ def get_user_router(
 
         hashed_password = get_password_hash(user.password)
         db_user = models.UserDB(
-            **user.dict(exclude={"id", "is_superuser", "is_active"}),
-            hashed_password=hashed_password
+            **user.create_update_dict(), hashed_password=hashed_password
         )
         created_user = await user_db.create(db_user)
         return created_user
@@ -110,4 +109,19 @@ def get_user_router(
     ):
         return user
 
+    @router.patch("/me", response_model=models.User)
+    async def update_me(
+        updated_user: models.UserUpdate,  # type: ignore
+        user: models.UserDB = Depends(get_current_active_user),  # type: ignore
+    ):
+        updated_user_data = updated_user.create_update_dict()
+        for field in updated_user_data:
+            if field == "password":
+                hashed_password = get_password_hash(updated_user_data[field])
+                user.hashed_password = hashed_password
+            else:
+                setattr(user, field, updated_user_data[field])
+
+        return await user_db.update(user)
+
     return router

tests/test_router.py

@@ -89,7 +89,7 @@ class TestRegister:
         response_json = response.json()
         assert "hashed_password" not in response_json
         assert "password" not in response_json
-        assert "id" in response_json
+        assert response_json["id"] is not None
 
     def test_valid_body_is_superuser(self, test_app_client: TestClient):
         json = {
@@ -289,3 +289,74 @@ class TestMe:
         response_json = response.json()
         assert response_json["id"] == user.id
         assert response_json["email"] == user.email
+
+
+class TestUpdateMe:
+    def test_missing_token(self, test_app_client: TestClient):
+        response = test_app_client.patch("/me")
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+    def test_inactive_user(
+        self, test_app_client: TestClient, inactive_user: BaseUserDB
+    ):
+        response = test_app_client.patch(
+            "/me", headers={"Authorization": f"Bearer {inactive_user.id}"}
+        )
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED
+
+    def test_empty_body(self, test_app_client: TestClient, user: BaseUserDB):
+        response = test_app_client.patch(
+            "/me", json={}, headers={"Authorization": f"Bearer {user.id}"}
+        )
+        assert response.status_code == status.HTTP_200_OK
+
+        response_json = response.json()
+        assert response_json["email"] == user.email
+
+    def test_valid_body(self, test_app_client: TestClient, user: BaseUserDB):
+        json = {"email": "king.arthur@tintagel.bt"}
+        response = test_app_client.patch(
+            "/me", json=json, headers={"Authorization": f"Bearer {user.id}"}
+        )
+        assert response.status_code == status.HTTP_200_OK
+
+        response_json = response.json()
+        assert response_json["email"] == "king.arthur@tintagel.bt"
+
+    def test_valid_body_is_superuser(
+        self, test_app_client: TestClient, user: BaseUserDB
+    ):
+        json = {"is_superuser": True}
+        response = test_app_client.patch(
+            "/me", json=json, headers={"Authorization": f"Bearer {user.id}"}
+        )
+        assert response.status_code == status.HTTP_200_OK
+
+        response_json = response.json()
+        assert response_json["is_superuser"] is False
+
+    def test_valid_body_is_active(self, test_app_client: TestClient, user: BaseUserDB):
+        json = {"is_active": False}
+        response = test_app_client.patch(
+            "/me", json=json, headers={"Authorization": f"Bearer {user.id}"}
+        )
+        assert response.status_code == status.HTTP_200_OK
+
+        response_json = response.json()
+        assert response_json["is_active"] is True
+
+    def test_valid_body_password(
+        self, mocker, mock_user_db, test_app_client: TestClient, user: BaseUserDB
+    ):
+        mocker.spy(mock_user_db, "update")
+        current_hashed_passord = user.hashed_password
+
+        json = {"password": "merlin"}
+        response = test_app_client.patch(
+            "/me", json=json, headers={"Authorization": f"Bearer {user.id}"}
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert mock_user_db.update.called is True
+
+        updated_user = mock_user_db.update.call_args[0][0]
+        assert updated_user.hashed_password != current_hashed_passord