""" Effect presets CRUD operations. """ from __future__ import annotations import json import uuid from typing import List, Optional from sqlalchemy.orm import Session from sqlalchemy.exc import IntegrityError from ..utils.effects import validate_effects_chain from ..database import EffectPreset as DBEffectPreset from ..models import EffectPresetResponse, EffectPresetCreate, EffectPresetUpdate, EffectConfig def _preset_response(p: DBEffectPreset) -> EffectPresetResponse: """Convert a DB preset row to a Pydantic response.""" effects_chain = [EffectConfig(**e) for e in json.loads(p.effects_chain)] return EffectPresetResponse( id=p.id, name=p.name, description=p.description, effects_chain=effects_chain, is_builtin=p.is_builtin or False, created_at=p.created_at, ) def list_presets(db: Session) -> List[EffectPresetResponse]: """List all effect presets (built-in + user-created).""" presets = db.query(DBEffectPreset).order_by(DBEffectPreset.sort_order, DBEffectPreset.name).all() return [_preset_response(p) for p in presets] def get_preset(preset_id: str, db: Session) -> Optional[EffectPresetResponse]: """Get a preset by ID.""" p = db.query(DBEffectPreset).filter_by(id=preset_id).first() if not p: return None return _preset_response(p) def get_preset_by_name(name: str, db: Session) -> Optional[EffectPresetResponse]: """Get a preset by name.""" p = db.query(DBEffectPreset).filter_by(name=name).first() if not p: return None return _preset_response(p) def create_preset(data: EffectPresetCreate, db: Session) -> EffectPresetResponse: """Create a new user effect preset.""" chain_dicts = [e.model_dump() for e in data.effects_chain] error = validate_effects_chain(chain_dicts) if error: raise ValueError(error) # Check for duplicate name before insert existing = db.query(DBEffectPreset).filter_by(name=data.name).first() if existing: raise ValueError(f"A preset named '{data.name}' already exists") preset = DBEffectPreset( id=str(uuid.uuid4()), name=data.name, description=data.description, effects_chain=json.dumps(chain_dicts), is_builtin=False, ) db.add(preset) try: db.commit() except IntegrityError: db.rollback() raise ValueError(f"A preset named '{data.name}' already exists") db.refresh(preset) return _preset_response(preset) def update_preset(preset_id: str, data: EffectPresetUpdate, db: Session) -> Optional[EffectPresetResponse]: """Update a user effect preset. Cannot modify built-in presets.""" preset = db.query(DBEffectPreset).filter_by(id=preset_id).first() if not preset: return None if preset.is_builtin: raise ValueError("Cannot modify built-in presets") if data.name is not None: preset.name = data.name if data.description is not None: preset.description = data.description if data.effects_chain is not None: chain_dicts = [e.model_dump() for e in data.effects_chain] error = validate_effects_chain(chain_dicts) if error: raise ValueError(error) preset.effects_chain = json.dumps(chain_dicts) db.commit() db.refresh(preset) return _preset_response(preset) def delete_preset(preset_id: str, db: Session) -> bool: """Delete a user effect preset. Cannot delete built-in presets.""" preset = db.query(DBEffectPreset).filter_by(id=preset_id).first() if not preset: return False if preset.is_builtin: raise ValueError("Cannot delete built-in presets") db.delete(preset) db.commit() return True