Spaces:
Sleeping
Sleeping
File size: 3,843 Bytes
3493993 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | from __future__ import annotations
from uuid import UUID
from fastapi import APIRouter, Request, Response, status
from app.brand.schemas import BrandKitCreate, BrandKitResponse
from app.security.errors import ForbiddenError
router = APIRouter(prefix="/v1/brand", tags=["brand"])
def _identity(request: Request) -> tuple[str, str]:
context = request.state.auth
if not context.workspace_id or not context.user_id:
raise ForbiddenError
return context.workspace_id, context.user_id
@router.post("", response_model=BrandKitResponse, status_code=status.HTTP_201_CREATED)
async def create_brand_kit(request: Request, payload: BrandKitCreate) -> BrandKitResponse:
workspace_id, user_id = _identity(request)
kit, version = await request.app.state.container.brand_kits.create_brand_kit(
workspace_id=workspace_id,
name=payload.name,
description=payload.description,
user_id=user_id,
)
latest = await request.app.state.container.brand_kits.get_brand_kit_version(
workspace_id=workspace_id,
brand_kit_id=kit.id,
user_id=user_id,
)
return BrandKitResponse(
id=kit.id,
name=kit.name,
description=kit.description,
status=kit.status,
is_default=kit.is_default,
active_version_id=latest.id,
created_at=kit.created_at.isoformat(),
updated_at=kit.updated_at.isoformat(),
)
@router.get("", response_model=list[BrandKitResponse])
async def list_brand_kits(request: Request) -> list[BrandKitResponse]:
workspace_id, user_id = _identity(request)
kits = await request.app.state.container.brand_kits.list_brand_kits(
workspace_id=workspace_id,
user_id=user_id,
)
response: list[BrandKitResponse] = []
for kit in kits:
try:
latest = await request.app.state.container.brand_kits.get_brand_kit_version(
workspace_id=workspace_id,
brand_kit_id=kit.id,
user_id=user_id,
)
active_version_id = latest.id
except Exception:
active_version_id = kit.active_version_id
response.append(BrandKitResponse(
id=kit.id,
name=kit.name,
description=kit.description,
status=kit.status,
is_default=kit.is_default,
active_version_id=active_version_id,
created_at=kit.created_at.isoformat(),
updated_at=kit.updated_at.isoformat(),
))
return response
@router.patch("/{brand_kit_id}", response_model=BrandKitResponse)
async def update_brand_kit(request: Request, brand_kit_id: UUID, payload: BrandKitCreate) -> BrandKitResponse:
workspace_id, user_id = _identity(request)
version = await request.app.state.container.brand_kits.update_brand_kit(
workspace_id=workspace_id,
brand_kit_id=str(brand_kit_id),
user_id=user_id,
data=payload.initial_version,
)
kit = await request.app.state.container.brand_kits.get_brand_kit(workspace_id, str(brand_kit_id), user_id=user_id)
return BrandKitResponse(
id=kit.id,
name=kit.name,
description=kit.description,
status=kit.status,
is_default=kit.is_default,
active_version_id=version.id,
created_at=kit.created_at.isoformat(),
updated_at=kit.updated_at.isoformat(),
)
@router.delete("/{brand_kit_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_brand_kit(request: Request, brand_kit_id: UUID) -> Response:
workspace_id, user_id = _identity(request)
await request.app.state.container.brand_kits.delete_brand_kit(
workspace_id=workspace_id,
brand_kit_id=str(brand_kit_id),
user_id=user_id,
)
return Response(status_code=status.HTTP_204_NO_CONTENT)
|