80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
from dataclasses import asdict, dataclass
|
|
from typing import Annotated, Literal
|
|
|
|
from litestar import Controller, Litestar, MediaType, Request, Response, get, put
|
|
from litestar.params import Parameter
|
|
from pydantic import UUID4
|
|
|
|
from secretsanta.domain import Group
|
|
from secretsanta.repo import S3Repo
|
|
from secretsanta.service import (
|
|
CreateGroupService,
|
|
GetGroupParticipantsService,
|
|
GetPairService,
|
|
)
|
|
from secretsanta.settings import get_config
|
|
|
|
|
|
@dataclass
|
|
class CreateGroup:
|
|
participants: list[str]
|
|
|
|
|
|
@dataclass
|
|
class PairResponse:
|
|
pair: str
|
|
|
|
|
|
class GroupController(Controller):
|
|
path = "api/v1/groups"
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.repo = S3Repo()
|
|
self.get_participants_service = GetGroupParticipantsService(self.repo)
|
|
self.create_group_service = CreateGroupService(self.repo)
|
|
self.get_pair_service = GetPairService(self.repo)
|
|
|
|
@staticmethod
|
|
def _group2urls(participants: list[str], url: str) -> list[str]:
|
|
return [f"{url}/pair/{participant_name}" for participant_name in participants]
|
|
|
|
@get("/{group_uuid:uuid}")
|
|
async def get_participants(self, request: Request, group_uuid: UUID4) -> list[str]:
|
|
participants = await self.get_participants_service.run(group_uuid)
|
|
return self._group2urls(participants, request.url)
|
|
|
|
@put("/{group_uuid:uuid}")
|
|
async def create_group(
|
|
self,
|
|
request: Request,
|
|
group_uuid: UUID4,
|
|
data: CreateGroup,
|
|
) -> list[str]:
|
|
group = await self.create_group_service.run(
|
|
Group(uuid=group_uuid, **asdict(data)),
|
|
)
|
|
return self._group2urls(group.participants, request.url)
|
|
|
|
@get("/{group_uuid:uuid}/pair/{participant_name:str}")
|
|
async def get_pair(
|
|
self,
|
|
group_uuid: UUID4,
|
|
participant_name: str,
|
|
response_type: Annotated[
|
|
Literal["json", "plain"],
|
|
Parameter(query="type"),
|
|
] = "plain",
|
|
) -> Response[str] | PairResponse:
|
|
pair = await self.get_pair_service.run(
|
|
group_uuid=group_uuid,
|
|
participant=participant_name,
|
|
)
|
|
if response_type == "json":
|
|
return PairResponse(pair=pair)
|
|
|
|
return Response(content=pair, media_type=MediaType.TEXT)
|
|
|
|
|
|
config = get_config()
|
|
app = Litestar(route_handlers=[GroupController], debug=config.environment == "local")
|