@leroysheep / MovieWebApp / commits / 6f5b5d1a89

Tested basic tables and configured getter and setter functions

Lee Roy Stevenson committed May 7, 2025 at 05:27 UTC 6f5b5d1a898e4bd6a0c2bc266da64cddc3e2e2d1
6 files changed +560
.idea/MovieWebApp.iml
+3
@@ -7,4 +7,7 @@
7 <orderEntry type="jdk" jdkName="Python 3.12 (MovieWebApp)" jdkType="Python SDK" />
8 <orderEntry type="sourceFolder" forTests="false" />
9 </component>
10 + <component name="PackageRequirementsSettings">
11 + <option name="removeUnused" value="true" />
12 + </component>
13 </module>
\ No newline at end of file
data_manager_interface.py new
+20
@@ -0,0 +1,20 @@
1 +from abc import ABC, abstractmethod
2 +
3 +class DataManagerInterface(ABC):
4 +
5 + @abstractmethod
6 + def users(self):
7 + pass
8 +
9 + @abstractmethod
10 + def users(self, user):
11 + pass
12 +
13 + @abstractmethod
14 + def get_user_movies(self, user_id):
15 + pass
16 +
17 + @abstractmethod
18 + def set_user_movies(self, user_id, movie):
19 + pass
20 +
data_models.py new
+62
@@ -0,0 +1,62 @@
1 +from email.policy import default
2 +
3 +from flask_sqlalchemy.session import Session
4 +from sqlalchemy import create_engine, Column, Integer, String, Float, ForeignKey, Table
5 +from sqlalchemy.orm import relationship, sessionmaker, declarative_base
6 +
7 +# Define the database connection string. Use an in-memory database for testing.
8 +TEST_DB_URL = "sqlite:///:memory:" # This could also be in a config.py
9 +
10 +engine = create_engine(TEST_DB_URL)
11 +
12 +Session = sessionmaker(bind=engine)
13 +session = Session()
14 +
15 +
16 +# Base for declarative models
17 +Base = declarative_base()
18 +
19 +# Define the association table
20 +class UserMovie(Base):
21 + __tablename__ = 'user_movies'
22 + id = Column(Integer, primary_key=True)
23 + user_id = Column('user_id', Integer, ForeignKey('users.id'))
24 + movie_id = Column('movie_id', Integer, ForeignKey('movies.id'))
25 + rating = Column(Float)
26 + user_rating = Column('user_rating', Float, default=0.0 )# Add the rating column here
27 +
28 + def __repr__(self):
29 + return f"<UserMovie(user_id={self.user_id}, movie_id={self.movie_id}, rating={self.rating})>"
30 +
31 +# Define the User model
32 +class User(Base):
33 + __tablename__ = 'users'
34 + id = Column(Integer, primary_key=True)
35 + name = Column(String)
36 + movies = relationship("Movie", secondary="user_movies",
37 + back_populates="users")
38 +
39 + def __repr__(self):
40 + return f"<User(name='{self.name}', id={self.id if self.id else 'None'})>"
41 +
42 +
43 +# Define the Movie model
44 +class Movie(Base):
45 + __tablename__ = 'movies'
46 + id = Column(Integer, primary_key=True)
47 + name = Column(String)
48 + director = Column(String)
49 + year = Column(Integer)
50 + poster = Column(String)
51 + rating = Column(Float)
52 + users = relationship("User", secondary="user_movies",
53 + back_populates="movies")
54 +
55 + def __repr__(self):
56 + return f"<Movie(name='{self.name}', id={self.id if self.id else 'None'})>"
57 +
58 +
59 +if __name__ == "__main__":
60 + # This block is for creating the tables if you want to do it directly.
61 + Base.metadata.create_all(engine)
62 + print("Tables created!")
\ No newline at end of file
database_test.py new
+234
@@ -0,0 +1,234 @@
1 +import pytest
2 +from sqlalchemy.orm import sessionmaker
3 +
4 +from sqlite_data_manager import SQliteDataManager
5 +from data_models import Base, User, Movie
6 +
7 +TEST_DB_URL = "sqlite:///:memory:"
8 +
9 +# Fixture for creating a new data manager for each test
10 +@pytest.fixture(scope="function")
11 +def data_manager():
12 + """
13 + Fixture to create an SQliteDataManager instance for each test. Uses an in-memory database.
14 + """
15 + return SQliteDataManager(TEST_DB_URL)
16 +
17 +
18 +def test_user_model(data_manager: SQliteDataManager):
19 + """Test creating a User object."""
20 + user = User(name="Test User")
21 + with data_manager.SessionFactory() as session:
22 + session.add(user)
23 + session.commit()
24 + retrieved_user = session.query(User).filter_by(name="Test User").first()
25 + assert retrieved_user == user
26 +
27 +
28 +def test_movie_model(data_manager: SQliteDataManager):
29 + """Test creating a Movie object."""
30 + movie = Movie(name="Test Movie", director="Test Director", year=2024, poster="test.jpg", rating=7.5)
31 + with data_manager.SessionFactory() as session:
32 + session.add(movie)
33 + session.commit()
34 + retrieved_movie = session.query(Movie).filter_by(name="Test Movie").first()
35 + assert retrieved_movie == movie
36 +
37 +
38 +def test_user_movie_model(data_manager: SQliteDataManager):
39 + """Test the association between User and Movie."""
40 + user = User(name="User 1")
41 + movie = Movie(name="Movie 1", director="Director A", year=2020, poster="m1.jpg", rating=8.0)
42 + with data_manager.SessionFactory() as session:
43 + session.add_all([user, movie])
44 + session.commit()
45 + user.movies.append(movie)
46 + session.commit()
47 + assert movie in user.movies
48 + assert user in movie.users
49 +
50 +
51 +def test_get_users(data_manager: SQliteDataManager):
52 + """Test getting users from the database using the getter."""
53 + with data_manager.SessionFactory() as session:
54 + # Initially, there should be no users
55 + users_from_getter = data_manager.users # This might return objects from a different session
56 + assert len(users_from_getter) == 0
57 +
58 + # Add some users directly to the database within the current session
59 + user_1 = User(name="User 1")
60 + user_2 = User(name="User 2")
61 + session.add_all([user_1, user_2])
62 + session.commit()
63 +
64 + # Now, retrieve the users using a query within the *same* session
65 + users_from_query = session.query(User).all()
66 + assert len(users_from_query) == 2
67 +
68 + # Check attributes of the queried users
69 + retrieved_user_names = [user.name for user in users_from_query]
70 + assert "User 1" in retrieved_user_names
71 + assert "User 2" in retrieved_user_names
72 +
73 + retrieved_user_ids = {user.id for user in users_from_query if user.id is not None}
74 + assert user_1.id in retrieved_user_ids
75 + assert user_2.id in retrieved_user_ids
76 +
77 +
78 +def test_set_user(data_manager: SQliteDataManager):
79 + """Test setting (adding) a user to the database."""
80 + user = User(name="New User")
81 + with data_manager.SessionFactory() as session:
82 + session.add(user)
83 + session.commit()
84 + retrieved_user = session.query(User).filter_by(name="New User").first()
85 + assert retrieved_user.name == "New User"
86 +
87 +
88 +
89 +def test_get_user_movies(data_manager: SQliteDataManager):
90 + """Test getting movies for a specific user."""
91 + user = User(name="Test User")
92 + movie1 = Movie(name="Movie 1", director="Director A", year=2020, poster="m1.jpg", rating=8.0)
93 + movie2 = Movie(name="Movie 2", director="Director B", year=2022, poster="m2.jpg", rating=7.0)
94 + with data_manager.SessionFactory() as session:
95 + session.add_all([user, movie1, movie2])
96 + session.commit()
97 + data_manager.set_user_movies(user.id, movie1.id, 9.0)
98 + data_manager.set_user_movies(user.id, movie2.id, 8.5)
99 +
100 + session.commit()
101 +
102 + #user_movies = data_manager.get_user_movies(user.id)
103 +
104 + user_movies = data_manager.get_user_movies(user.id) #Adapt this
105 + assert len(user_movies) == 2
106 + assert user_movies[0]['name'] == 'Movie 1'
107 + assert user_movies[0]['rating'] == 9.0
108 + assert user_movies[1]['name'] == 'Movie 2'
109 + assert user_movies[1]['rating'] == 8.5
110 +
111 +
112 +
113 +def test_set_user_movies(data_manager: SQliteDataManager):
114 + """Test setting (adding) a movie to a user's list."""
115 + user = User(name="Test User")
116 + movie = Movie(name="Test Movie", director="Test Director", year=2024, poster="test.jpg", rating=7.5)
117 + with data_manager.SessionFactory() as session:
118 + session.add_all([user, movie])
119 + session.commit()
120 + data_manager.set_user_movies(user.id, movie.id, 5.0) # Adapt this
121 + user_movies = data_manager.get_user_movies(user.id) # Adapt this
122 + assert len(user_movies) == 1
123 + assert user_movies[0]['name'] == 'Test Movie'
124 + assert user_movies[0]['rating'] == 5.0
125 +
126 +
127 +
128 +def test_get_movies(data_manager: SQliteDataManager):
129 + """Test getting all movies."""
130 + movie1 = Movie(name="Movie A", director="Director X", year=2019, poster="a.jpg", rating=6.5)
131 + movie2 = Movie(name="Movie B", director="Director Y", year=2021, poster="b.jpg", rating=8.5)
132 + with data_manager.SessionFactory() as session:
133 + session.add_all([movie1, movie2])
134 + session.commit()
135 + movies = session.query(Movie).all()
136 + assert len(movies) == 2
137 + assert movie1 in movies
138 + assert movie2 in movies
139 +
140 +
141 +
142 +def test_set_movie(data_manager: SQliteDataManager):
143 + """Test setting (adding) a movie."""
144 + movie = Movie(name="New Movie", director="New Director", year=2023, poster="new.jpg", rating=9.0)
145 + with data_manager.SessionFactory() as session:
146 + session.add(movie)
147 + session.commit()
148 + retrieved_movies = session.query(Movie).filter_by(name="New Movie").all()
149 + assert len(retrieved_movies) == 1
150 + assert retrieved_movies[0].name == "New Movie"
151 +
152 +
153 +
154 +def test_update_movie(data_manager: SQliteDataManager):
155 + """Test updating an existing movie."""
156 + movie = Movie(name="Old Movie", director="Old Director", year=2000, poster="old.jpg", rating=5.0)
157 + with data_manager.SessionFactory() as session:
158 + session.add(movie)
159 + session.commit()
160 + movie_to_update = session.query(Movie).get(movie.id)
161 +
162 + updated_movie_data = {
163 + "name": "Updated Movie",
164 + "director": "Updated Director",
165 + "year": 2022,
166 + "poster": "updated.jpg",
167 + "rating": 8.0,
168 + }
169 + for key, value in updated_movie_data.items():
170 + setattr(movie_to_update, key, value)
171 + session.commit()
172 +
173 + updated_movie = session.query(Movie).get(movie.id)
174 + assert updated_movie is not None
175 + assert updated_movie.name == "Updated Movie"
176 + assert updated_movie.director == "Updated Director"
177 + assert updated_movie.year == 2022
178 + assert updated_movie.poster == "updated.jpg"
179 + assert updated_movie.rating == 8.0
180 +
181 +
182 +
183 +def test_update_movie_partial(data_manager: SQliteDataManager):
184 + """Test updating a movie with partial data."""
185 + movie = Movie(name="Original Movie", director="Original Director", year=2000, poster="original.jpg", rating=5.0)
186 + with data_manager.SessionFactory() as session:
187 + session.add(movie)
188 + session.commit()
189 + movie_to_update = session.query(Movie).get(movie.id)
190 +
191 + updated_data = {"rating": 9.5, "name": "Partially Updated"}
192 + for key, value in updated_data.items():
193 + setattr(movie_to_update, key, value)
194 + session.commit()
195 +
196 + updated_movie = session.query(Movie).get(movie.id)
197 + assert updated_movie is not None
198 + assert updated_movie.name == "Partially Updated"
199 + assert updated_movie.rating == 9.5
200 + assert updated_movie.director == "Original Director" # Unchanged
201 + assert updated_movie.year == 2000 # Unchanged
202 + assert updated_movie.poster == "original.jpg" #Unchanged
203 +
204 +
205 +
206 +def test_update_movie_not_found(data_manager: SQliteDataManager):
207 + """Test updating a non-existent movie."""
208 + updated_data = {"name": "Nonexistent Movie"}
209 + with data_manager.SessionFactory() as session:
210 + result = session.query(Movie).get(999)
211 + assert result is None
212 +
213 +
214 +
215 +def test_delete_movie(data_manager: SQliteDataManager):
216 + """Test deleting a movie."""
217 + movie = Movie(name="Movie to Delete", director="Delete Director", year=2023, poster="delete.jpg", rating=9.0)
218 + with data_manager.SessionFactory() as session:
219 + session.add(movie)
220 + session.commit()
221 + movie_to_delete = session.query(Movie).get(movie.id)
222 + session.delete(movie_to_delete)
223 + session.commit()
224 + retrieved_movie = session.query(Movie).get(movie.id)
225 + assert retrieved_movie is None
226 +
227 +
228 +
229 +def test_delete_movie_not_found(data_manager: SQliteDataManager):
230 + """Test deleting a non-existent movie."""
231 + with data_manager.SessionFactory() as session:
232 + deleted = data_manager.delete_movie(999)
233 + assert deleted is False
234 +
requirements.txt new
+2
@@ -0,0 +1,2 @@
1 +flask
2 +flask_sqlalchemy
\ No newline at end of file
sqlite_data_manager.py new
+239
@@ -0,0 +1,239 @@
1 +
2 +from sqlalchemy import create_engine
3 +from sqlalchemy.orm import sessionmaker, joinedload, Session
4 +from contextlib import contextmanager
5 +from typing import List, Dict, Any
6 +from data_models import Base, User, Movie, UserMovie
7 +from data_manager_interface import DataManagerInterface
8 +
9 +
10 +# Define the database connection string.
11 +TEST_DB_URL = "sqlite:///:memory:"
12 +
13 +
14 +
15 +# Data manager class to handle database operations
16 +class SQliteDataManager(DataManagerInterface):
17 + def __init__(self, db_url: str):
18 + """
19 + Initialize the data manager with a database URL.
20 + """
21 + self.engine = create_engine(db_url)
22 + self.SessionFactory = sessionmaker(bind=self.engine)
23 + Base.metadata.create_all(self.engine)
24 +
25 + @contextmanager
26 + def get_db(self):
27 + """
28 + Provide a database session as a context manager.
29 + """
30 + session = self.SessionFactory()
31 + try:
32 + yield session
33 + session.commit()
34 + except Exception:
35 + session.rollback()
36 + raise
37 + finally:
38 + session.close()
39 +
40 + @property
41 + def users(self) -> List[User]:
42 + """
43 + Getter for users.
44 + Returns: a list of User objects.
45 + """
46 + with self.get_db() as db:
47 + return db.query(User).all()
48 +
49 + def add_user(self, user: User) -> None:
50 + """
51 + Add a user to the database.
52 + """
53 + with self.get_db() as db:
54 + db.add(user)
55 + db.commit()
56 +
57 + @property
58 + def movies(self) -> List[Movie]:
59 + """
60 + Getter for movies.
61 + Returns: a list of Movie objects
62 + """
63 + with self.get_db() as db:
64 + return db.query(Movie).options(joinedload(Movie.users)).all()
65 +
66 + def set_user_movies(self, user_id: int, movie_id: int, rating: float, user_rating: float = 0.0)\
67 + -> None:
68 + """
69 + Set (add) a movie to a user's list with a rating, or update the rating if it exists (self-contained session).
70 + """
71 + engine = self.engine
72 + with Session(engine) as db:
73 + user = db.query(User).filter_by(id=user_id).first()
74 + movie = db.query(Movie).filter_by(id=movie_id).first()
75 +
76 + if user and movie:
77 + existing_associations = db.query(UserMovie).filter_by(
78 + user_id=user_id, movie_id=movie_id
79 + ).all() #check if the association already exists
80 +
81 + if existing_associations:
82 + # existing association update
83 + db.execute(
84 + UserMovie.update().
85 + where(UserMovie.user_id == user_id).
86 + where(UserMovie.movie_id == movie_id).
87 + values(rating=rating, user_rating=user_rating)
88 + )
89 + if user_rating:
90 + print(f"Updated rating for user {user_id} and movie {movie_id} to "
91 + f"{rating} and also updated user rating to {user_rating}")
92 + else:
93 + print(f"Updated rating for user {user_id} and movie {movie_id} to {rating}")
94 + else:
95 + # create new association
96 + association = UserMovie(user_id=user_id, movie_id=movie_id, rating=rating,
97 + user_rating=user_rating)
98 + db.add(association)
99 + print(f"Added movie {movie_id} to user {user_id} with rating {rating}")
100 +
101 + db.commit() # Commit within the function
102 + else:
103 + print("User or movie not found.")
104 +
105 + def get_user_movies(self, user_id: int) -> List[Dict[str, Any]]:
106 + """
107 + Get movies for a specific user with their ratings.
108 + Returns: A list of dictionaries, where each dictionary contains movie details and the user's rating.
109 + """
110 + with self.get_db() as db:
111 + user = db.query(User).options(joinedload(User.movies)).filter_by(id=user_id).first()
112 + if user:
113 + movies_with_ratings = []
114 + for movie in user.movies:
115 + association = db.query(UserMovie).filter_by(user_id=user.id,
116 + movie_id=movie.id).first()
117 + if association:
118 + movies_with_ratings.append({
119 + "id": movie.id,
120 + "name": movie.name,
121 + "director": movie.director,
122 + "year": movie.year,
123 + "poster": movie.poster,
124 + "rating": association.rating,
125 + })
126 + return movies_with_ratings
127 + return []
128 +
129 +
130 + def set_movie(self, movie: Movie) -> None:
131 + """
132 + Add a new movie to the database.
133 + """
134 + with self.get_db() as db:
135 + db.add(movie)
136 + db.commit()
137 +
138 + def update_movie(self, movie_id: int, update_data: dict) -> Movie | None:
139 + session = self.SessionFactory()
140 + try:
141 + movie = session.query(Movie).filter_by(id=movie_id).first()
142 + if movie:
143 + for key, value in update_data.items():
144 + setattr(movie, key, value)
145 + session.commit()
146 + return movie
147 + return None
148 + finally:
149 + session.close()
150 +
151 + def delete_movie(self, movie_id: int) -> bool:
152 + """
153 + Delete a movie and its associations.
154 +
155 + Args:
156 + movie_id: The ID of the movie to delete.
157 +
158 + Returns:
159 + True if the movie was successfully deleted, False otherwise.
160 + """
161 + engine = self.engine
162 + with Session(engine) as db:
163 + movie = db.query(Movie).options(joinedload(Movie.users)).filter_by(id=movie_id).first()
164 + if movie:
165 + # Remove the movie from all associated users' movie lists
166 + for user in movie.users:
167 + user.movies.remove(movie)
168 + db.delete(movie)
169 + db.commit()
170 + return True
171 + return False
172 +
173 +def main():
174 + data_manager = SQliteDataManager(TEST_DB_URL)
175 +
176 + with data_manager.SessionFactory() as session:
177 + # Create some users
178 + user1 = User(name="Alice")
179 + user2 = User(name="Bob")
180 + data_manager.add_user(user1)
181 + data_manager.add_user(user2)
182 + session.add_all([user1, user2])
183 + session.commit()
184 +
185 + # Create some movies
186 + movie1 = Movie(name="The Matrix", director="Wachowskis", year=1999, poster="matrix.jpg",
187 + rating=8.7)
188 + movie2 = Movie(name="Inception", director="Nolan", year=2010, poster="inception.jpg",
189 + rating=8.8)
190 + data_manager.set_movie(movie1)
191 + data_manager.set_movie(movie2)
192 + session.add_all([movie1, movie2])
193 + session.commit()
194 +
195 + # Associate movies with users and assign ratings
196 + data_manager.set_user_movies(user1.id, movie1.id, 9.0,8.9)
197 + print(data_manager.get_user_movies(user1.id))
198 + data_manager.set_user_movies(user1.id, movie2.id, 8.5)
199 + data_manager.set_user_movies(user2.id, movie2.id, 9.2)
200 +
201 + # Get and print user movies (use a new session for querying)
202 +
203 + print("Alice's movies:", data_manager.get_user_movies(user1.id))
204 + print("Bob's movies:", data_manager.get_user_movies(user2.id))
205 +
206 + # Get and print all movies
207 + print("All movies:", session.query(Movie).all())
208 +
209 + # Example of updating a movie (use a new session)
210 + with data_manager.SessionFactory() as session_helper:
211 + updated_movie = data_manager.update_movie(movie1.id,
212 + {"name": "The Matrix Reloaded", "rating": 7.2})
213 + session_helper.add(updated_movie)
214 + session_helper.commit()
215 + print("Updated movie:", updated_movie)
216 +
217 + # Example of deleting a movie (use a new session)
218 + with data_manager.SessionFactory() as session_helper:
219 + deleted = data_manager.delete_movie(movie1.id)
220 + session_helper.commit()
221 + print("Deleted movie1:", deleted)
222 +
223 + print("All movies after deletion:", session.query(Movie).all())
224 +
225 +
226 +
227 +
228 +
229 +
230 +
231 +
232 +
233 +
234 +
235 +
236 +
237 +if __name__ == "__main__":
238 + main()
239 +