Refs #32519 -- Added JSONSet and JSONRemove functions.

django/db/backends/base/features.py

@@ -357,6 +357,15 @@
     has_json_object_function = True
     # Does the backend support negative JSON array indexing?
     supports_json_negative_indexing = True
+    # Does the backend support partial updates to JSONField?
+    supports_partial_json_update = True
+    # Does JSONSet only append to an array if the index equals to the array length?
+    json_set_array_append_requires_length_as_index = False
+    # Does JSONSet wrap an existing non-array value into an array when the path is an
+    # array index?
+    json_set_wraps_non_array_to_array = False
+    # Does JSONSet create missing parent path when it does not exist?
+    json_set_creates_missing_parent_path = False
 
     # Does the backend support column collations?
     supports_collation_on_charfield = True

django/db/backends/mysql/features.py

@@ -59,6 +59,7 @@ class DatabaseFeatures(BaseDatabaseFeatures):
     supports_virtual_generated_columns = True
 
     supports_json_negative_indexing = False
+    json_set_wraps_non_array_to_array = True
 
     @cached_property
     def minimum_database_version(self):

django/db/backends/oracle/features.py

@@ -219,3 +219,7 @@ def bare_select_suffix(self):
     def supports_tuple_lookups(self):
         # Support is known to be missing on 23.2 but available on 23.4.
         return self.connection.oracle_version >= (23, 4)
+
+    @cached_property
+    def supports_partial_json_update(self):
+        return self.connection.oracle_version >= (21,)

django/db/backends/sqlite3/features.py

@@ -57,6 +57,8 @@ class DatabaseFeatures(BaseDatabaseFeatures):
     insert_test_table_with_defaults = 'INSERT INTO {} ("null") VALUES (1)'
     supports_default_keyword_in_insert = False
     supports_unlimited_charfield = True
+    json_set_array_append_requires_length_as_index = True
+    json_set_creates_missing_parent_path = True
 
     @cached_property
     def django_test_skips(self):
@@ -126,6 +128,10 @@ def django_test_skips(self):
                     "notation": {
                         "model_fields.test_jsonfield.TestQuerying."
                         "test_lookups_special_chars_double_quotes",
+                        "db_functions.json.test_json_set.JSONSetTests."
+                        "test_set_special_chars_double_quotes",
+                        "db_functions.json.test_json_remove.JSONRemoveTests."
+                        "test_remove_special_chars_double_quotes",
                     },
                 }
             )

django/db/models/functions/__init__.py

@@ -25,7 +25,7 @@
     TruncWeek,
     TruncYear,
 )
