Refactor.

Author: WaVEV

django_mongodb_backend/expressions/builtins.py

@@ -6,13 +6,15 @@
 from django.core.exceptions import EmptyResultSet, FullResultSet
 from django.db import NotSupportedError
 from django.db.models.expressions import (
+    BaseExpression,
     Case,
     Col,
     ColPairs,
     CombinedExpression,
     Exists,
     ExpressionList,
     ExpressionWrapper,
+    Func,
     NegatedExpression,
     OrderBy,
     RawSQL,
@@ -23,9 +25,12 @@
     Value,
     When,
 )
+from django.db.models.fields.json import KeyTransform
 from django.db.models.sql import Query
 
-from ..query_utils import process_lhs
+from django_mongodb_backend.fields.array import Array
+
+from ..query_utils import is_direct_value, process_lhs
 
 
 # EXTRA IS TOTALLY IGNORED
@@ -234,6 +239,36 @@ def value(self, compiler, connection, as_path=False):  # noqa: ARG001
     return value
 
 
+@staticmethod
+def _is_constant_value(value):
+    if isinstance(value, Array):
+        return all(_is_constant_value(e) for e in value.get_source_expressions())
+    if isinstance(value, Value) or is_direct_value(value):
+        v = value.value if isinstance(value, Value) else value
+        return not isinstance(v, str) or "." not in v
+    return isinstance(value, Func | Value) and not (
+        value.contains_aggregate
+        or value.contains_over_clause
+        or value.contains_column_references
+        or value.contains_subquery
+    )
+
+
+@staticmethod
+def _is_simple_column(lhs):
+    while isinstance(lhs, KeyTransform):
+        if "." in getattr(lhs, "key_name", ""):
+            return False
+        lhs = lhs.lhs
+    col = lhs.source if isinstance(lhs, Ref) else lhs
+    # Foreign columns from parent cannot be addressed as single match
+    return isinstance(col, Col) and col.alias is not None
+
+
+def is_simple_expression(self):
+    return self.is_simple_column(self.lhs) and self.is_constant_value(self.rhs)
+
+
 def register_expressions():
     Case.as_mql = case
     Col.as_mql = col
@@ -252,3 +287,6 @@ def register_expressions():
     Subquery.as_mql = subquery
     When.as_mql = when
     Value.as_mql = value
+    BaseExpression.is_simple_expression = is_simple_expression
+    BaseExpression.is_simple_column = _is_simple_column
+    BaseExpression.is_constant_value = _is_constant_value

django_mongodb_backend/fields/array.py

@@ -7,7 +7,6 @@
 from django.utils.translation import gettext_lazy as _
 
 from ..forms import SimpleArrayField
-from ..lookups import is_constant_value, is_simple_column
 from ..query_utils import process_lhs, process_rhs
 from ..utils import prefix_validation_error
 from ..validators import ArrayMaxLengthValidator, LengthValidator
@@ -256,7 +255,7 @@ class ArrayContains(ArrayRHSMixin, FieldGetDbPrepValueMixin, Lookup):
     lookup_name = "contains"
 
     def as_mql(self, compiler, connection, as_path=False):
-        if as_path and is_simple_column(self.lhs) and is_constant_value(self.rhs):
+        if as_path and self.is_simple_expression():
             lhs_mql = process_lhs(self, compiler, connection, as_path=as_path)
             value = process_rhs(self, compiler, connection, as_path=as_path)
             if value is None:
@@ -346,7 +345,7 @@ def get_subquery_wrapping_pipeline(self, compiler, connection, field_name, expr)
         ]
 
     def as_mql(self, compiler, connection, as_path=False):
-        if as_path and is_simple_column(self.lhs) and is_constant_value(self.rhs):
+        if as_path and self.is_simple_expression():
             lhs_mql = process_lhs(self, compiler, connection, as_path=True)
             value = process_rhs(self, compiler, connection, as_path=True)
             return {lhs_mql: {"$in": value}}

django_mongodb_backend/fields/json.py

@@ -1,7 +1,6 @@
 from itertools import chain
 
 from django.db import NotSupportedError
-from django.db.models.expressions import Value
 from django.db.models.fields.json import (
     ContainedBy,
     DataContains,
@@ -17,7 +16,7 @@
     KeyTransformNumericLookupMixin,
 )
 
-from ..lookups import builtin_lookup, is_constant_value, is_simple_column
+from ..lookups import builtin_lookup
 from ..query_utils import process_lhs, process_rhs
 
 
@@ -72,23 +71,13 @@ def _has_key_predicate(path, root_column=None, negated=False, as_path=False):
     return result
 
 
-def has_key_lookup_rhs_check(rhs):
-    for key in rhs:
-        if not is_constant_value(key):
-            return False
-        value = key.value if isinstance(key, Value) else key
-        if isinstance(value, str) and "." in value:
-            return False
-    return True
-
-
 def has_key_lookup(self, compiler, connection, as_path=False):
     """Return MQL to check for the existence of a key."""
     rhs = self.rhs
     lhs = process_lhs(self, compiler, connection)
     if not isinstance(rhs, (list, tuple)):
         rhs = [rhs]
-    as_path = as_path and is_simple_column(self.lhs) and has_key_lookup_rhs_check(rhs)
+    as_path = as_path and self.is_simple_expression()
     paths = []
     # Transform any "raw" keys into KeyTransforms to allow consistent handling
     # in the code that follows.
@@ -132,7 +121,7 @@ def key_transform(self, compiler, connection, as_path=False):
     while isinstance(previous, KeyTransform):
         key_transforms.insert(0, previous.key_name)
         previous = previous.lhs
