0186fc8e70
This commit implements a simple movie database backend inspired by IMDb. It includes: - API endpoints for movies, actors, directors and genres - SQLAlchemy models with relationships - Alembic migrations - Pydantic schemas for request/response validation - Search and filtering functionality - Health check endpoint - Complete documentation
93 lines
2.2 KiB
Python
93 lines
2.2 KiB
Python
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 |