Feat/movie 02 create endpoints

pyproject.toml

@@ -72,7 +72,7 @@ ignore = [
     "N802",
     "B008",
     "F821",
-    "I001"
+    "I001",
 ]
 
 [tool.ruff.lint.pydocstyle]

src/database/crud/movies.py

@@ -0,0 +1,193 @@
+from typing import List, Optional, Type, cast
+
+from sqlalchemy.orm import Session, joinedload
+
+from database import Certification, Director, Genre, Movie, Star, User
+from database.models.movies import FavoriteMovie, MovieLike
+from schemas.movies import GenresSchema, MovieCreateSchema, MovieSortEnum, StarsSchema
+
+
+def get_movies_paginated(page: int, per_page: int,db: Session) -> [int, list[Movie]]:
+    offset = (page - 1) * per_page
+
+    query = db.query(Movie).order_by()
+
+    order_by = Movie.default_order_by()
+    if order_by:
+        query = query.order_by(*order_by)
+
+    total_items = query.count()
+    movies = query.offset(offset).limit(per_page).all()
+
+    return total_items, movies
+
+
+def filter_movies(
+        db: Session,
+        filters: dict[str, str],
+        sort_by: Optional[MovieSortEnum] = None
+) -> list[Movie]:
+    query = db.query(Movie)
+
+    if filters.get("name"):
+        query = query.filter(Movie.name.ilike(f"%{filters['name']}%"))
+    if filters.get("year"):
+        query = query.filter(Movie.year == filters["year"])
+    if filters.get("min_imdb"):
+        query = query.filter(Movie.imdb >= filters["min_imdb"])
+    if filters.get("max_imdb"):
+        query = query.filter(Movie.imdb <= filters["max_imdb"])
+    if filters.get("min_votes"):
+        query = query.filter(Movie.votes >= filters["min_votes"])
+    if filters.get("max_votes"):
+        query = query.filter(Movie.votes <= filters["max_votes"])
+    if filters.get("min_price"):
+        query = query.filter(Movie.price >= filters["min_price"])
+    if filters.get("max_price"):
+        query = query.filter(Movie.price <= filters["max_price"])
+
+    if sort_by:
+        if sort_by == MovieSortEnum.PRICE_ASC:
+            query = query.order_by(Movie.price)
+        elif sort_by == MovieSortEnum.PRICE_DESC:
+            query = query.order_by(Movie.price.desc())
+        elif sort_by == MovieSortEnum.RELEASE_YEAR_ASC:
+            query = query.order_by(Movie.year)
+        elif sort_by == MovieSortEnum.RELEASE_YEAR_DESC:
+            query = query.order_by(Movie.year.desc())
+        elif sort_by == MovieSortEnum.VOTES_ASC:
+            query = query.order_by(Movie.votes)
+        elif sort_by == MovieSortEnum.VOTES_DESC:
+            query = query.order_by(Movie.votes.desc())
+        elif sort_by == MovieSortEnum.IMDb_ASC:
+            query = query.order_by(Movie.imdb)
+        elif sort_by == MovieSortEnum.IMDb_DESC:
+            query = query.order_by(Movie.imdb.desc())
+
+    return cast(List[Movie], query.all())
+
+def get_detail_movies_by_id(db: Session, movie_id: int) -> Movie | None:
+    return (
+        db.query(Movie)
+        .options(
+            joinedload(Movie.certification),
+            joinedload(Movie.genres),
+            joinedload(Movie.stars),
+            joinedload(Movie.directors),
+        )
+        .filter(Movie.id == movie_id)
+        .first()
+    )
+
+def get_movie_by_id(db: Session, movie_id: int) -> Movie | None:
+    return db.query(Movie).filter(Movie.id == movie_id).first()
+
+
+def get_movie_by_name(db: Session, movie_data: MovieCreateSchema) -> Movie | None:
+    return (
+        db.query(Movie).filter(
+            Movie.name == movie_data.name
+        ).first()
+    )
+
+def get_certification_by_name(
+        db: Session,
+        movie_data: MovieCreateSchema
+) -> Certification | None:
+    return db.query(Certification).filter_by(name=movie_data.certification).first()
+
+def get_or_create_certification(
+        db: Session,
+        movie_data: MovieCreateSchema
+) -> Certification:
+    certification = get_certification_by_name(db, movie_data)
+    if not certification:
+        certification = Certification(name=movie_data.certification)
+        db.add(certification)
+        db.commit()
+        db.refresh(certification)
+
+    return certification
+
+def get_genre_by_id(db: Session, genre_id: int) -> Genre | None:
+    return db.query(Genre).filter_by(id=genre_id).first()
+
+def get_genre_by_name(db: Session, genres_data: GenresSchema) -> Genre | None:
+    return db.query(Genre).filter_by(name=genres_data.name).first()
+
+def get_all_genres(db: Session) -> list[Genre]:
+    return cast(List[Genre], db.query(Genre).all())
+
+def get_or_create_genres(
+        db: Session,
+        movie_data: MovieCreateSchema
+) -> list[Genre | Type[Genre]]:
+    genres = []
+
+    for genre_name in movie_data.genres:
+        genre = db.query(Genre).filter_by(name=genre_name).first()
+        if not genre:
+            genre = Genre(name=genre_name)
+            db.add(genre)
+            db.flush()
+        genres.append(genre)
+
+    return genres
+
+def get_star_by_name(db:Session, stars_data: StarsSchema) -> Star | None:
+    return db.query(Star).filter_by(name=stars_data.name).first()
+
+def get_star_by_id(db: Session, star_id: int) -> Star | None:
+    return db.query(Star).filter_by(id=star_id).first()
+
+def get_all_stars(db: Session) -> list[Star]:
+    return cast(List[Star], db.query(Star).all())
+
+def get_or_create_stars(
+        db: Session,
+        movie_data: MovieCreateSchema
+) -> list[Star | Type[Star]]:
+    stars = []
+
+    for star_name in movie_data.stars:
+        star = db.query(Star).filter_by(name=star_name).first()
+        if not star:
+            star = Star(name=star_name)
+            db.add(star)
+            db.flush()
+        stars.append(star)
+
+    return stars
+
+def get_or_create_directors(
+        db: Session,
+        movie_data: MovieCreateSchema
+) -> list[Director | Type[Director]]:
+    directors = []
+
+    for director_name in movie_data.directors:
+        director = db.query(Director).filter_by(name=director_name).first()
+        if not director:
+            director = Director(name=director_name)
+            db.add(director)
+            db.flush()
+        directors.append(director)
+
+    return directors
+
+def get_user_by_id(db: Session, user_id: int) -> User | None:
+    return db.query(User).filter_by(id=user_id).first()
+
+def get_liked_movie(db: Session, movie: Movie, user: User) -> MovieLike | None:
+    return (
+        db.query(MovieLike).filter_by(
+            movie_id=movie.id, user_id=user.id
+        ).first()
+    )
+
+def get_favourite_movie(db: Session, movie: Movie, user: User) -> FavoriteMovie | None:
+    return (
+        db.query(FavoriteMovie).filter_by(
+            movie_id=movie.id, user_id=user.id
+        ).first()
+    )

