Recommendation feedback [DB + API]

listenbrainz/db/model/recommendation_feedback.py

@@ -0,0 +1,57 @@
+import uuid
+
+from datetime import datetime
+from pydantic import BaseModel, ValidationError, validator
+
+
+def get_allowed_ratings():
+    """ Get rating values that can be submitted corresponding to a recommendation.
+    """
+    return ['like', 'love', 'dislike', 'hate', 'bad_recommendation']
+
+
+def check_recording_mbid_is_valid_uuid(rec_mbid):
+    try:
+        rec_mbid = uuid.UUID(rec_mbid)
+        return str(rec_mbid)
+    except (AttributeError, ValueError):
+        raise ValueError('Recording MBID must be a valid UUID.')
+
+
+class RecommendationFeedbackSubmit(BaseModel):
+    """ Represents a recommendation feedback submit object.
+        Args:
+            user_id: the row id of the user in the DB
+            recording_mbid: the MusicBrainz ID of the recording
+            rating: the feedback associated with the recommendation.
+                    Refer to "recommendation_feedback_type_enum" in admin/sql/create_types.py
+                    for allowed rating values.
+            created: (Optional)the timestamp when the feedback record was inserted into DB
+    """
+
+    user_id: int
+    recording_mbid: str
+    rating: str
+    created: datetime = None
+
+    @validator('rating')
+    def check_feedback_is_valid(cls, rating):
+        expected_rating = get_allowed_ratings()
+        if rating not in expected_rating:
+            raise ValueError('Feedback can only have a value in {}'.format(expected_rating))
+        return rating
+
+    _is_recording_mbid_valid: classmethod = validator("recording_mbid", allow_reuse=True)(check_recording_mbid_is_valid_uuid)
+
+
+class RecommendationFeedbackDelete(BaseModel):
+    """ Represents a recommendation feedback delete object.
+        Args:
+            user_id: the row id of the user in the DB
+            recording_mbid: the MusicBrainz ID of the recommendation
+    """
+
+    user_id: int
+    recording_mbid: str
+
+    _is_recording_mbid_valid: classmethod = validator("recording_mbid", allow_reuse=True)(check_recording_mbid_is_valid_uuid)

listenbrainz/db/recommendations_cf_recording_feedback.py

@@ -0,0 +1,107 @@
+import sqlalchemy
+
+from listenbrainz import db
+from listenbrainz.db.model.recommendation_feedback import (RecommendationFeedbackSubmit,
+                                                           RecommendationFeedbackDelete)
+from typing import List
+
+
+def insert(feedback_submit: RecommendationFeedbackSubmit):
+    """ Inserts a feedback record for a user's rated recommendation into the database.
+        If the record is already present for the user, the rating is updated to the new
+        value passed.
+
+        Args:
+            feedback_submit: An object of class RecommendationFeedbackSubmit
+    """
+
+    with db.engine.connect() as connection:
+        connection.execute(sqlalchemy.text("""
+            INSERT INTO recommendation_feedback (user_id, recording_mbid, rating)
+                 VALUES (:user_id, :recording_mbid, :rating)
+            ON CONFLICT (user_id, recording_mbid)
+          DO UPDATE SET rating = :rating,
+                        created = NOW()
+            """), {
+                'user_id': feedback_submit.user_id,
+                'recording_mbid': feedback_submit.recording_mbid,
+                'rating': feedback_submit.rating,
+            }
+        )
+
+
+def delete(feedback_delete: RecommendationFeedbackDelete):
+    """ Deletes the feedback record for a given recommendation for the user from the database
+
+        Args:
+            feedback_delete: An object of class RecommendationFeedbackDelete
+    """
+
+    with db.engine.connect() as connection:
+        connection.execute(sqlalchemy.text("""
+            DELETE FROM recommendation_feedback
+             WHERE user_id = :user_id
+               AND recording_mbid = :recording_mbid
+            """), {
+                'user_id': feedback_delete.user_id,
+                'recording_mbid': feedback_delete.recording_mbid,
+            }
+        )
+
+
+def get_feedback_for_user(user_id: int, limit: int, offset: int, rating: str = None) -> List[RecommendationFeedbackSubmit]:
+    """ Get a list of recommendation feedback given by the user in descending order of their creation.
+        Feedback will be filtered based on limit, offset and rating, if passed.
+
+        Args:
+            user_id: the row ID of the user in the DB
+            rating: the rating value by which the results are to be filtered.
+            limit: number of rows to be returned
+            offset: number of feedback to skip from the beginning
+
+        Returns:
+            A list of Feedback objects
+    """
+
+    args = {"user_id": user_id, "limit": limit, "offset": offset}
+    query = """ SELECT user_id,
+                       recording_mbid::text,
+                       rating,
+                       created
+                  FROM recommendation_feedback
+                 WHERE user_id = :user_id """
+
+    if rating:
+        query += " AND rating = :rating"
+        args["rating"] = rating
+
+    query += """ ORDER BY created DESC
+                 LIMIT :limit
+                 OFFSET :offset """
+
+    with db.engine.connect() as connection:
+        result = connection.execute(sqlalchemy.text(query), args)
+        return [RecommendationFeedbackSubmit(**dict(row)) for row in result.fetchall()]
+
+
+def get_feedback_count_for_user(user_id: int) -> int:
+    """ Get total number of recommendation feedback given by the user
+
+        Args:
+            user_id: the row ID of the user in the DB
+
+        Returns:
+            The total number of recommendation feedback given by the user
+    """
+    with db.engine.connect() as connection:
+        result = connection.execute(sqlalchemy.text("""
+            SELECT count(*) AS count
+              FROM recommendation_feedback
+             WHERE user_id = :user_id
+            """), {
+                'user_id': user_id,
+            }
+        )
+        count = int(result.fetchone()["count"])
+
+    return count

