Add NoneQuery

This makes fast SQL queries for singletons possible

Author: Thomas Scholtes

beets/dbcore/query.py

@@ -83,6 +83,23 @@ def value_match(cls, pattern, value):
         return pattern == value
 
 
+class NoneQuery(FieldQuery):
+
+    def __init__(self, field, fast=True):
+        self.field = field
+        self.fast = fast
+
+    def col_clause(self):
+        return self.field + " IS NULL", ()
+
+    @classmethod
+    def match(self, item):
+        try:
+            return item[self.field] is None
+        except KeyError:
+            return True
+
+
 class StringFieldQuery(FieldQuery):
     """A FieldQuery that converts values to strings before matching
     them.

test/test_query.py

@@ -16,14 +16,26 @@
 """
 import _common
 from _common import unittest
-from helper import TestHelper
+import helper
 
 import beets.library
 from beets import dbcore
 from beets.dbcore import types
+from beets.dbcore.query import NoneQuery
 from beets.library import Library, Item
 
 
+class TestHelper(helper.TestHelper):
+
+    def assertInResult(self, item, results):
+        result_ids = map(lambda i: i.id, results)
+        self.assertIn(item.id, result_ids)
+
+    def assertNotInResult(self, item, results):
+        result_ids = map(lambda i: i.id, results)
+        self.assertNotIn(item.id, result_ids)
+
+
 class AnyFieldQueryTest(_common.LibTestCase):
     def test_no_restriction(self):
         q = dbcore.query.AnyFieldQuery(
@@ -469,14 +481,6 @@ def test_flex_parse_any_string(self):
         self.assertInResult(item_false, matched)
         self.assertNotInResult(item_true, matched)
 
-    def assertInResult(self, item, results):
-        result_ids = map(lambda i: i.id, results)
-        self.assertIn(item.id, result_ids)
-
-    def assertNotInResult(self, item, results):
-        result_ids = map(lambda i: i.id, results)
-        self.assertNotIn(item.id, result_ids)
-
 
 class DefaultSearchFieldsTest(DummyDataTestCase):
     def test_albums_matches_album(self):
@@ -496,6 +500,30 @@ def test_items_does_not_match_year(self):
         self.assert_matched(items, [])
 
 
+class NoneQueryTest(unittest.TestCase, TestHelper):
+
+    def setUp(self):
+        self.lib = Library(':memory:')
+
+    def test_match_singletons(self):
+        singleton = self.add_item()
+        album_item = self.add_album().items().get()
+
+        matched = self.lib.items(NoneQuery('album_id'))
+        self.assertInResult(singleton, matched)
+        self.assertNotInResult(album_item, matched)
+
+    def test_match_after_set_none(self):
+        item = self.add_item(rg_track_gain=0)
+        matched = self.lib.items(NoneQuery('rg_track_gain'))
+        self.assertNotInResult(item, matched)
+
+        item['rg_track_gain'] = None
+        item.store()
+        matched = self.lib.items(NoneQuery('rg_track_gain'))
+        self.assertInResult(item, matched)
+
+
 def suite():
     return unittest.TestLoader().loadTestsFromName(__name__)