pub/mongoengine
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add _read_concern to copied properties. Add read_concern to aggregate. Add test to check the read_concern and read_preference values are kept after cloning.

Browse Source
This commit is contained in:
Agustin Barto 2020-05-20 18:56:13 -03:00
parent 4275c2d7b7
commit 1698f398eb
2 changed files with 30 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
mongoengine/queryset/base.py
View File

@ -789,6 +789,7 @@ class BaseQuerySet:
"_snapshot",
"_timeout",
"_read_preference",
"_read_concern",
"_iter",
"_scalar",
"_as_pymongo",
@ -1311,10 +1312,11 @@ class BaseQuerySet:
final_pipeline = initial_pipeline + user_pipeline
collection = self._collection
if self._read_preference is not None:
if self._read_preference is not None or self._read_concern is not None:
collection = self._collection.with_options(
read_preference=self._read_preference
read_preference=self._read_preference, read_concern=self._read_concern
)
return collection.aggregate(final_pipeline, cursor={}, **kwargs)
# JS functionality

26
tests/queryset/test_queryset.py
View File

@ -4021,6 +4021,32 @@ class TestQueryset(unittest.TestCase):
Number.drop_collection()
def test_clone_retains_settings(self):
"""Ensure that cloning retains the read_preference and read_concern
"""
class Number(Document):
n = IntField()
Number.drop_collection()
qs = Number.objects
qs_clone = qs.clone()
assert qs._read_preference == qs_clone._read_preference
assert qs._read_concern == qs_clone._read_concern
qs = Number.objects.read_preference(ReadPreference.PRIMARY_PREFERRED)
qs_clone = qs.clone()
assert qs._read_preference == ReadPreference.PRIMARY_PREFERRED
assert qs._read_preference == qs_clone._read_preference
qs = Number.objects.read_concern({'level': 'majority'})
qs_clone = qs.clone()
assert qs._read_concern.document == {'level': 'majority'}
assert qs._read_concern == qs_clone._read_concern
Number.drop_collection()
def test_using(self):
"""Ensure that switching databases for a queryset is possible
"""