QMDB – کوئریها ۲ (بهروزرسانی و حذف)
این تمرین در اصل برای ارسال چند فایل (در یک zip) نوشته شده است. در این نسخه همهچیز را در همین یک فایل بنویس: اول مدلها، بعد Managerها. خطهای
from models import …هم همانطور که در صورت سؤال آمده کار میکنند. داور یک پایگاهدادهی SQLite در حافظه میسازد، جدولها را از رویBaseتو درست میکند و متغیرsessionرا در اختیار آزمونها میگذارد.
در این تمرین شما باید توابع مربوط به عملیات پایه و کوئریهای پرکاربرد روی پایگاهداده برای سیستم امتیازدهی فیلم کوئرا را بنویسید. شما باید این توابع را برای تعامل با مدلهایی که در تمرین اول تعریف کردهاید و با استفاده از کلاسهایی که برای مدیریت این مدلها در تمرین قبل نوشتید، پیادهسازی کنید.
🎯 کلاس GenreManager
این کلاس در فایل genre_manager.py تعریف شدهاست.
class GenreManager:
def __init__(self, session: Session):
self.session = session
def update(self, genre_id: int, new_name: str) -> Genre:
pass
def delete(self, genre_id: int) -> bool:
pass
def get_genres_with_most_movies(self) -> list[tuple]:
pass
📄 genre_manager.py
🔹 update(self, genre_id: int, new_name: str) -> Genre
نام یک ژانر را بهروزرسانی میکند. اگر ژانر موجود باشد، مقدار بهروزشده را بازمیگرداند، در غیر این صورت
Noneبرمیگرداند.ورودیها:
genre_id(شناسه ژانر)،new_name(نام جدید)خروجی: شیء
GenreیاNone
🔹 delete(self, genre_id: int) -> bool
ژانری با شناسهی مشخص را از پایگاهداده حذف میکند.
ورودیها:
genre_id(شناسه ژانر)خروجی:
bool(موفقیت یا عدم موفقیت)
🔹 get_genres_with_most_movies(self) -> list[tuple]
ژانرهایی را که بیشترین تعداد فیلمها را دارند، به همراه تعداد فیلمهای هر ژانر بازمیگرداند. دادهها بر اساس تعداد فیلمها بهصورت نزولی مرتب شدهاند.
خروجی: لیستی از تاپلهای
(Genre, movie_count)
🎯 کلاس MovieManager
این کلاس در فایل movie_manager.py تعریف شدهاست.
class MovieManager:
def __init__(self, session: Session) -> None:
self.session = session
def update(self, movie_id: int, update_data: dict) -> Movie:
pass
def delete(self, movie_id: int) -> bool:
pass
def remove_genre(self, movie_id: int, genre: Genre) -> Movie:
pass
def get_top_movies_by_rating(self, limit: int = 10) -> list[tuple]:
pass
def get_movies_by_genre(self, genre_name: str) -> list[Movie]:
pass
def get_top_rated_movies_by_genre(self):
pass
📄 movie_manager.py
🔹 update(self, movie_id: int, update_data: dict) -> Movie
محتوای یک فیلم را بهروزرسانی میکند. اطلاعات جدید از طریق یک دیکشنری شامل فیلدهای فیلم دریافت میشود.
ورودیها:
movie_id،update_data(دیکشنری فیلدهای جدید)خروجی: شیء
MovieیاNone
🔹 delete(self, movie_id: int) -> bool
فیلمی با شناسهی مشخص را از پایگاهداده حذف میکند.
ورودیها:
movie_idخروجی:
bool(موفقیت یا عدم موفقیت)
🔹 remove_genre(self, movie_id: int, genre: Genre) -> Movie
ژانر مورد نظر را از لیست ژانرهای یک فیلم حذف میکند.
ورودیها:
movie_id،genre(شیء ژانر)خروجی: شیء
Movie
🔹 get_top_movies_by_rating(self, limit: int = 10) -> list[tuple]
تعداد مشخصی از فیلمهایی را که بالاترین میانگین امتیاز را دارند، برمیگرداند.
ورودیها:
limit(پیشفرض: ۱۰)خروجی: لیستی از تاپلهای
(Movie, average_rating)
🔹 get_movies_by_genre(self, genre_name: str) -> list[Movie]
لیستی از تمام فیلمهایی که در یک ژانر خاص قرار دارند را برمیگرداند.
ورودیها:
genre_name(نام ژانر)خروجی: لیستی از اشیاء
Movie
🔹 get_top_rated_movies_by_genre(self)
برای هر ژانر، فیلمهایی که بالاترین میانگین امتیاز را دارند، برمیگرداند. دادهها بر اساس نام ژانر و میانگین امتیاز به صورت نزولی مرتب شدهاند.
خروجی: لیستی از تاپلهای
(Genre.name, Movie.title, average_rating)
🎯 کلاس ReviewManager
این کلاس در فایل review_manager.py تعریف شدهاست.
class ReviewManager:
def __init__(self, session: Session):
self.session = session
def update(self, review_id: int, update_data: dict) -> Review:
pass
def delete(self, review_id: int) -> bool:
pass
📄 review_manager.py
🔹 update(self, review_id: int, update_data: dict) -> Review
محتوای یک نقد را بهروزرسانی میکند. اطلاعات جدید شامل
commentوratingاست.ورودیها:
review_id،update_data(دیکشنری فیلدهای جدید)خروجی: شیء
ReviewیاNone
🔹 delete(self, review_id: int) -> bool
نقدی با شناسهی مشخص را از پایگاهداده حذف میکند.
ورودیها:
review_idخروجی:
bool(موفقیت یا عدم موفقیت)
🎯 کلاس UserManager
این کلاس در فایل user_manager.py تعریف شدهاست.
class UserManager:
def __init__(self, session: Session):
self.session = session
def update(self, user_id: int, update_data: dict) -> User:
pass
def delete(self, user_id: int) -> bool:
pass
def verify_user(self, user_id: int) -> User:
pass
📄 user_manager.py
🔹 update(self, user_id: int, update_data: dict) -> User
اطلاعات یک کاربر را بهروزرسانی میکند. فیلدهای قابل بهروزرسانی:
name،is_verifiedورودیها:
user_id،update_data(دیکشنری فیلدهای جدید)خروجی: شیء
UserیاNone
🔹 delete(self, user_id: int) -> bool
کاربری با شناسهی مشخص را از پایگاهداده حذف میکند.
ورودیها:
user_idخروجی:
bool(موفقیت یا عدم موفقیت)
🔹 verify_user(self, user_id: int) -> User
ایمیل یک کاربر را با تنظیم
is_verifiedبهTrueتأیید میکند.ورودیها:
user_idخروجی: شیء
UserیاNone
✅ پاسخ نهایی: پیادهسازی کامل Managerها
📄 genre_manager.py
from sqlalchemy import select, func
from sqlalchemy.orm import Session
from models import Genre, MovieGenre
class GenreManager:
def __init__(self, session: Session):
self.session = session
def update(self, genre_id: int, new_name: str) -> Genre:
genre = self.session.get(Genre, genre_id)
if genre is None:
return None
genre.name = new_name
self.session.commit()
return genre
def delete(self, genre_id: int) -> bool:
genre = self.session.get(Genre, genre_id)
if genre is None:
return False
self.session.delete(genre)
self.session.commit()
return True
def get_genres_with_most_movies(self) -> list[tuple]:
stmt = select(
Genre,
func.count(MovieGenre.movie_id).label('movie_count')
).join(MovieGenre, Genre.id == MovieGenre.genre_id
).group_by(Genre.id
).order_by(func.count(MovieGenre.movie_id).desc())
return self.session.execute(stmt).all()
📄 movie_manager.py
from sqlalchemy import select, func, and_
from sqlalchemy.orm import Session
from models import Movie, Genre, Review, MovieGenre
class MovieManager:
def __init__(self, session: Session) -> None:
self.session = session
def update(self, movie_id: int, update_data: dict) -> Movie:
movie = self.session.get(Movie, movie_id)
if movie is None:
return None
for key, value in update_data.items():
if hasattr(movie, key):
setattr(movie, key, value)
self.session.commit()
return movie
def delete(self, movie_id: int) -> bool:
movie = self.session.get(Movie, movie_id)
if movie is None:
return False
self.session.delete(movie)
self.session.commit()
return True
def remove_genre(self, movie_id: int, genre: Genre) -> Movie:
movie = self.session.get(Movie, movie_id)
if movie is None:
raise ValueError(f"Movie with id {movie_id} not found")
if genre in movie.genres:
movie.genres.remove(genre)
self.session.commit()
return movie
def get_top_movies_by_rating(self, limit: int = 10) -> list[tuple]:
stmt = select(
Movie,
func.avg(Review.rating).label('average_rating')
).join(Review, Movie.id == Review.movie_id
).group_by(Movie.id
).order_by(func.avg(Review.rating).desc()
).limit(limit)
return self.session.execute(stmt).all()
def get_movies_by_genre(self, genre_name: str) -> list[Movie]:
stmt = select(Movie).join(MovieGenre).join(Genre).where(Genre.name == genre_name)
return self.session.execute(stmt).scalars().all()
def get_top_rated_movies_by_genre(self):
# زیرکوئری برای میانگین امتیاز هر فیلم
subquery = select(
Review.movie_id,
func.avg(Review.rating).label('avg_rating')
).group_by(Review.movie_id).subquery()
# کوئری اصلی با JOIN و window function
stmt = select(
Genre.name.label('genre_name'),
Movie.title.label('movie_title'),
subquery.c.avg_rating
).select_from(Genre
).join(MovieGenre, Genre.id == MovieGenre.genre_id
).join(Movie, Movie.id == MovieGenre.movie_id
).join(subquery, Movie.id == subquery.c.movie_id
).where(Movie.id == subquery.c.movie_id
).order_by(Genre.name, subquery.c.avg_rating.desc())
return self.session.execute(stmt).all()
📄 review_manager.py
from sqlalchemy.orm import Session
from models import Review
class ReviewManager:
def __init__(self, session: Session):
self.session = session
def update(self, review_id: int, update_data: dict) -> Review:
review = self.session.get(Review, review_id)
if review is None:
return None
for key, value in update_data.items():
if hasattr(review, key):
setattr(review, key, value)
self.session.commit()
return review
def delete(self, review_id: int) -> bool:
review = self.session.get(Review, review_id)
if review is None:
return False
self.session.delete(review)
self.session.commit()
return True
📄 user_manager.py
from sqlalchemy.orm import Session
from models import User
class UserManager:
def __init__(self, session: Session):
self.session = session
def update(self, user_id: int, update_data: dict) -> User:
user = self.session.get(User, user_id)
if user is None:
return None
for key, value in update_data.items():
if hasattr(user, key):
setattr(user, key, value)
self.session.commit()
return user
def delete(self, user_id: int) -> bool:
user = self.session.get(User, user_id)
if user is None:
return False
self.session.delete(user)
self.session.commit()
return True
def verify_user(self, user_id: int) -> User:
user = self.session.get(User, user_id)
if user is None:
return None
user.is_verified = True
self.session.commit()
return user
- برای بهروزرسانی با
update: دریافت شیء → تغییر ویژگیها →commit- برای حذف با
delete: دریافت شیء →session.delete()→commit- برای حذف از رابطه: از
list.remove()روی رابطه استفاده کنید- برای کوئریهای گروهبندی: از
group_by()و توابع تجمیعی استفاده کنید- برای JOIN: از
join()وouterjoin()استفاده کنید- برای زیرکوئریها: از
subquery()استفاده کنید- همیشه
is_verifiedرا برای تأیید کاربر تنظیم کنید
🚀 پس از پیادهسازی کامل Managerها، میتوانید از آنها برای مدیریت کامل پایگاهداده سیستم QMDB استفاده کنید.
نمونهها
data = seed() genres = GenreManager(session) genres.update(data["drama"].id, "Drama & Life").name genres.update(999, "x") is None genres.delete(data["drama"].id) genres.delete(data["drama"].id) sorted(genre.name for genre in session.query(Genre).all()) [genre.name for genre in session.get(Movie, data["inception"].id).genres]
'Drama & Life' True True False ['Sci-Fi'] ['Sci-Fi']
data = seed() [(genre.name, count) for genre, count in GenreManager(session).get_genres_with_most_movies()]
[('Sci-Fi', 3), ('Drama', 1)]data = seed()
movies = MovieManager(session)
changed = movies.update(data["up"].id, {"title": "Up!", "release_year": 2010})
[changed.title, changed.release_year]
movies.update(999, {"title": "x"}) is None
movies.remove_genre(data["inception"].id, data["drama"]).title
[genre.name for genre in session.get(Movie, data["inception"].id).genres]
movies.delete(data["solaris"].id)
movies.delete(data["solaris"].id)
sorted(movie.title for movie in session.query(Movie).all())['Up!', 2010] True 'Inception' ['Sci-Fi'] True False ['Inception', 'Tenet', 'Up!']
data = seed()
movies = MovieManager(session)
[(movie.title, float(average)) for movie, average in movies.get_top_movies_by_rating(2)]
sorted(movie.title for movie in movies.get_movies_by_genre("Sci-Fi"))
[(genre, title, float(average)) for genre, title, average in movies.get_top_rated_movies_by_genre()][('Inception', 4.5), ('Up', 3.0)]
['Inception', 'Solaris', 'Tenet']
[('Drama', 'Inception', 4.5), ('Sci-Fi', 'Inception', 4.5), ('Sci-Fi', 'Tenet', 2.5)]data = seed()
reviews = ReviewManager(session)
target = session.query(Review).filter_by(user_id=data["reza"].id).one()
changed = reviews.update(target.id, {"rating": 5, "comment": "changed my mind"})
[changed.rating, changed.comment]
reviews.update(999, {"rating": 1}) is None
reviews.delete(target.id)
reviews.delete(target.id)
session.query(Review).count()[5, 'changed my mind'] True True False 4
data = seed()
users = UserManager(session)
changed = users.update(data["ali"].id, {"name": "Ali Alavi", "email": "a.alavi@example.com"})
[changed.name, changed.email, changed.is_verified]
users.verify_user(data["sara"].id).is_verified
users.verify_user(999) is None
users.delete(data["mina"].id)
users.delete(data["mina"].id)
sorted(user.name for user in session.query(User).all())['Ali Alavi', 'a.alavi@example.com', False] True True True False ['Ali Alavi', 'Reza', 'Sara']
برای نوشتن پاسخ و دریافت امتیاز، ابتدا وارد حساب شوید.
ورود