feat: add logout endpoint

Author: fiqrikm18

src/core/authorization/dependencies.py

@@ -28,7 +28,7 @@ async def dependency(
         ),
     ) -> dict:
         allowed = await authorization_service.can(
-            subject=str(current_user["id"]),
+            subject=str(current_user.get("id")),
             resource=resource,
             action=action,
         )

src/modules/user/application/logout_user/command.py

@@ -0,0 +1,7 @@
+from uuid import UUID
+
+from pydantic import BaseModel
+
+
+class LogoutUserCommand(BaseModel):
+    user_id: str

src/modules/user/application/logout_user/handler.py

@@ -0,0 +1,18 @@
+from src.modules.user.application.logout_user.command import LogoutUserCommand
+from src.modules.user.domain.repositories.refresh_token_repository import (
+    RefreshTokenRepository,
+)
+from src.shared.unit_of_work import UnitOfWork
+
+
+class LogoutUserCommandHandler:
+    def __init__(
+        self, refresh_token_repository: RefreshTokenRepository, unit_of_work: UnitOfWork
+    ):
+        self._refresh_token_repository = refresh_token_repository
+        self._unit_of_work = unit_of_work
+
+    async def excute(self, command: LogoutUserCommand) -> None:
+        async with self._unit_of_work:
+            self._refresh_token_repository.revoke_by_user_id(command.user_id)
+            await self._unit_of_work.commit()

src/modules/user/presentation/dependency.py

@@ -6,6 +6,7 @@
 from src.core.database.session import get_db, get_unit_of_work
 from src.modules.user.application.detail_user.handler import DetailUserQueryHandler
 from src.modules.user.application.login_user.handler import LoginUserCommandHandler
+from src.modules.user.application.logout_user.handler import LogoutUserCommandHandler
 from src.modules.user.application.refresh_token.handler import (
     RefreshTokenCommandHandler,
 )
@@ -62,3 +63,10 @@ def get_refresh_token_handler(
     unit_of_work: UnitOfWork = Depends(get_unit_of_work),
 ) -> RefreshTokenCommandHandler:
     return RefreshTokenCommandHandler(refresh_token_repo, unit_of_work)
+
+
+def get_logout_handler(
+    refresh_token_repo: RefreshTokenRepository = Depends(get_refresh_token_repository),
+    unit_of_work: UnitOfWork = Depends(get_unit_of_work),
+) -> LogoutUserCommandHandler:
+    return LogoutUserCommandHandler(refresh_token_repo, unit_of_work)

src/modules/user/presentation/routers/user_router.py

@@ -1,12 +1,14 @@
-from fastapi import APIRouter, Depends, HTTPException, status
-from fastapi.security import OAuth2PasswordRequestForm
-
-from src.core.authorization.dependencies import require_permission
-from src.core.authorization.permissions import ME_ACTION, USER_RESOURCE
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordRequestForm
+
+from src.core.authorization.dependencies import require_permission
+from src.core.authorization.permissions import ME_ACTION, UPDATE_ACTION, USER_RESOURCE
 from src.modules.user.application.detail_user.handler import DetailUserQueryHandler
 from src.modules.user.application.detail_user.query import DetailUserQuery
 from src.modules.user.application.login_user.command import LoginUserCommand
 from src.modules.user.application.login_user.handler import LoginUserCommandHandler
+from src.modules.user.application.logout_user.command import LogoutUserCommand
+from src.modules.user.application.logout_user.handler import LogoutUserCommandHandler
 from src.modules.user.application.refresh_token.command import RefreshTokenCommand
 from src.modules.user.application.refresh_token.handler import (
     RefreshTokenCommandHandler,
@@ -18,6 +20,7 @@
 from src.modules.user.domain.exceptions.user_exception import UserAlreadyExistsError
 from src.modules.user.presentation.dependency import (
     get_login_handler,
+    get_logout_handler,
     get_refresh_token_handler,
     get_register_handler,
     get_user_detail_handler,
@@ -69,9 +72,6 @@ async def refresh_token(
     handler: RefreshTokenCommandHandler = Depends(get_refresh_token_handler),
 ):
     try:
-        # Note: In strict rotation, we might not even require a valid access token here,
-        # just the refresh token. But requiring it adds a layer of security.
-        # We pass current_user["id"] to ensure the RT belongs to the user making the request.
         result = await handler.execute(
             RefreshTokenCommand(
                 token=request.refresh_token,
@@ -87,10 +87,10 @@ async def refresh_token(
 
 
 @router.get("/me")
-async def get_me(
-    current_user: dict = Depends(require_permission(USER_RESOURCE, ME_ACTION)),
-    handler: DetailUserQueryHandler = Depends(get_user_detail_handler),
-):
+async def get_me(
+    current_user: dict = Depends(require_permission(USER_RESOURCE, ME_ACTION)),
+    handler: DetailUserQueryHandler = Depends(get_user_detail_handler),
+):
     user = await handler.execute(
         DetailUserQuery(
             user_id=current_user.get("id"),
@@ -99,4 +99,9 @@ async def get_me(
     return {"id": str(user.id), "email": user.email}
 
 
-# TODO: need logout endpoint
+@router.post("/logout", status_code=status.HTTP_204_NO_CONTENT)
+async def logout(
+    current_user: dict = Depends(require_permission(USER_RESOURCE, UPDATE_ACTION)),
+    handler: LogoutUserCommandHandler = Depends(get_logout_handler),
+):
+    await handler.excute(LogoutUserCommand(user_id=str(current_user.get("id"))))