Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Support multiple operators in nested query documents

Fix the query parser to allow multiple operators like:
{'$gte': high_value, '$lte': low_value}

Signed-off-by: Doug Hellmann <doug.hellmann@dreamhost.com>
  • Loading branch information...
commit d489adbeb66b8315808732d0302ec8f1d932a860 1 parent 05e33a9
Doug Hellmann dhellmann authored rick446 committed
Showing with 44 additions and 6 deletions.
  1. +6 −6 ming/mim.py
  2. +38 −0 ming/tests/test_mim.py
12 ming/mim.py
View
@@ -523,18 +523,18 @@ def match(spec, doc):
else:
return False
else:
- op, value = _parse_query(v)
- if not _part_match(op, value, k.split('.'), doc):
- return False
+ for op, value in _parse_query(v):
+ if not _part_match(op, value, k.split('.'), doc):
+ return False
except (AttributeError, KeyError), ex:
return False
return True
def _parse_query(v):
- if isinstance(v, dict) and len(v) == 1 and v.keys()[0].startswith('$'):
- return v.keys()[0], v.values()[0]
+ if isinstance(v, dict) and all(k.startswith('$') for k in v.keys()):
+ return v.items()
else:
- return '$eq', v
+ return [('$eq', v)]
def _part_match(op, value, key_parts, doc, allow_list_compare=True):
if not key_parts:
38 ming/tests/test_mim.py
View
@@ -8,6 +8,44 @@ def setUp(self):
self.bind = create_datastore('mim:///testdb')
self.bind.conn.drop_all()
self.bind.db.coll.insert({'_id':'foo', 'a':2, 'c':[1,2,3]})
+ for r in range(4):
+ self.bind.db.rcoll.insert({'_id':'r%s' % r, 'd':r})
+
+ def test_gt(self):
+ f = self.bind.db.rcoll.find
+ assert 1 == f(dict(d={'$gt': 2})).count()
+ assert 0 == f(dict(d={'$gt': 3})).count()
+
+ def test_gte(self):
+ f = self.bind.db.rcoll.find
+ assert 2 == f(dict(d={'$gte': 2})).count()
+ assert 1 == f(dict(d={'$gte': 3})).count()
+
+ def test_lt(self):
+ f = self.bind.db.rcoll.find
+ assert 0 == f(dict(d={'$lt': 0})).count()
+ assert 1 == f(dict(d={'$lt': 1})).count()
+ assert 2 == f(dict(d={'$lt': 2})).count()
+
+ def test_lte(self):
+ f = self.bind.db.rcoll.find
+ assert 1 == f(dict(d={'$lte': 0})).count()
+ assert 2 == f(dict(d={'$lte': 1})).count()
+ assert 3 == f(dict(d={'$lte': 2})).count()
+
+ def test_range_equal(self):
+ f = self.bind.db.rcoll.find
+ assert 1 == f(dict(d={'$gte': 2, '$lte': 2})).count()
+ assert 2 == f(dict(d={'$gte': 1, '$lte': 2})).count()
+ assert 0 == f(dict(d={'$gte': 4, '$lte': -1})).count()
+
+ def test_range_inequal(self):
+ f = self.bind.db.rcoll.find
+ assert 0 == f(dict(d={'$gt': 2, '$lt': 2})).count()
+ assert 1 == f(dict(d={'$gt': 2, '$lt': 4})).count()
+ assert 0 == f(dict(d={'$gt': 1, '$lt': 2})).count()
+ assert 1 == f(dict(d={'$gt': 1, '$lt': 3})).count()
+ assert 0 == f(dict(d={'$gt': 4, '$lt': -1})).count()
def test_exists(self):
f = self.bind.db.coll.find
Please sign in to comment.
Something went wrong with that request. Please try again.