src/database/migrations/env.py

@@ -7,7 +7,7 @@
     movies,
     orders,
     payments,
-    shoping_cart,
+    shopping_cart,
 )
 from database.models.base import Base
 from database.session_postgresql import postgresql_engine

src/database/models/accounts.py

@@ -91,6 +91,12 @@ class User(Base):
     cart: Mapped["Cart"] = relationship("Cart", back_populates="user", uselist=False)
     orders: Mapped[list["Order"]] = relationship("Order", back_populates="user")
     payments: Mapped[list["Payment"]] = relationship("Payment", back_populates="user")
+    likes: Mapped[list["MovieLike"]] = relationship(
+        "MovieLike", back_populates="user", cascade="all, delete-orphan"
+    )
+    favorites: Mapped[list["FavoriteMovie"]] = relationship(
+        "FavoriteMovie", back_populates="user", cascade="all, delete-orphan"
+    )
 
     def __repr__(self) -> str:
         return f"<User(id={self.id}, email={self.email}, is_active={self.is_active})>"

src/database/models/movies.py

@@ -3,11 +3,14 @@
 
 from sqlalchemy import (
     DECIMAL,
+    TIMESTAMP,
+    Boolean,
     Float,
     ForeignKey,
     String,
     Text,
     UniqueConstraint,
+    func
 )
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
@@ -142,10 +145,60 @@ class Movie(Base):
     order_items: Mapped[list["OrderItem"]] = relationship(
         "OrderItem", back_populates="movie"
     )
