forked from HSE_team/BetterCallPraskovia
141 lines
5.0 KiB
Python
141 lines
5.0 KiB
Python
"""
|
||
API роутеры для работы с коллекциями
|
||
"""
|
||
from uuid import UUID
|
||
from fastapi import APIRouter, status, Depends, Request
|
||
from fastapi.responses import JSONResponse
|
||
from typing import List, Annotated
|
||
from dishka.integrations.fastapi import FromDishka, inject
|
||
from src.domain.repositories.user_repository import IUserRepository
|
||
from src.presentation.middleware.auth_middleware import get_current_user
|
||
from src.presentation.schemas.collection_schemas import (
|
||
CollectionCreate,
|
||
CollectionUpdate,
|
||
CollectionResponse,
|
||
CollectionAccessGrant,
|
||
CollectionAccessResponse
|
||
)
|
||
from src.application.use_cases.collection_use_cases import CollectionUseCases
|
||
from src.domain.entities.user import User
|
||
|
||
router = APIRouter(prefix="/collections", tags=["collections"])
|
||
|
||
|
||
@router.post("", response_model=CollectionResponse, status_code=status.HTTP_201_CREATED)
|
||
@inject
|
||
async def create_collection(
|
||
collection_data: CollectionCreate,
|
||
request: Request,
|
||
user_repo: Annotated[IUserRepository, FromDishka()],
|
||
use_cases: Annotated[CollectionUseCases, FromDishka()]
|
||
):
|
||
"""Создать коллекцию"""
|
||
current_user = await get_current_user(request, user_repo)
|
||
collection = await use_cases.create_collection(
|
||
name=collection_data.name,
|
||
owner_id=current_user.user_id,
|
||
description=collection_data.description,
|
||
is_public=collection_data.is_public
|
||
)
|
||
return CollectionResponse.from_entity(collection)
|
||
|
||
|
||
@router.get("/{collection_id}", response_model=CollectionResponse)
|
||
@inject
|
||
async def get_collection(
|
||
collection_id: UUID,
|
||
use_cases: Annotated[CollectionUseCases, FromDishka()]
|
||
):
|
||
"""Получить коллекцию по ID"""
|
||
collection = await use_cases.get_collection(collection_id)
|
||
return CollectionResponse.from_entity(collection)
|
||
|
||
|
||
@router.put("/{collection_id}", response_model=CollectionResponse)
|
||
@inject
|
||
async def update_collection(
|
||
collection_id: UUID,
|
||
collection_data: CollectionUpdate,
|
||
request: Request,
|
||
user_repo: Annotated[IUserRepository, FromDishka()],
|
||
use_cases: Annotated[CollectionUseCases, FromDishka()]
|
||
):
|
||
"""Обновить коллекцию"""
|
||
current_user = await get_current_user(request, user_repo)
|
||
collection = await use_cases.update_collection(
|
||
collection_id=collection_id,
|
||
user_id=current_user.user_id,
|
||
name=collection_data.name,
|
||
description=collection_data.description,
|
||
is_public=collection_data.is_public
|
||
)
|
||
return CollectionResponse.from_entity(collection)
|
||
|
||
|
||
@router.delete("/{collection_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||
@inject
|
||
async def delete_collection(
|
||
collection_id: UUID,
|
||
request: Request,
|
||
user_repo: Annotated[IUserRepository, FromDishka()],
|
||
use_cases: Annotated[CollectionUseCases, FromDishka()]
|
||
):
|
||
"""Удалить коллекцию"""
|
||
current_user = await get_current_user(request, user_repo)
|
||
await use_cases.delete_collection(collection_id, current_user.user_id)
|
||
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT, content=None)
|
||
|
||
|
||
@router.get("", response_model=List[CollectionResponse])
|
||
@inject
|
||
async def list_collections(
|
||
request: Request,
|
||
user_repo: Annotated[IUserRepository, FromDishka()],
|
||
use_cases: Annotated[CollectionUseCases, FromDishka()],
|
||
skip: int = 0,
|
||
limit: int = 100
|
||
):
|
||
"""Получить список коллекций, доступных пользователю"""
|
||
current_user = await get_current_user(request, user_repo)
|
||
collections = await use_cases.list_user_collections(
|
||
user_id=current_user.user_id,
|
||
skip=skip,
|
||
limit=limit
|
||
)
|
||
return [CollectionResponse.from_entity(c) for c in collections]
|
||
|
||
|
||
@router.post("/{collection_id}/access", response_model=CollectionAccessResponse, status_code=status.HTTP_201_CREATED)
|
||
@inject
|
||
async def grant_access(
|
||
collection_id: UUID,
|
||
access_data: CollectionAccessGrant,
|
||
request: Request,
|
||
user_repo: Annotated[IUserRepository, FromDishka()],
|
||
use_cases: Annotated[CollectionUseCases, FromDishka()]
|
||
):
|
||
"""Предоставить доступ пользователю к коллекции"""
|
||
current_user = await get_current_user(request, user_repo)
|
||
access = await use_cases.grant_access(
|
||
collection_id=collection_id,
|
||
user_id=access_data.user_id,
|
||
owner_id=current_user.user_id
|
||
)
|
||
return CollectionAccessResponse.from_entity(access)
|
||
|
||
|
||
@router.delete("/{collection_id}/access/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||
@inject
|
||
async def revoke_access(
|
||
collection_id: UUID,
|
||
user_id: UUID,
|
||
request: Request,
|
||
user_repo: Annotated[IUserRepository, FromDishka()],
|
||
use_cases: Annotated[CollectionUseCases, FromDishka()]
|
||
):
|
||
"""Отозвать доступ пользователя к коллекции"""
|
||
current_user = await get_current_user(request, user_repo)
|
||
await use_cases.revoke_access(collection_id, user_id, current_user.user_id)
|
||
return JSONResponse(status_code=status.HTTP_204_NO_CONTENT, content=None)
|
||
|