feat: update profile

app/blueprints/user.py

@@ -16,3 +16,25 @@ def get_users(user_controller: UserController = Provide[Container.user_controlle
 @inject
 def register(user_controller: UserController = Provide[Container.user_controller]):
     return user_controller.register()
+
+
+@user_blueprint.route("/user/update", methods=["POST"])
+@inject
+def update_user(user_controller: UserController = Provide[Container.user_controller]):
+    return user_controller.update_profile()
+
+
+@user_blueprint.route("/user/change-password", methods=["POST"])
+@inject
+def change_password(
+    user_controller: UserController = Provide[Container.user_controller],
+):
+    return user_controller.change_password()
+
+
+@user_blueprint.route("/user/<string:user_id>", methods=["GET"])
+@inject
+def get_user(
+    user_id, user_controller: UserController = Provide[Container.user_controller]
+):
+    return user_controller.get_user_by_id(user_id)

app/controllers/user_controller.py

@@ -4,8 +4,9 @@ from app.services import UserService
 from app.schemas import RegisterSchema
 from pydantic import ValidationError
 from app.schemas import ResponseSchema
-from app.exception import AlreadyExistException
+from app.exception import AlreadyExistException, DataNotFoundException
 from app.helpers import make_response
+from app.schemas.requests import ProfileUpdateSchema
 
 
 class UserController:
@@ -23,7 +24,6 @@ class UserController:
             current_app.logger.error(f"Validation error: {e}")
             response = ResponseSchema(message="Invalid input", data=None, meta=None)
             return make_response("Invalid input", status_code=400)
-
         except AlreadyExistException as e:
             return make_response("User already exists", status_code=409)
         except Exception as e:
@@ -31,3 +31,86 @@ class UserController:
                 f"Error during Google login: {str(e)}", exc_info=True
             )
             return make_response("Internal server error", status_code=500)
+
+    def get_user_by_id(self, user_id):
+        try:
+            if not user_id:
+                return make_response("User ID is required", status_code=400)
+
+            user = self.user_service.get_user_by_id(user_id)
+            if user:
+                return make_response("User found", data=user)
+            else:
+                return make_response("User not found", status_code=404)
+        except Exception as e:
+            current_app.logger.error(
+                f"Error while retrieving user: {str(e)}", exc_info=True
+            )
+            return make_response(
+                message="An internal server error occurred. Please try again later.",
+                status_code=500,
+            )
+
+    def update_profile(self):
+        try:
+            body = request.get_json()
+            reqBody = ProfileUpdateSchema(**body)
+            result = self.user_service.update_profile(reqBody)
+
+            if result:
+                return make_response(message="User profile updated successfully.")
+            else:
+                return make_response(
+                    message="Failed to update user profile. Please check the submitted data.",
+                    status_code=400,
+                )
+        except DataNotFoundException as e:
+            return make_response(message="User data not found.", status_code=404)
+        except ValueError as e:
+            return make_response(
+                message=f"Invalid data provided: {str(e)}", status_code=400
+            )
+        except Exception as e:
+            current_app.logger.error(
+                f"Error while updating profile: {str(e)}", exc_info=True
+            )
+            return make_response(
+                message="An internal server error occurred. Please try again later.",
+                status_code=500,
+            )
+
+    def change_password(self):
+        try:
+            body = request.get_json()
+            user_id = body.get("id")
+            current_password = body.get("current_password")
+            new_password = body.get("new_password")
+
+            if not all([user_id, current_password, new_password]):
+                return make_response(
+                    message="Missing required fields: id, current_password, new_password",
+                    status_code=400,
+                )
+
+            result = self.user_service.change_password(
+                user_id, current_password, new_password
+            )
+            if result:
+                return make_response(message="Password changed successfully.")
+            else:
+                return make_response(
+                    message="Failed to change password.",
+                    status_code=400,
+                )
+        except DataNotFoundException as e:
+            return make_response(message="User data not found.", status_code=404)
+        except ValueError as e:
+            return make_response(message=f"{str(e)}", status_code=400)
+        except Exception as e:
+            current_app.logger.error(
+                f"Error while changing password: {str(e)}", exc_info=True
+            )
+            return make_response(
+                message="An internal server error occurred. Please try again later.",
+                status_code=500,
+            )

app/schemas/requests/__init__.py

@@ -11,6 +11,9 @@ from .answer.answer_item_request_schema import AnswerItemSchema
 from .subject.create_subject_schema import SubjectCreateRequest
 from .subject.update_subject_schema import SubjectUpdateRequest
 
+from .user.profile_update_schema import ProfileUpdateSchema
+from .user.password_change_schema import PasswordChangeSchema
+
 
 __all__ = [
     "RegisterSchema",
@@ -20,4 +23,6 @@ __all__ = [
     "AnswerItemSchema",
     "SubjectCreateRequest",
     "SubjectUpdateRequest",
+    "PasswordChangeSchema",
+    "ProfileUpdateSchema",
 ]

app/schemas/requests/user/password_change_schema.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+
+
+class PasswordChangeSchema(BaseModel):
+    """Schema for changing user password"""
+
+    current_password: str
+    new_password: str

app/schemas/requests/user/profile_update_schema.py

@@ -0,0 +1,12 @@
+from typing import Optional
+from pydantic import BaseModel
+from datetime import datetime
+
+
+class ProfileUpdateSchema(BaseModel):
+
+    id: str
+    name: Optional[str] = None
+    birth_date: Optional[str] = None
+    locale: Optional[str] = None
+    phone: Optional[str] = None

app/schemas/response/__init__.py

@@ -7,6 +7,7 @@ from .history.detail_history_response import QuizHistoryResponse, QuestionResult
 from .recomendation.recomendation_response_schema import ListingQuizResponse
 from .subject.get_subject_schema import GetSubjectResponse
 from .auth.login_response import LoginResponseSchema
+from .user.user_response_scema import UserResponseSchema
 
 __all__ = [
     "QuizCreationResponse",
@@ -19,4 +20,5 @@ __all__ = [
     "ListingQuizResponse",
     "GetSubjectResponse",
     "LoginResponseSchema",
+    "UserResponseSchema",
 ]

app/schemas/response/user/user_response_scema.py

@@ -0,0 +1,15 @@
+from pydantic import BaseModel
+from typing import Optional
+
+
+class UserResponseSchema(BaseModel):
+    id: str
+    google_id: Optional[str]
+    email: str
+    name: str
+    birth_date: Optional[str]
+    pic_url: Optional[str]
+    phone: Optional[str]
+    locale: str
+    created_at: str
+    updated_at: str

app/services/user_service.py

@@ -1,9 +1,12 @@
-from flask import current_app
+from datetime import datetime
 from app.repositories import UserRepository
 from app.schemas import RegisterSchema
+from app.schemas.requests import ProfileUpdateSchema
+from app.schemas.response import UserResponseSchema
 from app.mapper import UserMapper
-from app.exception import AlreadyExistException
+from app.exception import AlreadyExistException, DataNotFoundException
-from werkzeug.security import generate_password_hash
+from werkzeug.security import generate_password_hash, check_password_hash
+from app.helpers import DatetimeUtil
 
 
 class UserService:
@@ -23,3 +26,71 @@ class UserService:
 
         data = UserMapper.from_register(user_data)
         return self.user_repository.insert_user(data)
+
+    def update_profile(self, new_profile: ProfileUpdateSchema):
+        user = self.user_repository.get_user_by_id(new_profile.id)
+        if not user:
+            raise DataNotFoundException(entity="User")
+
+        update_data = {}
+        if new_profile.name is not None:
+            update_data["name"] = new_profile.name
+        if new_profile.birth_date is not None:
+            update_data["birth_date"] = DatetimeUtil.from_string(
+                new_profile.birth_date, fmt="%d-%m-%Y"
+            )
+        if new_profile.locale is not None:
+            update_data["locale"] = new_profile.locale
+        if new_profile.phone is not None:
+            update_data["phone"] = new_profile.phone
+
+        if not update_data:
+            return True
+
+        update_data["updated_at"] = DatetimeUtil.now_iso()
+
+        return self.user_repository.update_user(new_profile.id, update_data)
+
+    def change_password(self, user_id: str, current_password: str, new_password: str):
+
+        user = self.user_repository.get_user_by_id(user_id)
+        if not user:
+            raise DataNotFoundException(entity="User")
+
+        if not user.password or not check_password_hash(
+            user.password, current_password
+        ):
+            raise ValueError("Current password is incorrect")
+
+        encrypted_password = generate_password_hash(new_password)
+
+        update_data = {
+            "password": encrypted_password,
+            "updated_at": DatetimeUtil.now_iso(),
+        }
+        return self.user_repository.update_user(user_id, update_data)
+
+    def get_user_by_id(self, user_id: str):
+        user = self.user_repository.get_user_by_id(user_id)
+        if not user:
+            raise DataNotFoundException(entity="User")
+        user_dict = user.model_dump()
+
+        if "password" in user_dict:
+            del user_dict["password"]
+
+        if "id" in user_dict:
+            user_dict["id"] = str(user.id)
+
+        if "birth_date" in user_dict and user_dict["birth_date"]:
+            user_dict["birth_date"] = DatetimeUtil.to_string(
+                user_dict["birth_date"], fmt="%d-%m-%Y"
+            )
+
+        if "created_at" in user_dict and user_dict["created_at"]:
+            user_dict["created_at"] = DatetimeUtil.to_string(user_dict["created_at"])
+
+        if "updated_at" in user_dict and user_dict["updated_at"]:
+            user_dict["updated_at"] = DatetimeUtil.to_string(user_dict["updated_at"])
+
+        return UserResponseSchema(**user_dict)