Add endpoint to update user

app/routers/users.py

@@ -8,9 +8,11 @@
 from app.services.users import (
     User,
     UserAdd,
+    UserUpdate,
     process_activate,
     process_add_user,
     process_get_me,
+    process_update,
 )
 
 router = APIRouter(prefix="/users")
@@ -71,3 +73,21 @@ async def activate(
     Personalize user experiences within the application using the JSON response containing user-specific data.
     """
     return await process_activate(sig, session)
+
+
+@router.put(
+    "/",
+    response_model=User,
+    name="update_user",
+)
+async def update(
+    user: UserUpdate,
+    token: Annotated[AccessTokenData, Depends(oauth2_scheme)],
+    session: AsyncSession = Depends(get_async_session),
+) -> User:
+    """Update user details.
+
+    This endpoint is crucial for users to manage and maintain accurate profile information,
+    often including authentication and authorization checks for security.
+    """
+    return await process_update(token, user, session)

app/services/users.py

@@ -1,6 +1,7 @@
 from datetime import datetime
 from gettext import gettext as _
 from json import dumps, loads
+from typing import Dict
 from urllib.parse import urlencode
 from uuid import UUID
 
@@ -60,13 +61,17 @@ class UserBase(BaseModel):
     )
 
 
-class UserAdd(UserBase):
+class PasswordHashMixin:
     @field_validator("password", mode="before")
     @classmethod
     def hash_password(cls, value: str) -> str:
         return hasher.encode(value)
 
 
+class UserAdd(UserBase, PasswordHashMixin):
+    ...
+
+
 class User(UserBase):
     id: int
     is_confirmed: bool = Field(alias="isConfirmed")
@@ -167,3 +172,55 @@ async def process_activate(signature: str, session: AsyncSession, /) -> User:
     await session.commit()
 
     return User.model_validate(user)
+
+
+class UserUpdate(BaseModel, PasswordHashMixin):
+    name: str | None = Field(
+        min_length=1,
+        max_length=64,
+        default=None,
+    )
+    surname: str | None = Field(
+        min_length=1,
+        max_length=64,
+        default=None,
+    )
+    middle_name: str | None = Field(
+        default=None,
+        alias="middleName",
+        min_length=1,
+        max_length=64,
+    )
+    password: str | None = Field(
+        default=None,
+        min_length=8,
+        max_length=128,
+    )
+    is_subscribed: bool | None = Field(
+        default=None,
+        alias="isSubscribed",
+    )
+
+    model_config = ConfigDict(
+        from_attributes=True,
+        populate_by_name=True,
+    )
+
+
+async def process_update(
+    token: AccessTokenData,
+    request_user: UserUpdate,
+    session: AsyncSession,
+    /,
+) -> User:
+    request_user_dict: Dict = request_user.model_dump(exclude_none=True)
+    if not request_user_dict:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
+    try:
+        user: UserModel = await UserModel.get(session, token.uuid, field=UserModel.uuid)
+    except UserDoesNotExist:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
+    for field, value in request_user_dict.items():
+        setattr(user, field, value)
+    await session.commit()
+    return user