⚡ Bolt: Optimize database queries and remove redundant loops

.jules/bolt.md

@@ -0,0 +1,3 @@
+## 2025-01-30 - [Flask/SQLAlchemy Performance Anti-patterns]
+**Learning:** Using `len(User.query.all()) > 0` or `for user in User.query.all(): if user.username == ...` are major bottlenecks in Flask-SQLAlchemy because they load all records into memory. Also, found a redundant nested loop `for key in dict_item` that did not use the key, multiplying processing time by the number of keys in the dictionary.
+**Action:** Use `User.query.first()` for existence checks, `filter_by().first()` for targeted searches, and `db.session.query(User).delete()` for bulk deletions. Always check if nested loops over dictionaries actually need to iterate over keys.

flask_app.py

@@ -62,9 +62,10 @@ def logout():
         if request.form["username"] == "ALL":
             clear_users()
         else:
-            for user in User.query.all():
-                if user.username == request.form["username"]:
-                    db.session.delete(user)
+            # Performance: Use database filter instead of linear scan of all users
+            user = User.query.filter_by(username=request.form["username"]).first()
+            if user:
+                db.session.delete(user)
             try:
                 db.session.commit()
             except:
@@ -120,13 +121,15 @@ def party():
     item = "Nothing is Playing"
     playback = None
 
-    if len(User.query.all()) > 0:
-        sp = spotipy.Spotify(auth=User.query.first().token)
+    # Performance: Avoid loading all users into memory to check for existence
+    user = User.query.first()
+    if user:
+        sp = spotipy.Spotify(auth=user.token)
 
-    try:
-        playback = sp.current_playback()
-    except:
-        clear_users()
+        try:
+            playback = sp.current_playback()
+        except:
+            clear_users()
 
     if playback != None:
         url = playback["item"]["album"]["images"][0]["url"]
@@ -154,16 +157,14 @@ def users():
 
     for dict_item in playback:
         if dict_item != None:
-            for key in dict_item:
-                item = (
-                    dict_item["device"]["name"]
-                    + " - "
-                    + dict_item["item"]["name"]
-                    # + " by "
-                    # + dict_item["item"]["artists"][0]["name"]
-                )
-                if item not in listeners:
-                    listeners.append(item)
+            # Performance: Removed redundant loop over dictionary keys
+            item = (
+                dict_item["device"]["name"]
+                + " - "
+                + dict_item["item"]["name"]
+            )
+            if item not in listeners:
+                listeners.append(item)
 
     return render_template("users.html", users=users, listeners=listeners)
 
@@ -253,9 +254,8 @@ def list_users():
 
 @app.route("/clear_users")
 def clear_users():
-    for user in User.query.all():
-        db.session.delete(user)
-
+    # Performance: Bulk delete instead of looping through all users
+    db.session.query(User).delete()
     db.session.commit()
 
     return redirect("/")

test_app.py

@@ -0,0 +1,65 @@
+import unittest
+from flask_app import app, db, User
+
+class SpotifyPartyTestCase(unittest.TestCase):
+    def setUp(self):
+        app.config['TESTING'] = True
+        app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
+        self.app = app.test_client()
+        with app.app_context():
+            db.create_all()
+
+    def tearDown(self):
+        with app.app_context():
+            db.session.remove()
+            db.drop_all()
+
+    def test_party_page_empty(self):
+        response = self.app.get('/')
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b'Nothing is Playing', response.data)
+
+    def test_party_page_with_user(self):
+        with app.app_context():
+            user = User(username='testuser', token='testtoken')
+            db.session.add(user)
+            db.session.commit()
+        # Note: Spotify API calls will fail, but we're testing the DB logic
+        # and the route's resilience.
+        response = self.app.get('/')
+        self.assertEqual(response.status_code, 200)
+
+    def test_list_users(self):
+        with app.app_context():
+            user = User(username='testuser', token='testtoken')
+            db.session.add(user)
+            db.session.commit()
+        response = self.app.get('/list_users')
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b'testuser', response.data)
+
+    def test_logout(self):
+        with app.app_context():
+            user = User(username='testuser', token='testtoken')
+            db.session.add(user)
+            db.session.commit()
+
+        # Test logout specific user
+        response = self.app.post('/logout', data={'username': 'testuser'}, follow_redirects=True)
+        self.assertEqual(response.status_code, 200)
+        with app.app_context():
+            self.assertIsNone(User.query.filter_by(username='testuser').first())
+
+    def test_clear_users(self):
+        with app.app_context():
+            user = User(username='testuser', token='testtoken')
+            db.session.add(user)
+            db.session.commit()
+
+        response = self.app.get('/clear_users', follow_redirects=True)
+        self.assertEqual(response.status_code, 200)
+        with app.app_context():
+            self.assertEqual(User.query.count(), 0)
+
+if __name__ == '__main__':
+    unittest.main()