Merge pull request #2790 from bagerard/idoshr-__raw__and_mongoengine

Clone [Idoshr raw and mongoengine]

Author: bagerard

docs/changelog.rst

@@ -8,6 +8,7 @@ Development
 ===========
 - (Fill this out as you fix issues and develop your features).
 - Fix for uuidRepresentation not read when provided in URI #2741
+- Fix combination of __raw__ and mongoengine syntax #2773
 - Add tests against MongoDB 6.0 and MongoDB 7.0 in the pipeline
 - Fix validate() not being called when inheritance is used in EmbeddedDocument and validate is overriden #2784
 - Add support for readPreferenceTags in connection parameters #2644

mongoengine/queryset/transform.py

@@ -62,14 +62,23 @@
 )
 
 
+def handle_raw_query(value, mongo_query):
+    """Combine a raw query with an existing one"""
+    for op, v in value.items():
+        if op not in mongo_query:
+            mongo_query[op] = v
+        elif op in mongo_query and isinstance(mongo_query[op], dict):
+            mongo_query[op].update(v)
+
+
 # TODO make this less complex
 def query(_doc_cls=None, **kwargs):
     """Transform a query from Django-style format to Mongo format."""
     mongo_query = {}
     merge_query = defaultdict(list)
     for key, value in sorted(kwargs.items()):
         if key == "__raw__":
-            mongo_query.update(value)
+            handle_raw_query(value, mongo_query)
             continue
 
         parts = key.rsplit("__")
@@ -234,7 +243,7 @@ def update(_doc_cls=None, **update):
 
     for key, value in update.items():
         if key == "__raw__":
-            mongo_update.update(value)
+            handle_raw_query(value, mongo_update)
             continue
 
         parts = key.split("__")

tests/queryset/test_queryset.py

@@ -2269,6 +2269,46 @@ class BlogPost(Document):
         post.reload()
         assert post.slug == "When test test it"
 
+    def test_combination_of_mongoengine_and__raw__(self):
+        """Ensure that the '__raw__' update/query works in combination with mongoengine syntax correctly."""
+
+        class BlogPost(Document):
+            slug = StringField()
+            foo = StringField()
+            tags = ListField(StringField())
+
+        BlogPost.drop_collection()
+
+        post = BlogPost(slug="test", foo="bar")
+        post.save()
+
+        BlogPost.objects(slug="test").update(
+            foo="baz",
+            __raw__={"$set": {"slug": "test test"}},
+        )
+        post.reload()
+        assert post.slug == "test test"
+        assert post.foo == "baz"
+
+        assert BlogPost.objects(foo="baz", __raw__={"slug": "test test"}).count() == 1
+        assert (
+            BlogPost.objects(foo__ne="bar", __raw__={"slug": {"$ne": "test"}}).count()
+            == 1
+        )
+        assert (
+            BlogPost.objects(foo="baz", __raw__={"slug": {"$ne": "test test"}}).count()
+            == 0
+        )
+        assert (
+            BlogPost.objects(foo__ne="baz", __raw__={"slug": "test test"}).count() == 0
+        )
+        assert (
+            BlogPost.objects(
+                foo__ne="baz", __raw__={"slug": {"$ne": "test test"}}
+            ).count()
+            == 0
+        )
+
     def test_add_to_set_each(self):
         class Item(Document):
             name = StringField(required=True)