from typing import List, Optional, Dict, Any from sqlalchemy.orm import Session from app.models.genre import Genre from app.models.movie_genre import MovieGenre from app.api.schemas.genre import GenreCreate, GenreUpdate def get(db: Session, genre_id: int) -> Optional[Genre]: """ Get a genre by ID. """ return db.query(Genre).filter(Genre.id == genre_id).first() def get_by_name(db: Session, name: str) -> Optional[Genre]: """ Get a genre by name. """ return db.query(Genre).filter(Genre.name == name).first() def get_multi( db: Session, skip: int = 0, limit: int = 100, filters: Optional[Dict[str, Any]] = None ) -> List[Genre]: """ Get multiple genres with pagination and optional filters. """ query = db.query(Genre) # Apply filters if provided if filters: if name := filters.get("name"): query = query.filter(Genre.name.ilike(f"%{name}%")) if movie_id := filters.get("movie_id"): query = query.join(MovieGenre).filter(MovieGenre.movie_id == movie_id) # Sort by name query = query.order_by(Genre.name) # Get total count before pagination total = query.count() return query.offset(skip).limit(limit).all(), total def create(db: Session, genre_in: GenreCreate) -> Genre: """ Create a new genre. """ db_genre = Genre(**genre_in.dict()) db.add(db_genre) db.commit() db.refresh(db_genre) return db_genre def update( db: Session, db_genre: Genre, genre_in: GenreUpdate ) -> Genre: """ Update a genre. """ update_data = genre_in.dict(exclude_unset=True) for field, value in update_data.items(): setattr(db_genre, field, value) db.commit() db.refresh(db_genre) return db_genre def delete(db: Session, genre_id: int) -> bool: """ Delete a genre by ID. """ db_genre = db.query(Genre).filter(Genre.id == genre_id).first() if db_genre: # Delete associated movie-genre relationships db.query(MovieGenre).filter(MovieGenre.genre_id == genre_id).delete() # Delete the genre db.delete(db_genre) db.commit() return True return False