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
85 lines
2.1 KiB
Python
85 lines
2.1 KiB
Python
from typing import List, Optional, Dict, Any
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.director import Director
|
|
from app.models.movie import Movie
|
|
from app.api.schemas.director import DirectorCreate, DirectorUpdate
|
|
|
|
|
|
def get(db: Session, director_id: int) -> Optional[Director]:
|
|
"""
|
|
Get a director by ID.
|
|
"""
|
|
return db.query(Director).filter(Director.id == director_id).first()
|
|
|
|
|
|
def get_multi(
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
filters: Optional[Dict[str, Any]] = None
|
|
) -> List[Director]:
|
|
"""
|
|
Get multiple directors with pagination and optional filters.
|
|
"""
|
|
query = db.query(Director)
|
|
|
|
# Apply filters if provided
|
|
if filters:
|
|
if name := filters.get("name"):
|
|
query = query.filter(Director.name.ilike(f"%{name}%"))
|
|
|
|
# Sort by name
|
|
query = query.order_by(Director.name)
|
|
|
|
# Get total count before pagination
|
|
total = query.count()
|
|
|
|
return query.offset(skip).limit(limit).all(), total
|
|
|
|
|
|
def create(db: Session, director_in: DirectorCreate) -> Director:
|
|
"""
|
|
Create a new director.
|
|
"""
|
|
db_director = Director(**director_in.dict())
|
|
db.add(db_director)
|
|
db.commit()
|
|
db.refresh(db_director)
|
|
return db_director
|
|
|
|
|
|
def update(
|
|
db: Session,
|
|
db_director: Director,
|
|
director_in: DirectorUpdate
|
|
) -> Director:
|
|
"""
|
|
Update a director.
|
|
"""
|
|
update_data = director_in.dict(exclude_unset=True)
|
|
|
|
for field, value in update_data.items():
|
|
setattr(db_director, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_director)
|
|
return db_director
|
|
|
|
|
|
def delete(db: Session, director_id: int) -> bool:
|
|
"""
|
|
Delete a director by ID.
|
|
"""
|
|
db_director = db.query(Director).filter(Director.id == director_id).first()
|
|
if db_director:
|
|
# Unlink movies from this director
|
|
db.query(Movie).filter(Movie.director_id == director_id).update(
|
|
{Movie.director_id: None}, synchronize_session=False
|
|
)
|
|
|
|
# Delete the director
|
|
db.delete(db_director)
|
|
db.commit()
|
|
return True
|
|
return False |