from fastapi import APIRouter, Depends, Response, Request from sqlalchemy.ext.asyncio import AsyncSession from typing import List from app.database.session import get_db from app.domain.entities.auth import AuthEntity from app.domain.entities.responses.session import SessionEntity from app.domain.entities.token_entity import TokenEntity from app.infrastructure.auth_service import AuthService from app.infrastructure.dependencies import get_current_user from app.domain.models.users import User router = APIRouter() @router.post( "/login/", response_model=TokenEntity, responses={401: {"description": "Invalid username or password"}}, summary="User authentication", description="Logs in the user and outputs the `access_token` in the `cookie'", ) async def auth_user( response: Response, user_data: AuthEntity, request: Request, db: AsyncSession = Depends(get_db) ): auth_service = AuthService(db) token = await auth_service.authenticate_user(user_data.login, user_data.password, request) response.set_cookie( key="users_access_token", value=token["access_token"], httponly=True, samesite="Lax", ) return token @router.get( "/sessions/", response_model=List[SessionEntity], summary="Get user sessions", description="Returns a list of active sessions for the current user", ) async def get_sessions( user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db) ): auth_service = AuthService(db) return await auth_service.get_user_sessions(user.id) @router.post( "/sessions/{session_id}/logout/", summary="Log out from a specific session", description="Deactivates a specific session by ID", ) async def logout_session( session_id: int, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db) ): auth_service = AuthService(db) await auth_service.deactivate_session(session_id, user.id) return {"message": "Session deactivated"} @router.post( "/sessions/logout_all/", summary="Log out from all sessions", description="Deactivates all sessions for the current user", ) async def logout_all_sessions( user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db) ): auth_service = AuthService(db) await auth_service.deactivate_all_sessions(user.id) return {"message": "All sessions deactivated"}