| 1 | |
| 2 | from sqlalchemy import create_engine |
| 3 | from sqlalchemy.orm import sessionmaker |
| 4 | from contextlib import contextmanager |
| 5 | from typing import List, Dict, Any, Type |
| 6 | |
| 7 | from data_models import Base, User, Movie, UserMovie |
| 8 | from movie_api import OMDBClient |
| 9 | from datamanager.data_manager_interface import DataManagerInterface |
| 10 | |
| 11 | |
| 12 | # Define the database connection string. |
| 13 | TEST_DB_URL = "sqlite:///:memory:" |
| 14 | |
| 15 | |
| 16 | |
| 17 | # Data manager class to handle database operations |
| 18 | class SQliteDataManager(DataManagerInterface): |
| 19 | def __init__(self, db_url: str): |
| 20 | """ |
| 21 | Initialize the data manager with a database URL. |
| 22 | """ |
| 23 | self.engine = create_engine(db_url) |
| 24 | self.SessionFactory = sessionmaker(bind=self.engine, expire_on_commit=False) |
| 25 | Base.metadata.create_all(self.engine) |
| 26 | |
| 27 | @contextmanager |
| 28 | def get_db(self): |
| 29 | """ |
| 30 | Provide a database session as a context manager. |
| 31 | """ |
| 32 | session = self.SessionFactory() |
| 33 | try: |
| 34 | yield session |
| 35 | session.commit() |
| 36 | except Exception: |
| 37 | session.rollback() |
| 38 | raise ConnectionError("Database connection error!") |
| 39 | finally: |
| 40 | session.close() |
| 41 | |
| 42 | |
| 43 | @property |
| 44 | def users(self) -> list[Type[User]]: |
| 45 | """ |
| 46 | Getter for users. |
| 47 | Returns: a list of User objects. |
| 48 | """ |
| 49 | with self.SessionFactory() as session: |
| 50 | users = session.query(User).all() |
| 51 | return users |
| 52 | |
| 53 | |
| 54 | def get_user(self, user_id: int) -> List[User] | None: |
| 55 | """ |
| 56 | Get a user by ID. |
| 57 | :param user_id: |
| 58 | :return: User |
| 59 | """ |
| 60 | with self.SessionFactory() as session: |
| 61 | user = session.query(User).filter_by(id=user_id).first() |
| 62 | return user |
| 63 | |
| 64 | def add_user(self, user: User) -> None: |
| 65 | """ |
| 66 | Add a user to the database. |
| 67 | """ |
| 68 | with self.SessionFactory() as session: |
| 69 | session.add(user) |
| 70 | session.commit() |
| 71 | |
| 72 | |
| 73 | @property |
| 74 | def movies(self) -> list[Type[Movie]]: |
| 75 | """ |
| 76 | Getter for movies. |
| 77 | Returns: a list of Movie objects |
| 78 | """ |
| 79 | with self.SessionFactory() as session: |
| 80 | return session.query(Movie).all() |
| 81 | |
| 82 | def set_user_movies(self, user_id: int, movie_id: int, user_rating: float = 0.0)\ |
| 83 | -> str | None: |
| 84 | """ |
| 85 | Set (add) a movie to a user's list with a rating, |
| 86 | either create a new association or update the rating if it exists (self-contained session). |
| 87 | """ |
| 88 | |
| 89 | with self.SessionFactory() as session: |
| 90 | user = session.query(User).filter_by(id=user_id).first() |
| 91 | movie = session.query(Movie).filter_by(id=movie_id).first() |
| 92 | |
| 93 | if user and movie: |
| 94 | existing_associations = session.query(UserMovie).filter_by( |
| 95 | user_id=user_id, movie_id=movie_id |
| 96 | ).first() #check if the association already exists |
| 97 | if existing_associations: |
| 98 | # existing association update |
| 99 | session.query(UserMovie).filter_by(user_id=user_id, movie_id=movie_id).update( |
| 100 | {"user_rating": user_rating} |
| 101 | ) |
| 102 | else: |
| 103 | # create new association |
| 104 | association = UserMovie(user_id=user_id, movie_id=movie_id, |
| 105 | user_rating=user_rating) |
| 106 | session.add(association) |
| 107 | |
| 108 | session.commit() # Commit within the function |
| 109 | |
| 110 | |
| 111 | elif movie is None: |
| 112 | return "Failed to add movie, check ID and try again!" |
| 113 | |
| 114 | else: |
| 115 | return "User not found." |
| 116 | |
| 117 | def get_user_movies(self, user_id: int) -> List[Dict[str, Any]]: |
| 118 | """ |
| 119 | Get movies for a specific user with their ratings. |
| 120 | Returns: A list of dictionaries, where each dictionary contains movie details |
| 121 | (name, director, year, poster) and the user's rating. |
| 122 | """ |
| 123 | with self.SessionFactory() as session: |
| 124 | user_movies = session.query(UserMovie).filter_by(user_id=user_id).all() |
| 125 | if user_movies: |
| 126 | movies_with_ratings = [] |
| 127 | for association in user_movies: |
| 128 | movie = session.query(Movie).filter_by(id=association.movie_id).first() |
| 129 | if movie: |
| 130 | movies_with_ratings.append({ |
| 131 | "id": movie.id, |
| 132 | "name": movie.name, |
| 133 | "director": movie.director, |
| 134 | "year": movie.year, |
| 135 | "poster": movie.poster, |
| 136 | "genre": movie.genre, |
| 137 | "country": movie.country, |
| 138 | "plot": movie.plot, |
| 139 | "rating": movie.rating, |
| 140 | "user_rating": association.user_rating, |
| 141 | "user_comment": association.user_comment if association.user_comment |
| 142 | else None |
| 143 | }) |
| 144 | return movies_with_ratings |
| 145 | return [] |
| 146 | |
| 147 | |
| 148 | def get_user_movie(self, user_id: int, movie_id: int) -> Dict[str, Any]: |
| 149 | with self.SessionFactory() as session: |
| 150 | user_movie = session.query(UserMovie).filter_by(user_id=user_id, movie_id=movie_id).first() |
| 151 | if user_movie: |
| 152 | movie = session.query(Movie).filter_by(id=movie_id).first() |
| 153 | return { |
| 154 | "id": movie.id, |
| 155 | "name": movie.name, |
| 156 | "director": movie.director, |
| 157 | "year": movie.year, |
| 158 | "poster": movie.poster, |
| 159 | "genre": movie.genre, |
| 160 | "country": movie.country, |
| 161 | "plot": movie.plot, |
| 162 | "rating": movie.rating, |
| 163 | "user_rating": user_movie.user_rating, |
| 164 | "user_comment": user_movie.user_comment if user_movie.user_comment else None |
| 165 | } |
| 166 | return {} |
| 167 | |
| 168 | |
| 169 | def set_movie(self, movie_title: str) -> Type[Movie] | Movie: |
| 170 | """ |
| 171 | Add a new movie to the database. |
| 172 | :param movie_title: The title of the movie to add. |
| 173 | """ |
| 174 | with self.SessionFactory() as session: |
| 175 | movie = session.query(Movie).filter_by(name=movie_title).first() |
| 176 | if movie: |
| 177 | return movie |
| 178 | new_movie = OMDBClient().get_movie(title=movie_title) |
| 179 | if new_movie is None: |
| 180 | raise ValueError("Movie not found") |
| 181 | movie = Movie(**new_movie) |
| 182 | session.add(movie) |
| 183 | session.commit() |
| 184 | return movie |
| 185 | |
| 186 | def update_user_movie(self, user_id: int, movie_id: int, update_data: dict) -> dict | None: |
| 187 | with self.SessionFactory() as session: |
| 188 | movie = session.query(Movie).filter_by(id=movie_id).first() |
| 189 | if movie: |
| 190 | session.query(UserMovie).filter_by(user_id=user_id, |
| 191 | movie_id=movie_id).update(update_data) |
| 192 | session.commit() |
| 193 | return { |
| 194 | "id": movie.id, |
| 195 | "name": movie.name, |
| 196 | "director": movie.director, |
| 197 | "year": movie.year, |
| 198 | "poster": movie.poster, |
| 199 | "genre": movie.genre, |
| 200 | "country": movie.country, |
| 201 | "plot": movie.plot, |
| 202 | "rating": movie.rating, |
| 203 | "user_rating": update_data["user_rating"] if "user_rating" in update_data |
| 204 | else 0.0, |
| 205 | "user_comment": update_data["user_comment"] if "user_comment" in update_data |
| 206 | else None |
| 207 | } |
| 208 | return None |
| 209 | |
| 210 | def delete_movie(self, movie_id: int) -> bool: |
| 211 | """ |
| 212 | Delete a movie from the movies table in the database |
| 213 | :param movie_id: |
| 214 | :return: |
| 215 | """ |
| 216 | with self.SessionFactory() as session: |
| 217 | movie = session.query(Movie).filter_by(id = movie_id).first() |
| 218 | if movie: |
| 219 | session.flush() # Make sure pending updates are flushed |
| 220 | session.query(UserMovie).filter_by(movie_id = movie_id).delete() |
| 221 | session.query(Movie).filter_by(id = movie_id).delete() |
| 222 | session.commit() |
| 223 | return True |
| 224 | return False |
| 225 | |
| 226 | def delete_user(self, user_id: int) -> bool: |
| 227 | """ |
| 228 | Delete a user from the users table in the database |
| 229 | :param user_id: |
| 230 | :return: |
| 231 | """ |
| 232 | with self.SessionFactory() as session: |
| 233 | user = session.query(User).filter_by(id = user_id).first() |
| 234 | if user: |
| 235 | session.flush() # Make sure pending updates are flushed |
| 236 | session.query(UserMovie).filter_by(user_id = user_id).delete() |
| 237 | session.query(User).filter_by(id = user_id).delete() |
| 238 | session.commit() |
| 239 | return True |
| 240 | return False |
| 241 | |
| 242 | |
| 243 | def delete_user_movie(self, user_id: int, movie_id: int) -> bool: |
| 244 | with self.SessionFactory() as session: |
| 245 | association = session.query(UserMovie).filter_by(user_id=user_id, movie_id=movie_id).first() |
| 246 | if association: |
| 247 | session.delete(association) |
| 248 | session.commit() |
| 249 | return True |
| 250 | return False |
| 251 | |
| 252 | def get_movie(self, movie_id: int) -> Movie | None: |
| 253 | """ |
| 254 | Get a movie from the movies table in the database |
| 255 | :param movie_id: |
| 256 | :return: |
| 257 | """ |
| 258 | with self.SessionFactory() as session: |
| 259 | movie = session.query(Movie).filter_by(id = movie_id).first() |
| 260 | return movie |
| 261 | |
| 262 | def main(): |
| 263 | data_manager = SQliteDataManager("sqlite:///movie_app.db") |
| 264 | |
| 265 | with data_manager.SessionFactory() as session: |
| 266 | # Create some users |
| 267 | user1 = User(name="Alice") |
| 268 | user2 = User(name="Bob") |
| 269 | data_manager.add_user(user1) |
| 270 | data_manager.add_user(user2) |
| 271 | session.add_all([user1, user2]) |
| 272 | session.commit() |
| 273 | |
| 274 | # Create some movies |
| 275 | movie1 = data_manager.set_movie("The Matrix") |
| 276 | movie2 = data_manager.set_movie("Interstellar") |
| 277 | |
| 278 | # Associate movies with users and assign ratings |
| 279 | print(f"user1:{user1}, movie1:{movie1}") |
| 280 | data_manager.set_user_movies(user1.id, movie1.id) |
| 281 | print(data_manager.get_user_movies(user1.id)) |
| 282 | data_manager.set_user_movies(user1.id, movie2.id, 8.5) |
| 283 | data_manager.set_user_movies(user2.id, movie2.id, 9.2) |
| 284 | |
| 285 | # Get and print user movies (use a new session for querying) |
| 286 | |
| 287 | print("Alice's movies:", data_manager.get_user_movies(user1.id)) |
| 288 | print("Bob's movies:", data_manager.get_user_movies(user2.id)) |
| 289 | |
| 290 | # Get and print all movies |
| 291 | print("All movies:", session.query(Movie).all()) |
| 292 | |
| 293 | # Example of updating a movie (use a new session) |
| 294 | updated_movie = data_manager.update_user_movie(user1.id, movie1.id, |
| 295 | {"user_rating": |
| 296 | 7.2}) |
| 297 | print("Updated movie:", updated_movie) |
| 298 | |
| 299 | # Example of deleting a movie (use a new session) |
| 300 | deleted = data_manager.delete_movie(movie1.id) |
| 301 | print("Deleted movie1:", deleted) |
| 302 | |
| 303 | print("All movies after deletion:", session.query(Movie).all()) |
| 304 | |
| 305 | |
| 306 | if __name__ == "__main__": |
| 307 | main() |
| 308 |