74 lines
2.9 KiB
Python
Executable File
74 lines
2.9 KiB
Python
Executable File
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from rehearsalhub.db.models import Band
|
|
from rehearsalhub.repositories.band import BandRepository
|
|
from rehearsalhub.schemas.band import BandCreate, BandReadWithMembers
|
|
from rehearsalhub.storage.nextcloud import NextcloudClient
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class BandService:
|
|
def __init__(self, session: AsyncSession, storage: NextcloudClient | None = None) -> None:
|
|
self._repo = BandRepository(session)
|
|
self._storage = storage
|
|
|
|
async def create_band(
|
|
self,
|
|
data: BandCreate,
|
|
creator_id: uuid.UUID,
|
|
creator: object | None = None,
|
|
) -> Band:
|
|
if await self._repo.get_by_slug(data.slug):
|
|
raise ValueError(f"Slug already taken: {data.slug}")
|
|
|
|
nc_folder = (data.nc_base_path or f"bands/{data.slug}/").strip("/") + "/"
|
|
storage = NextcloudClient.for_member(creator) if creator else self._storage
|
|
|
|
if data.nc_base_path:
|
|
# User explicitly specified a folder — verify it actually exists in NC.
|
|
log.info("Checking NC folder existence: %s", nc_folder)
|
|
try:
|
|
await storage.get_file_metadata(nc_folder.rstrip("/"))
|
|
except Exception as exc:
|
|
log.warning("NC folder '%s' not accessible: %s", nc_folder, exc)
|
|
raise LookupError(f"Nextcloud folder '{nc_folder}' not found or not accessible")
|
|
else:
|
|
# Auto-generated path — create it (idempotent MKCOL).
|
|
log.info("Creating NC folder: %s", nc_folder)
|
|
try:
|
|
await storage.create_folder(nc_folder)
|
|
except Exception as exc:
|
|
# Not fatal — NC may be temporarily unreachable during dev/test.
|
|
log.warning("Could not create NC folder '%s': %s", nc_folder, exc)
|
|
|
|
band = await self._repo.create(
|
|
name=data.name,
|
|
slug=data.slug,
|
|
genre_tags=data.genre_tags,
|
|
nc_folder_path=nc_folder,
|
|
)
|
|
await self._repo.add_member(band.id, creator_id, role="admin")
|
|
log.info("Created band '%s' (slug=%s, nc_folder=%s)", data.name, data.slug, nc_folder)
|
|
return band
|
|
|
|
async def get_band_with_members(self, band_id: uuid.UUID) -> Band | None:
|
|
return await self._repo.get_with_members(band_id)
|
|
|
|
async def assert_membership(self, band_id: uuid.UUID, member_id: uuid.UUID) -> str:
|
|
"""Returns the member's role or raises PermissionError."""
|
|
role = await self._repo.get_member_role(band_id, member_id)
|
|
if role is None:
|
|
raise PermissionError("Not a member of this band")
|
|
return role
|
|
|
|
async def assert_admin(self, band_id: uuid.UUID, member_id: uuid.UUID) -> None:
|
|
role = await self.assert_membership(band_id, member_id)
|
|
if role != "admin":
|
|
raise PermissionError("Admin role required")
|