Implement CRUD for Movies

src/routes/movies.py

@@ -1,12 +1,184 @@
+from math import ceil
+from typing import Dict, Union, List, Optional, Annotated
+
 from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi import status
+from pydantic_core.core_schema import AnySchema
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session, joinedload
+from sqlalchemy import update
 
 from database import get_db
 from database.models import MovieModel, CountryModel, GenreModel, ActorModel, LanguageModel
 
+from schemas.movies import (
+    MovieListResponseSchema,
+    MovieDetailsResponseSchema,
+    MovieAddSchema,
+    MovieUpdateSchema,
+    MovieUpdateResponseSchema
+)
+
 
 router = APIRouter()
 
 
-# Write your code here
+@router.get(
+    "/movies/",
+    response_model=Dict[str, Union[List[MovieListResponseSchema], Optional[str], int]]
+)
+def movies(
+        db: Session = Depends(get_db),
+        page: Annotated[int, Query(..., title="The page number to fetch", ge=1)] = 1,
+        per_page: Annotated[int, Query(
+            ..., title="Number of movies to fetch per page", ge=1, le=20
+        )] = 10
+) -> Dict[str, Union[List[MovieListResponseSchema], Optional[str], int]]:
+
+    data = db.query(MovieModel).order_by(MovieModel.id.desc()).all()
+
+    total_pages = ceil(len(data) / per_page)
+
+    start = (page - 1) * per_page
+    end = start + per_page
+
+    items = len(data)
+    if not items or page > total_pages:
+        raise HTTPException(status_code=404, detail="No movies found.")
+
+    if end > items:
+        next_page = None
+    else:
+        next_page = f"/theater/movies/?page={page + 1}&per_page={per_page}"
+    if page == 1:
+        prev_page = None
+    else:
+        prev_page = f"/theater/movies/?page={page - 1}&per_page={per_page}"
+
+    response = {
+        "movies": data[start:end],
+        "prev_page": prev_page,
+        "next_page": next_page,
+        "total_pages": total_pages,
+        "total_items": items
+    }
+
+    return response
+
+
+def add_related_entities(entities: List[str], model, db) -> List[AnySchema]:
+    existing_entities = []
+    for entity in entities:
+        existing_entity = db.query(model).filter(model.name == entity).first()
+        if not existing_entity:
+            existing_entity = model(name=entity)
+            db.add(existing_entity)
+            db.commit()
+            db.refresh(existing_entity)
+        existing_entities.append(existing_entity)
+    return existing_entities
+
+
+@router.post(
+    "/movies/",
+    response_model=MovieDetailsResponseSchema,
+    status_code=status.HTTP_201_CREATED
+)
+def add_movie(movie: MovieAddSchema, db: Session = Depends(get_db)):
+    required_fields = ["name", "date", "score", "overview", "status", "budget", "revenue", "country"]
+    for field in required_fields:
+        if not getattr(movie, field, None):
+            raise HTTPException(status_code=400, detail="Invalid input data.")
+
+    country = db.query(CountryModel).filter(CountryModel.code == movie.country).first()
+    if not country:
+        country = CountryModel(code=movie.country)
+
+    existing_movie = db.query(MovieModel).filter(
+        (MovieModel.name == movie.name) & (MovieModel.date == movie.date)
+    ).first()
+    if existing_movie:
+        raise HTTPException(
+            status_code=409, detail=f"A movie with the name '{movie.name}' "
+                                    f"and release date '{movie.date}' already exists."
+        )
+
+    db_movie = MovieModel(
+        name=movie.name,
+        date=movie.date,
+        score=movie.score,
+        overview=movie.overview,
+        status=movie.status,
+        budget=movie.budget,
+        revenue=movie.revenue,
+        country=country,
+    )
+
+    db.add(db_movie)
+    db.commit()
+    db.refresh(db_movie)
+
+    if movie.genres:
+        db_movie.genres = add_related_entities(movie.genres, GenreModel, db)
+
+    if movie.actors:
+        db_movie.actors = add_related_entities(movie.actors, ActorModel, db)
+
+    if movie.languages:
+        db_movie.languages = add_related_entities(movie.languages, LanguageModel, db)
+
+    db.commit()
+    db.refresh(db_movie)
+
+    return db_movie
+
+
+@router.get("/movies/{movie_id}/", response_model=MovieDetailsResponseSchema)
+def get_movie(movie_id: int, db: Session = Depends(get_db)):
+    movie = db.query(MovieModel).filter(MovieModel.id == movie_id).first()
+    if not movie:
+        raise HTTPException(status_code=404, detail="Movie with the given ID was not found.")
+    return movie
+
+
+@router.delete("/movies/{movie_id}/", status_code=status.HTTP_204_NO_CONTENT)
+def delete_movie(movie_id: int, db: Session = Depends(get_db)):
+    movie = db.query(MovieModel).filter(MovieModel.id == movie_id).first()
+    if not movie:
+        raise HTTPException(status_code=404, detail="Movie with the given ID was not found.")
+    db.delete(movie)
+    db.commit()
+
+
+@router.patch(
+    "/movies/{movie_id}/",
+    response_model=MovieUpdateResponseSchema,
+    status_code=status.HTTP_200_OK,
+)
+def update_movie(movie_update: MovieUpdateSchema, movie_id: int, db: Session = Depends(get_db)):
+    existing_movie = db.query(MovieModel).filter(MovieModel.id == movie_id).first()
+    if not existing_movie:
+        raise HTTPException(status_code=404, detail="Movie with the given ID was not found.")
+
+    new_attributes = movie_update.model_dump(exclude_unset=True)
+
+    if 'date' not in new_attributes:
+        new_attributes['date'] = existing_movie.date
+
+    try:
+        db.execute(
+            update(MovieModel)
+            .where(MovieModel.id == movie_id)
+            .values(new_attributes)
+        )
+
+        db.commit()
+        db.refresh(existing_movie)
+
+    except ValueError:
+        db.rollback()
+        existing_movie.detail = "Invalid input data."
+        raise HTTPException(status_code=400, detail="Invalid input data.")
+
+    existing_movie.detail = "Movie updated successfully."
+    return existing_movie