+    likes: Mapped[list["MovieLike"]] = relationship(
+        "MovieLike", back_populates="movie", cascade="all, delete-orphan"
+    )
+
+    favorites: Mapped[list["FavoriteMovie"]] = relationship(
+        "FavoriteMovie", back_populates="movie", cascade="all, delete-orphan"
+    )
 
     __table_args__ = (
         UniqueConstraint("name", "year", "time", name="unique_movie_constraint"),
     )
 
     def __repr__(self) -> str:
         return f"<Movie (name='{self.name}', imdb='{self.imdb}', time='{self.time}')>"
+
+
+
+class MovieLike(Base):
+    __tablename__ = "movie_likes"
+
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), primary_key=True)
+    movie_id: Mapped[int] = mapped_column(ForeignKey("movies.id"), primary_key=True)
+    is_liked: Mapped[bool] = mapped_column(Boolean, nullable=False)
+    created_at: Mapped[TIMESTAMP] = mapped_column(
+        TIMESTAMP,
+        nullable=False,
+        server_default=func.now()
+    )
+
+    user: Mapped["User"] = relationship("User", back_populates="likes")
+    movie: Mapped["Movie"] = relationship("Movie", back_populates="likes")
+
+    def __repr__(self) -> str:
+        return (
+            f"<MovieLike (user_id='{self.user_id}', "
+            f"movie_id='{self.movie_id}', "
+            f"is_liked='{self.is_liked}')>"
+        )
+
+
+class FavoriteMovie(Base):
+    __tablename__ = "favorite_movies"
+
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), primary_key=True)
+    movie_id: Mapped[int] = mapped_column(ForeignKey("movies.id"), primary_key=True)
+    is_favorited: Mapped[bool] = mapped_column(Boolean, nullable=False)
+    created_at: Mapped[TIMESTAMP] = mapped_column(
+        TIMESTAMP,
+        nullable=False,
+        server_default=func.now()
+    )
+
+    user: Mapped["User"] = relationship("User", back_populates="favorites")
+    movie: Mapped["Movie"] = relationship("Movie", back_populates="favorites")
+
+    def __repr__(self) -> str:
+        return f"<FavoriteMovie (user_id='{self.user_id}', movie_id='{self.movie_id}')>"

src/main.py

@@ -1,7 +1,7 @@
 from fastapi import FastAPI
 from fastapi_pagination import add_pagination
 
-from routes import accounts_router, payments_router, profiles_router
+from routes import accounts_router, movies_router, payments_router, profiles_router
 from routes.shopping_cart import router as shopping_carts_router
 
 app = FastAPI(
@@ -16,6 +16,9 @@
 app.include_router(
     profiles_router, prefix=f"{api_version_prefix}/profiles", tags=["profiles"]
 )
+app.include_router(
+    movies_router, prefix=f"{api_version_prefix}/cinema", tags=["cinema"]
+)
 app.include_router(
     payments_router, prefix=f"{api_version_prefix}/payments", tags=["payments"]
 )

src/routes/__init__.py

@@ -1,3 +1,4 @@
 from routes.accounts import router as accounts_router
 from routes.payments import router as payments_router
 from routes.profiles import router as profiles_router
+from routes.movies import router as movies_router