95 lines
2.6 KiB
Python
95 lines
2.6 KiB
Python
from typing import List, Optional, Dict, Any
|
|
import uuid
|
|
from sqlalchemy.orm import Session
|
|
from models.movie import Movie
|
|
from schemas.movie import MovieCreate, MovieSchema
|
|
|
|
def get_all_movies(db: Session) -> List[MovieSchema]:
|
|
"""
|
|
Retrieves all movies from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[MovieSchema]: A list of all movie objects.
|
|
"""
|
|
movies = db.query(Movie).all()
|
|
return [MovieSchema.from_orm(movie) for movie in movies]
|
|
|
|
def create_movie(db: Session, movie_data: MovieCreate) -> MovieSchema:
|
|
"""
|
|
Creates a new movie in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
movie_data (MovieCreate): The data for the movie to create.
|
|
|
|
Returns:
|
|
MovieSchema: The newly created movie object.
|
|
"""
|
|
db_movie = Movie(**movie_data.dict())
|
|
db.add(db_movie)
|
|
db.commit()
|
|
db.refresh(db_movie)
|
|
return MovieSchema.from_orm(db_movie)
|
|
|
|
def get_movie_by_id(db: Session, movie_id: uuid.UUID) -> Optional[MovieSchema]:
|
|
"""
|
|
Retrieves a single movie by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
movie_id (UUID): The ID of the movie to retrieve.
|
|
|
|
Returns:
|
|
Optional[MovieSchema]: The movie object if found, otherwise None.
|
|
"""
|
|
db_movie = db.query(Movie).filter(Movie.id == movie_id).first()
|
|
if db_movie:
|
|
return MovieSchema.from_orm(db_movie)
|
|
return None
|
|
|
|
# In-memory store for movies (if no database model provided)
|
|
_movies_store: List[Dict[str, Any]] = []
|
|
|
|
def create_movie_in_memory(movie_data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Adds a movie to the in-memory list.
|
|
|
|
Args:
|
|
movie_data (Dict[str, Any]): Data for the new movie.
|
|
|
|
Returns:
|
|
Dict[str, Any]: The created movie data, with an added ID.
|
|
"""
|
|
if not movie_data.get("title"):
|
|
raise ValueError("Movie title is required")
|
|
new_movie = movie_data.copy()
|
|
new_movie["id"] = str(uuid.uuid4())
|
|
_movies_store.append(new_movie)
|
|
return new_movie
|
|
|
|
def get_all_movies_in_memory() -> List[Dict[str, Any]]:
|
|
"""
|
|
Retrieves all movies from the in-memory list.
|
|
|
|
Returns:
|
|
List[Dict[str, Any]]: A list of all movies.
|
|
"""
|
|
return _movies_store
|
|
|
|
def get_movie_by_id_in_memory(movie_id: str) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Finds a movie by ID in the in-memory list.
|
|
|
|
Args:
|
|
movie_id (str): The ID of the movie to find.
|
|
|
|
Returns:
|
|
Optional[Dict[str, Any]]: The movie if found, otherwise None.
|
|
"""
|
|
for movie in _movies_store:
|
|
if movie.get("id") == movie_id:
|
|
return movie
|
|
return None |