andrei
dc47e4b003
Добавлена функциональность управления сессиями пользователей, включая создание сессий при входе, получение списка активных сессий, деактивацию отдельных сессий и деактивацию всех сессий пользователя.
83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
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"}
|