src/schemas/movies.py

@@ -1 +1,138 @@
-# Write your code here
+from enum import Enum
+from re import match
+
+from dateutil.relativedelta import relativedelta
+from pydantic import BaseModel, field_validator
+from typing import List, Optional
+from datetime import date
+
+
+class MovieStatusEnum(str, Enum):
+    released = "Released"
+    post_production = "Post Production"
+    in_production = "In Production"
+
+    @field_validator("status")
+    def validate_status_acceptable(cls, v):
+        if v not in ["Released", "Post Production", "In Production"]:
+            raise ValueError("status must be one of (Released | Post Production | In Production)")
+        return v
+
+
+class CountrySchema(BaseModel):
+    id: int
+    code: str
+    name: Optional[str] = None
+
+    # validator for checking format ISO 3166-1 alpha-2
+    @field_validator("code")
+    def validate_code_iso_3166_1(cls, v):
+        if not match(r"^[A-Z]{2}$", v):
+            raise ValueError("code must be a valid ISO 3166-1 alpha-2 code")
+        return v
+
+
+class GenreSchema(BaseModel):
+    id: int
+    name: str
+
+
+class ActorSchema(BaseModel):
+    id: int
+    name: str
+
+
+class LanguageSchema(BaseModel):
+    id: int
+    name: str
+
+
+class MovieBase(BaseModel):
+    name: str
+    date: date
+    score: float
+    overview: str
+    status: MovieStatusEnum
+    budget: float
+    revenue: float
+
+    class Config:
+        from_attributes = True
+
+    @field_validator("name")
+    def validate_name_length(cls, v):
+        if len(v) > 255:
+            raise ValueError("The name must not exceed 255 characters.")
+        return v
+
+    @field_validator("date")
+    def validate_date_not_more_1_year(cls, v):
+        if v > date.today() + relativedelta(years=1):
+            raise ValueError("The date must not be more than one year in the future.")
+        return v
+
+    @field_validator("score")
+    def validate_score_0_100(cls, v):
+        if not 0 <= v <= 100:
+            raise ValueError("The score must be between 0 and 100.")
+        return v
+
+    @field_validator("budget")
+    def validate_budget_positive(cls, v):
+        if not v >= 0:
+            raise ValueError("The budget must be non-negative.")
+        return v
+
+    @field_validator("revenue")
+    def validate_revenue_positive(cls, v):
+        if not v >= 0:
+            raise ValueError("The revenue must be non-negative.")
+        return v
+
+
+class MovieListResponseSchema(BaseModel):
+    id: int
+    name: str
+    date: date
+    score: float
+    overview: str
+
+
+class PageSchema(BaseModel):
+    items: List[MovieListResponseSchema]
+    total: int
+    prev_page: Optional[str] = None
+    next_page: Optional[str]
+
+
+class MovieAddSchema(MovieBase):
+    country: str
+    genres: List[str] = None
+    actors: List[str] = None
+    languages: List[str] = None
+
+
+class MovieDetailsResponseSchema(MovieBase):
+    id: int
+    country: CountrySchema
+    genres: List[GenreSchema]
+    actors: List[ActorSchema]
+    languages: List[LanguageSchema]
+
+
+class MovieUpdateSchema(MovieBase):
+    name: Optional[str] = None
+    date: Optional[date] = None
+    score: Optional[float] = None
+    overview: Optional[str] = None
+    status: Optional[MovieStatusEnum] = None
+    budget: Optional[float] = None
+    revenue: Optional[float] = None
+
+    class Config:
+        from_attributes = True
+        exclude = {"country", "genres", "actors", "languages"}
+
+
+class MovieUpdateResponseSchema(MovieDetailsResponseSchema):
+    detail: str