-from .json import JSONArray, JSONObject
+from .json import JSONArray, JSONObject, JSONRemove, JSONSet
 from .math import (
     Abs,
     ACos,
@@ -128,6 +128,8 @@
     # json
     "JSONArray",
     "JSONObject",
+    "JSONRemove",
+    "JSONSet",
     # math
     "Abs",
     "ACos",

django/db/models/functions/comparison.py

@@ -19,6 +19,7 @@ def as_sql(self, compiler, connection, **extra_context):
 
     def as_sqlite(self, compiler, connection, **extra_context):
         db_type = self.output_field.db_type(connection)
+        output_type = self.output_field.get_internal_type()
         if db_type in {"datetime", "time"}:
             # Use strftime as datetime/time don't keep fractional seconds.
             template = "strftime(%%s, %(expressions)s)"
@@ -33,6 +34,11 @@ def as_sqlite(self, compiler, connection, **extra_context):
             return super().as_sql(
                 compiler, connection, template=template, **extra_context
             )
+        elif output_type == "JSONField":
+            template = "JSON(%(expressions)s)"
+            return super().as_sql(
+                compiler, connection, template=template, **extra_context
+            )
         return self.as_sql(compiler, connection, **extra_context)
 
     def as_mysql(self, compiler, connection, **extra_context):

django/db/models/functions/json.py

@@ -1,7 +1,8 @@
 from django.db import NotSupportedError
+from django.db.models.constants import LOOKUP_SEP
 from django.db.models.expressions import Func, Value
 from django.db.models.fields import TextField
-from django.db.models.fields.json import JSONField
+from django.db.models.fields.json import JSONField, compile_json_path
 from django.db.models.functions import Cast
 
 
@@ -122,3 +123,244 @@
 
     def as_oracle(self, compiler, connection, **extra_context):
         return self.as_native(compiler, connection, returning="CLOB", **extra_context)
+
+
+class ToJSONB(Func):
+    function = "TO_JSONB"
+
+
+class JSONSet(Func):
+    def __init__(self, expression, output_field=None, **fields):
+        if not fields:
+            raise TypeError("JSONSet requires at least one key-value pair to be set.")
+        self.fields = fields
+        super().__init__(expression, output_field=output_field)
+
+    def _get_repr_options(self):
+        return {**super().get_repr_options(), **self.fields}
+
+    def resolve_expression(
+        self, query=None, allow_joins=True, reuse=None, summarize=False, for_save=False
+    ):
+        c = super().resolve_expression(query, allow_joins, reuse, summarize, for_save)
+        # Resolve expressions in the JSON update values.
+        c.fields = {
+            key: (
+                value.resolve_expression(query, allow_joins, reuse, summarize, for_save)
+                # If it's an expression, resolve it and use it as-is
+                if hasattr(value, "resolve_expression")
+                # Otherwise, use Value to serialize the data to string
+                else Value(value, output_field=c.output_field)
+            )
+            for key, value in self.fields.items()
+        }
+        return c
+
+    def join(self, args):
+        key, value = next(iter(self.fields.items()))
+        key_paths = key.split(LOOKUP_SEP)
+        key_paths_join = compile_json_path(key_paths)
+
+        template = f"{args[0]}, SET q'\uffff{key_paths_join}\uffff' = {args[-1]}"
+
+        if isinstance(value, Value) and isinstance(value.output_field, JSONField):
+            # Use the FORMAT JSON clause in JSON_TRANSFORM so the value is automatically
+            # treated as JSON.
+            return f"{template} FORMAT JSON"
+        return template
+
+    def as_sql(
+        self,
+        compiler,
+        connection,
+        function=None,
+        template=None,
+        arg_joiner=None,
+        **extra_context,
+    ):
+        if not connection.features.supports_partial_json_update:
+            raise NotSupportedError(
+                "JSONSet() is not supported on this database backend."
+            )
+        copy = self.copy()
+        new_source_expressions = copy.get_source_expressions()
+
+        for key, value in self.fields.items():
+            key_paths = key.split(LOOKUP_SEP)
+            key_paths_join = compile_json_path(key_paths)
+            new_source_expressions.append(Value(key_paths_join))
+
+            # If it's a Value, assume it to be a JSON-formatted string.
+            # Use Cast to ensure the string is treated as JSON on the database.
+            if isinstance(value, Value) and isinstance(value.output_field, JSONField):
+                value = Cast(value, output_field=self.output_field)
+
+            new_source_expressions.append(value)
+
+        copy.set_source_expressions(new_source_expressions)
+
+        return super(JSONSet, copy).as_sql(
+            compiler,
+            connection,
+            function="JSON_SET",
+            **extra_context,
+        )
+
+    def as_postgresql(self, compiler, connection, **extra_context):
+        copy = self.copy()
+        (key, value), *rest = self.fields.items()
+
+        # JSONB_SET does not support arbitrary number of arguments,
+        # so convert multiple updates into recursive calls.
+        if rest:
+            copy.fields = {key: value}
+            return JSONSet(copy, **dict(rest)).as_postgresql(
+                compiler, connection, **extra_context
+            )
+
+        new_source_expressions = copy.get_source_expressions()
+
+        key_paths = key.split(LOOKUP_SEP)
+        new_source_expressions.append(Value(key_paths))
+
+        if hasattr(value, "resolve_expression") and not isinstance(
+            value.output_field, JSONField
+        ):
+            # Database expressions may return any type. We cannot use Cast() here
+            # because ::jsonb only works with JSON-formatted strings, not with
+            # other types like integers. The TO_JSONB function is available for
+            # this purpose, i.e. to convert any SQL type to JSONB.
+            value = ToJSONB(value, output_field=self.output_field)
+        elif isinstance(value, Value) and value.value is None:
+            # Avoid None from being interpreted as SQL NULL.
+            value = Value(None, output_field=self.output_field)
+
+        new_source_expressions.append(value)
+        copy.set_source_expressions(new_source_expressions)
+        return super(JSONSet, copy).as_sql(
+            compiler, connection, function="JSONB_SET", **extra_context
+        )
+
+    def as_oracle(self, compiler, connection, **extra_context):
+        if not connection.features.supports_partial_json_update:
+            raise NotSupportedError(
+                "JSONSet() is not supported on this database backend."
+            )
+        copy = self.copy()
+        (key, value), *rest = self.fields.items()
+
+        # JSON_TRANSFORM does not support arbitrary number of arguments,
+        # so convert multiple updates into recursive calls.
+        if rest:
+            copy.fields = {key: value}
+            return JSONSet(copy, **dict(rest)).as_oracle(
+                compiler, connection, **extra_context
+            )
+
+        new_source_expressions = copy.get_source_expressions()
+        new_source_expressions.append(value)
+        copy.set_source_expressions(new_source_expressions)
+
+        return super(JSONSet, copy).as_sql(
+            compiler,
+            connection,
+            function="JSON_TRANSFORM",
+            arg_joiner=self,
+            **extra_context,
+        )
+
+
+class JSONRemove(Func):
+    def __init__(self, expression, *paths, **kwargs):
+        if not paths:
+            raise TypeError("JSONRemove requires at least one path to remove")
+        self.paths = paths
+        super().__init__(expression, **kwargs)
+
+    def _get_repr_options(self):
+        return {**super().get_repr_options(), **self.fields}
+
+    def join(self, args):
+        path = self.paths[0]
+        key_paths = path.split(LOOKUP_SEP)
+        key_paths_join = compile_json_path(key_paths)
+
+        return f"{args[0]}, REMOVE q'\uffff{key_paths_join}\uffff'"
+
+    def as_sql(
+        self,
+        compiler,
+        connection,
+        function=None,
+        template=None,
+        arg_joiner=None,
+        **extra_context,
+    ):
+        if not connection.features.supports_partial_json_update:
+            raise NotSupportedError(
+                "JSONRemove() is not supported on this database backend."
+            )
+
+        copy = self.copy()
+        new_source_expressions = copy.get_source_expressions()
+
+        for path in self.paths:
+            key_paths = path.split(LOOKUP_SEP)
+            key_paths_join = compile_json_path(key_paths)
+            new_source_expressions.append(Value(key_paths_join))
+
+        copy.set_source_expressions(new_source_expressions)
+
+        return super(JSONRemove, copy).as_sql(
+            compiler,
+            connection,
+            function="JSON_REMOVE",
+            **extra_context,
+        )
+
+    def as_postgresql(self, compiler, connection, **extra_context):
+        copy = self.copy()
+        path, *rest = self.paths
+
+        if rest:
+            copy.paths = (path,)
+            return JSONRemove(copy, *rest).as_postgresql(
+                compiler, connection, **extra_context
+            )
+
+        new_source_expressions = copy.get_source_expressions()
+        key_paths = path.split(LOOKUP_SEP)
+        new_source_expressions.append(Value(key_paths))
+        copy.set_source_expressions(new_source_expressions)
+
+        return super(JSONRemove, copy).as_sql(
+            compiler,
+            connection,
+            template="%(expressions)s",
+            arg_joiner="#- ",
+            **extra_context,
+        )
+
+    def as_oracle(self, compiler, connection, **extra_context):
+        if not connection.features.supports_partial_json_update:
+            raise NotSupportedError(
+                "JSONRemove() is not supported on this database backend."
+            )
+
+        all_items = self.paths
+        path, *rest = all_items
+
+        if rest:
+            copy = self.copy()
+            copy.paths = (path,)
+            return JSONRemove(copy, *rest).as_oracle(
+                compiler, connection, **extra_context
+            )
+
+        return super().as_sql(
+            compiler,
+            connection,
+            function="JSON_TRANSFORM",
+            arg_joiner=self,
+            **extra_context,
+        )