-    if as_path and is_simple_column(self.lhs):
+    if as_path and self.is_simple_column(self.lhs):
         lhs_mql = previous.as_mql(compiler, connection, as_path=True)
         return build_json_mql_path(lhs_mql, key_transforms, as_path=True)
     # Collect all key transforms in order.
@@ -147,7 +136,7 @@ def key_transform_in(self, compiler, connection, as_path=False):
     Return MQL to check if a JSON path exists and that its values are in the
     set of specified values (rhs).
     """
-    if as_path and is_simple_column(self.lhs) and is_constant_value(self.rhs):
+    if as_path and self.is_simple_expression():
         return builtin_lookup(self, compiler, connection, as_path=True)
 
     lhs_mql = process_lhs(self, compiler, connection)
@@ -175,7 +164,7 @@ def key_transform_is_null(self, compiler, connection, as_path=False):
 
     Reference: https://code.djangoproject.com/ticket/32252
     """
-    if as_path and is_simple_column(self.lhs) and is_constant_value(self.rhs):
+    if as_path and self.is_simple_expression():
         lhs_mql = process_lhs(self, compiler, connection, as_path=True)
         rhs_mql = process_rhs(self, compiler, connection)
         return _has_key_predicate(lhs_mql, None, negated=rhs_mql, as_path=True)
@@ -195,7 +184,7 @@ def key_transform_numeric_lookup_mixin(self, compiler, connection, as_path=False
     Return MQL to check if the field exists (i.e., is not "missing" or "null")
     and that the field matches the given numeric lookup expression.
     """
-    if is_simple_column(self.lhs) and is_constant_value(self.rhs) and as_path:
+    if as_path and self.is_simple_expression():
         return builtin_lookup(self, compiler, connection, as_path=True)
 
     lhs = process_lhs(self, compiler, connection, as_path=False)
@@ -209,7 +198,7 @@ def key_transform_numeric_lookup_mixin(self, compiler, connection, as_path=False
 
 
 def key_transform_exact(self, compiler, connection, as_path=False):
-    if is_simple_column(self.lhs) and is_constant_value(self.rhs) and as_path:
+    if as_path and self.is_simple_expression():
         lhs_mql = process_lhs(self, compiler, connection, as_path=True)
         return {
             "$and": [

django_mongodb_backend/functions.py

@@ -39,7 +39,6 @@
     Upper,
 )
 
-from .lookups import is_constant_value
 from .query_utils import process_lhs
 
 MONGO_OPERATORS = {
@@ -167,7 +166,7 @@ def preserve_null(operator):
     # If the argument is null, the function should return null, not
     # $toLower/Upper's behavior of returning an empty string.
     def wrapped(self, compiler, connection, as_path=False):
-        if is_constant_value(self.lhs) and as_path:
+        if as_path and self.is_constant_value(self.lhs):
             if self.lhs is None:
                 return None
             lhs_mql = process_lhs(self, compiler, connection, as_path=True)

django_mongodb_backend/lookups.py

@@ -1,6 +1,4 @@
 from django.db import NotSupportedError
-from django.db.models.expressions import Col, Func, Ref, Value
-from django.db.models.fields.json import KeyTransform
 from django.db.models.fields.related_lookups import In, RelatedIn
 from django.db.models.lookups import (
     BuiltinLookup,
@@ -10,38 +8,11 @@
     UUIDTextMixin,
 )
 
-from .query_utils import is_direct_value, process_lhs, process_rhs
-
-
-def is_constant_value(value):
-    from django_mongodb_backend.fields.array import Array  # noqa: PLC0415
-
-    if isinstance(value, Array):
-        return all(is_constant_value(e) for e in value.get_source_expressions())
-
-    return is_direct_value(value) or (
-        isinstance(value, Func | Value)
-        and not (
-            value.contains_aggregate
-            or value.contains_over_clause
-            or value.contains_column_references
-            or value.contains_subquery
-        )
-    )
-
-
-def is_simple_column(lhs):
-    while isinstance(lhs, KeyTransform):
-        if "." in lhs.key_name:
-            return False
-        lhs = lhs.lhs
-    col = lhs.source if isinstance(lhs, Ref) else lhs
-    # Foreign columns from parent cannot be addressed as single match
-    return isinstance(col, Col) and col.alias is not None
+from .query_utils import process_lhs, process_rhs
 
 
 def builtin_lookup(self, compiler, connection, as_path=False):
-    if is_simple_column(self.lhs) and is_constant_value(self.rhs) and as_path:
+    if as_path and self.is_simple_expression():
         lhs_mql = process_lhs(self, compiler, connection, as_path=True)
         value = process_rhs(self, compiler, connection, as_path=True)
         return connection.mongo_operators_match[self.lookup_name](lhs_mql, value)
@@ -114,7 +85,7 @@ def get_subquery_wrapping_pipeline(self, compiler, connection, field_name, expr)
 def is_null(self, compiler, connection, as_path=False):
     if not isinstance(self.rhs, bool):
         raise ValueError("The QuerySet value for an isnull lookup must be True or False.")
-    if is_constant_value(self.rhs) and as_path and is_simple_column(self.lhs):
+    if as_path and self.is_simple_expression():
         lhs_mql = process_lhs(self, compiler, connection, as_path=as_path)
         return connection.mongo_operators_match["isnull"](lhs_mql, self.rhs)
     lhs_mql = process_lhs(self, compiler, connection, as_path=False)