Daniil
71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from sqlalchemy import Select, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from cpv3.modules.captions.models import CaptionPreset
|
|
from cpv3.modules.captions.schemas import CaptionPresetCreate, CaptionPresetUpdate
|
|
|
|
|
|
class CaptionPresetRepository:
|
|
"""Repository for CaptionPreset database operations."""
|
|
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self._session = session
|
|
|
|
async def list_all_for_user(self, user_id: uuid.UUID) -> list[CaptionPreset]:
|
|
"""Return system presets + user's own presets."""
|
|
stmt: Select[tuple[CaptionPreset]] = (
|
|
select(CaptionPreset)
|
|
.where(CaptionPreset.is_active.is_(True))
|
|
.where(
|
|
or_(
|
|
CaptionPreset.is_system.is_(True),
|
|
CaptionPreset.user_id == user_id,
|
|
)
|
|
)
|
|
.order_by(CaptionPreset.is_system.desc(), CaptionPreset.created_at.desc())
|
|
)
|
|
result = await self._session.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_by_id(self, preset_id: uuid.UUID) -> CaptionPreset | None:
|
|
result = await self._session.execute(
|
|
select(CaptionPreset)
|
|
.where(CaptionPreset.id == preset_id)
|
|
.where(CaptionPreset.is_active.is_(True))
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def create(
|
|
self, *, user_id: uuid.UUID | None, data: CaptionPresetCreate
|
|
) -> CaptionPreset:
|
|
preset = CaptionPreset(
|
|
user_id=user_id,
|
|
name=data.name,
|
|
description=data.description,
|
|
is_system=user_id is None,
|
|
style_config=data.style_config.model_dump(mode="json"),
|
|
)
|
|
self._session.add(preset)
|
|
await self._session.commit()
|
|
await self._session.refresh(preset)
|
|
return preset
|
|
|
|
async def update(self, preset: CaptionPreset, data: CaptionPresetUpdate) -> CaptionPreset:
|
|
for key, value in data.model_dump(exclude_unset=True).items():
|
|
if value is not None:
|
|
if key == "style_config":
|
|
setattr(preset, key, value.model_dump(mode="json"))
|
|
else:
|
|
setattr(preset, key, value)
|
|
await self._session.commit()
|
|
await self._session.refresh(preset)
|
|
return preset
|
|
|
|
async def deactivate(self, preset: CaptionPreset) -> None:
|
|
preset.is_active = False
|
|
await self._session.commit()
|