listenbrainz/db/tests/test_recommendations_cf_recording_feedback.py

@@ -0,0 +1,195 @@
+# -*- coding: utf-8 -*-
+import json
+import os
+import uuid
+from listenbrainz.db.model.recommendation_feedback import (RecommendationFeedbackSubmit,
+                                                           RecommendationFeedbackDelete,
+                                                           get_allowed_ratings)
+import listenbrainz.db.recommendations_cf_recording_feedback as db_feedback
+import listenbrainz.db.user as db_user
+
+from listenbrainz.db.testing import DatabaseTestCase
+
+
+class RecommendationFeedbackDatabaseTestCase(DatabaseTestCase):
+
+    def setUp(self):
+        DatabaseTestCase.setUp(self)
+        self.user = db_user.get_or_create(1, "vansika")
+        self.user1 = db_user.get_or_create(2, "vansika_1")
+        self.user2 = db_user.get_or_create(3, "vansika__2")
+
+        self.sample_feedback = [
+            {
+                "recording_mbid": "d23f4719-9212-49f0-ad08-ddbfbfc50d6f",
+                "rating": 'love',
+                'user_id': self.user['id']
+            },
+            {
+                "recording_mbid": "222eb00d-9ead-42de-aec9-8f8c1509413d",
+                "rating": 'bad_recommendation',
+                "user_id": self.user1['id']
+            },
+            {
+                "recording_mbid": "922eb00d-9ead-42de-aec9-8f8c1509413d",
+                "rating": 'hate',
+                "user_id": self.user1['id']
+            }
+        ]
+
+    def insert_test_data(self):
+        """ Insert test data into the database """
+
+        for fb in self.sample_feedback:
+            db_feedback.insert(
+                RecommendationFeedbackSubmit(
+                    user_id=fb['user_id'],
+                    recording_mbid=fb["recording_mbid"],
+                    rating=fb["rating"]
+                )
+            )
+
+    def test_insert(self):
+        self.insert_test_data()
+        result = db_feedback.get_feedback_for_user(user_id=self.user['id'], limit=25, offset=0)
+        self.assertEqual(len(result), 1)
+
+        result = db_feedback.get_feedback_for_user(user_id=self.user1['id'], limit=25, offset=0)
+        self.assertEqual(len(result), 2)
+
+    def test_update_rating_when_feedback_already_exits(self):
+        update_fb = self.sample_feedback[0]
+
+        self.insert_test_data()
+        result = db_feedback.get_feedback_for_user(user_id=self.user["id"], limit=25, offset=0)
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0].recording_mbid, update_fb['recording_mbid'])
+        self.assertEqual(result[0].rating, 'love')
+
+        new_rating = "like"  # change the score to -1
+
+        # update a record by inserting a record with updated score value
+        db_feedback.insert(
+            RecommendationFeedbackSubmit(
+                user_id=self.user["id"],
+                recording_mbid=update_fb["recording_mbid"],
+                rating=new_rating
+            )
+        )
+
+        result = db_feedback.get_feedback_for_user(user_id=self.user["id"], limit=25, offset=0)
+        self.assertEqual(len(result), 1)
+
+        self.assertEqual(result[0].recording_mbid, update_fb["recording_mbid"])
+        self.assertEqual(result[0].rating, 'like')
+
+    def test_delete(self):
+        del_fb = self.sample_feedback[1]
+
+        self.insert_test_data()
+        result = db_feedback.get_feedback_for_user(user_id=self.user1["id"], limit=25, offset=0)
+        self.assertEqual(len(result), 2)
+
+        db_feedback.delete(
+            RecommendationFeedbackDelete(
+                user_id=self.user1["id"],
+                recording_mbid=del_fb["recording_mbid"],
+            )
+        )
+
+        result = db_feedback.get_feedback_for_user(user_id=self.user1["id"], limit=25, offset=0)
+        self.assertEqual(len(result), 1)
+
+        self.assertNotEqual(result[0].recording_mbid, del_fb["recording_mbid"])
+
+    def test_get_feedback_for_user(self):
+        self.insert_test_data()
+        result = db_feedback.get_feedback_for_user(user_id=self.user["id"], limit=25, offset=0)
+        self.assertEqual(len(result), 1)
+
+        self.assertEqual(result[0].user_id, self.user["id"])
+        self.assertEqual(result[0].recording_mbid, self.sample_feedback[0]["recording_mbid"])
+        self.assertEqual(result[0].rating, self.sample_feedback[0]["rating"])
+
+        feedback_love = []
+        for i in range(60):
+            submit_obj = RecommendationFeedbackSubmit(
+                user_id=self.user2['id'],
+                recording_mbid=str(uuid.uuid4()),
+                rating='love'
+            )
+
+            db_feedback.insert(submit_obj)
+            # prepended to the list since ``get_feedback_for_users`` returns data in descending
+            # order of creation.
+            feedback_love.insert(0, submit_obj)
+
+        feedback_hate = []
+        for i in range(50):
+            submit_obj = RecommendationFeedbackSubmit(
+                user_id=self.user2['id'],
+                recording_mbid=str(uuid.uuid4()),
+                rating='hate'
+            )
+
+            db_feedback.insert(submit_obj)
+            # prepended to the list since ``get_feedback_for_users`` returns data in descending
+            # order of creation.
+            feedback_hate.insert(0, submit_obj)
+        # ``get_feddback_for_user`` will return feedback_hate data followed by feedback_love
+        # data
+        result = db_feedback.get_feedback_for_user(user_id=self.user2['id'], limit=120, offset=0)
+        self.assertEqual(len(result), 110)
+
+        # test the rating argument
+        result = db_feedback.get_feedback_for_user(user_id=self.user2['id'], limit=70, offset=0, rating='love')
+        self.assertEqual(len(result), 60)
+        for i in range(60):
+            self.assertEqual(result[i].user_id, feedback_love[i].user_id)
+            self.assertEqual(result[i].recording_mbid, feedback_love[i].recording_mbid)
+            self.assertEqual(result[i].rating, feedback_love[i].rating)
+
+        result = db_feedback.get_feedback_for_user(user_id=self.user2['id'], limit=70, offset=0, rating='hate')
+        self.assertEqual(len(result), 50)
+        for i in range(50):
+            self.assertEqual(result[i].user_id, feedback_hate[i].user_id)
+            self.assertEqual(result[i].recording_mbid, feedback_hate[i].recording_mbid)
+            self.assertEqual(result[i].rating, feedback_hate[i].rating)
+
+        # test the limit argument
+        result = db_feedback.get_feedback_for_user(user_id=self.user2['id'], limit=20, offset=0, rating='love')
+        self.assertEqual(len(result), 20)
+        for i in range(20):
+            self.assertEqual(result[i].user_id, feedback_love[i].user_id)
+            self.assertEqual(result[i].recording_mbid, feedback_love[i].recording_mbid)
+            self.assertEqual(result[i].rating, feedback_love[i].rating)
+
+        # test the offset argument
+        result = db_feedback.get_feedback_for_user(user_id=self.user2['id'], limit=25, offset=10)
+        self.assertEqual(len(result), 25)
+        for i in range(25):
+            self.assertEqual(result[i].user_id, feedback_hate[i+10].user_id)
+            self.assertEqual(result[i].recording_mbid, feedback_hate[i+10].recording_mbid)
+            self.assertEqual(result[i].rating, feedback_hate[i+10].rating)
+
+        result = db_feedback.get_feedback_for_user(user_id=self.user2['id'], limit=25, offset=100)
+        self.assertEqual(len(result), 10)
+        for i in range(10):
+            self.assertEqual(result[i].user_id, feedback_love[i+50].user_id)
+            self.assertEqual(result[i].recording_mbid, feedback_love[i+50].recording_mbid)
+            self.assertEqual(result[i].rating, feedback_love[i+50].rating)
+
+        result = db_feedback.get_feedback_for_user(user_id=self.user2['id'], limit=30, offset=110)
+        self.assertEqual(len(result), 0)
+
+        result = db_feedback.get_feedback_for_user(user_id=self.user2['id'], limit=30, offset=30, rating='hate')
+        self.assertEqual(len(result), 20)
+        for i in range(20):
+            self.assertEqual(result[i].user_id, feedback_hate[i+30].user_id)
+            self.assertEqual(result[i].recording_mbid, feedback_hate[i+30].recording_mbid)
+            self.assertEqual(result[i].rating, feedback_hate[i+30].rating)
+
+    def test_get_feedback_count_for_user(self):
+        self.insert_test_data()
+        result = db_feedback.get_feedback_count_for_user(user_id=self.user1["id"])
+        self.assertEqual(result, 2)