mongodb/mongo-python-driver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix cursor cloning issue with deepcopy and regex #179

Browse Source
This commit is contained in:
Ross Lawley 2013-05-08 10:33:14 +00:00
parent 161543068e
commit 3f5cb6ff28
2 changed files with 26 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
pymongo/cursor.py
View File

@ -841,22 +841,33 @@ class Cursor(object):
return self.__clone(deepcopy=True)
def __deepcopy(self, x, memo=None):
"""Deepcopy helper for the data dictionary.
"""Deepcopy helper for the data dictionary or list.
Regular expressions cannot be deep copied but as they are immutable we
don't have to copy them when cloning.
"""
y = {}
if not hasattr(x, 'items'):
y, is_list, iterator = [], True, enumerate(x)
else:
y, is_list, iterator = {}, False, x.iteritems()
if memo is None:
memo = {}
val_id = id(x)
if val_id in memo:
return memo.get(val_id)
memo[val_id] = y
for key, value in x.iteritems():
if isinstance(value, dict) and not isinstance(value, SON):
for key, value in iterator:
if isinstance(value, (dict, list)) and not isinstance(value, SON):
value = self.__deepcopy(value, memo)
elif not isinstance(value, RE_TYPE):
value = copy.deepcopy(value, memo)
y[copy.deepcopy(key, memo)] = value
if is_list:
y.append(value)
else:
if not isinstance(key, RE_TYPE):
key = copy.deepcopy(key, memo)
y[key] = value
return y

10
test/test_cursor.py
View File

@ -529,6 +529,16 @@ class TestCursor(unittest.TestCase):
self.assertTrue(isinstance(cursor2._Cursor__hint, SON))
self.assertEqual(cursor._Cursor__hint, cursor2._Cursor__hint)
def test_deepcopy_cursor_littered_with_regexes(self):
cursor = self.db.test.find({"x": re.compile("^hmmm.*"),
"y": [re.compile("^hmm.*")],
"z": {"a": [re.compile("^hm.*")]},
re.compile("^key.*"): {"a": [re.compile("^hm.*")]}})
cursor2 = copy.deepcopy(cursor)
self.assertEqual(cursor._Cursor__spec, cursor2._Cursor__spec)
def test_add_remove_option(self):
cursor = self.db.test.find()
self.assertEqual(0, cursor._Cursor__query_options())