Repository: getmoto/moto
Pull request #9016 (2025-06-27 03:22:30)

Core: Serialization Updates
Various refactorings and added features to allow finer-grained control and customization of the serialization pipeline.


Patch:
diff --git a/moto/core/responses.py b/moto/core/responses.py
index be15dab19..e53bd5bae 100644
--- a/moto/core/responses.py
+++ b/moto/core/responses.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import datetime
 import functools
 import json
@@ -5,6 +7,7 @@ import logging
 import os
 import re
 from collections import OrderedDict, defaultdict
+from dataclasses import dataclass
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -24,13 +27,14 @@ from xml.dom.minidom import parseString as parseXML
 import boto3
 import requests
 import xmltodict
+from botocore.model import OperationModel, ServiceModel
 from jinja2 import DictLoader, Environment, Template
 from werkzeug.exceptions import HTTPException
 
 from moto import settings
 from moto.core.common_types import TYPE_IF_NONE, TYPE_RESPONSE
 from moto.core.exceptions import DryRunClientError, ServiceException
-from moto.core.serialize import SERIALIZERS, XFormedAttributePicker
+from moto.core.serialize import SERIALIZERS, ResponseSerializer, XFormedAttributePicker
 from moto.core.utils import (
     camelcase_to_underscores,
     get_service_model,
@@ -279,6 +283,14 @@ class ActionAuthenticatorMixin(object):
         return decorator
 
 
+@dataclass
+class ActionContext:
+    service_model: ServiceModel
+    operation_model: OperationModel
+    serializer_class: type[ResponseSerializer]
+    response_class: type[BaseResponse]
+
+
 class ActionResult:
     """Wrapper class for serializable results returned from `responses.py` methods."""
 
@@ -289,6 +301,34 @@ class ActionResult:
     def result(self) -> object:
         return self._result
 
+    def execute_result(self, context: ActionContext) -> TYPE_RESPONSE:
+        """
+        Execute the result in the context of the given service and operation model.
+        This is a placeholder for any logic that might be needed to process the result
+        based on the service and operation context.
+        """
+        serializer_cls = context.serializer_class
+        response_transformers = getattr(
+            context.response_class, "RESPONSE_KEY_PATH_TO_TRANSFORMER", None
+        )
+        value_picker = XFormedAttributePicker(
+            response_transformers=response_transformers
+        )
+        serializer = serializer_cls(
+            operation_model=context.operation_model,
+            pretty_print=settings.PRETTIFY_RESPONSES,
+            value_picker=value_picker,
+        )
+        serialized = serializer.serialize(self.result)
+        return serialized["status_code"], serialized["headers"], serialized["body"]  # type: ignore[return-value]
+
+
+class EmptyResult(ActionResult):
+    """A special ActionResult that represents an empty result."""
+
+    def __init__(self) -> None:
+        super().__init__(None)
+
 
 class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
     default_region = "us-east-1"
@@ -595,13 +635,10 @@ class BaseResponse(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
         service_model = get_service_model(self.service_name)
         operation_model = service_model.operation_model(self._get_action())
         serializer_cls = SERIALIZERS[service_model.protocol]
-        serializer = serializer_cls(
-            operation_model=operation_model,
-            pretty_print=settings.PRETTIFY_RESPONSES,
-            value_picker=XFormedAttributePicker(),
+        context = ActionContext(
+            service_model, operation_model, serializer_cls, self.__class__
         )
-        serialized = serializer.serialize(action_result.result)
-        return serialized["status_code"], serialized["headers"], serialized["body"]  # type: ignore[return-value]
+        return action_result.execute_result(context)
 
     def call_action(self) -> TYPE_RESPONSE:
         headers = self.response_headers
diff --git a/moto/core/serialize.py b/moto/core/serialize.py
index a94c11e7e..59c9cc3f4 100644
--- a/moto/core/serialize.py
+++ b/moto/core/serialize.py
@@ -58,7 +58,16 @@ import calendar
 import json
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Generator, Mapping, MutableMapping, Optional, TypedDict, Union
+from typing import (
+    Any,
+    Callable,
+    Generator,
+    Mapping,
+    MutableMapping,
+    Optional,
+    TypedDict,
+    Union,
+)
 
 import xmltodict
 from botocore import xform_name
@@ -68,6 +77,7 @@ from botocore.model import (
     MapShape,
     NoShapeFoundError,
     OperationModel,
+    ServiceModel,
     Shape,
     StructureShape,
 )
@@ -255,6 +265,7 @@ class ResponseSerializer(ShapeHelpersMixin):
             value_picker = DefaultAttributePicker()
         self._value_picker = value_picker
         self._timestamp_serializer = TimestampSerializer(self.DEFAULT_TIMESTAMP_FORMAT)
+        self.path = []
 
     def _create_default_response(self) -> ResponseDict:
         response_dict: ResponseDict = {
@@ -334,7 +345,14 @@ class ResponseSerializer(ShapeHelpersMixin):
         return base64.b64encode(value).strip().decode(self.DEFAULT_ENCODING)
 
     def get_value(self, value: Any, key: str, shape: Shape) -> Any:
-        context = AttributePickerContext(obj=value, key=key, shape=shape)
+        context = AttributePickerContext(
+            obj=value,
+            key=key,
+            shape=shape,
+            operation_model=self.operation_model,
+            service_model=self.operation_model.service_model,
+            key_path=".".join(self.path),
+        )
         return self._value_picker(context)
 
     @staticmethod
@@ -377,7 +395,12 @@ class ResponseSerializer(ShapeHelpersMixin):
         method = getattr(
             self, "_serialize_type_%s" % shape.type_name, self._default_serialize
         )
+        if not key:
+            self.path.append(shape.name)
+        else:
+            self.path.append(key)
         method(serialized, value, shape, key)
+        self.path.pop()
 
     @staticmethod
     def _default_serialize(
@@ -555,9 +578,10 @@ class BaseJSONSerializer(ResponseSerializer):
             # a __current__ key on a wrapper dict to serialize each
             # list item before appending it to the serialized list.
             assert isinstance(shape.member, Shape)  # mypy hint
-            self._serialize(wrapper, list_item, shape.member, "__current__")
-            if "__current__" in wrapper:
-                list_obj.append(wrapper["__current__"])
+            item_key = shape.member.name
+            self._serialize(wrapper, list_item, shape.member, item_key)
+            if item_key in wrapper:
+                list_obj.append(wrapper[item_key])
             else:
                 list_obj.append(list_item)
 
@@ -677,9 +701,10 @@ class BaseXMLSerializer(ResponseSerializer):
             items_name = self.get_serialized_name(shape.member, "member")
             serialized[key] = {items_name: list_obj}
         for list_item in value:
+            item_key = shape.member.name
             wrapper = {}
-            self._serialize(wrapper, list_item, shape.member, "__current__")
-            list_obj.append(wrapper["__current__"])
+            self._serialize(wrapper, list_item, shape.member, item_key)
+            list_obj.append(wrapper[item_key])
         if not list_obj:
             serialized[key] = ""
 
@@ -908,6 +933,9 @@ class AttributePickerContext:
     obj: Any
     key: str
     shape: Shape
+    operation_model: OperationModel | None = None
+    service_model: ServiceModel | None = None
+    key_path: str = ""
 
 
 class DefaultAttributePicker:
@@ -915,15 +943,20 @@ class DefaultAttributePicker:
         return get_value(context.obj, context.key, None)
 
 
-class AliasedAttributePicker(DefaultAttributePicker):
+class AttributePicker(DefaultAttributePicker):
     """Uses alias providers to find the value of an attribute in a Python object"""
 
     def __init__(
-        self, alias_providers: list[type[AttributeAliasProvider]] | None = None
+        self,
+        alias_providers: list[type[AttributeAliasProvider]] | None = None,
+        response_transformers: dict[str, Callable[[Any], Any]] | None = None,
     ) -> None:
         self.alias_providers = (
             alias_providers if alias_providers is not None else DEFAULT_ALIAS_PROVIDERS
         )
+        self.response_transformers = (
+            response_transformers if response_transformers is not None else {}
+        )
 
     def __call__(self, context: AttributePickerContext) -> Any:
         obj = context.obj
@@ -933,6 +966,10 @@ class AliasedAttributePicker(DefaultAttributePicker):
                 break
         else:
             value = None
+        key_path = f"{context.key_path}.{context.key}"
+        if key_path in self.response_transformers:
+            transform = self.response_transformers[key_path]
+            value = transform(value)
         return value
 
     def get_possible_keys(self, context: AttributePickerContext) -> Generator[str]:
@@ -944,7 +981,7 @@ class AliasedAttributePicker(DefaultAttributePicker):
                 yield alias
 
 
-class XFormedAttributePicker(AliasedAttributePicker):
+class XFormedAttributePicker(AttributePicker):
     """Can be injected into a ResponseSerializer to aid in plucking AWS model
     attributes specified in `camelCase` or `PascalCase` from Python objects
     with standard `snake_case` attribute names.
@@ -1089,3 +1126,29 @@ DEFAULT_ALIAS_PROVIDERS = [
     ClassPrefixAlias,
     ModelAlias,
 ]
+
+
+# Response transformers can be used to modify the value of a key in the response.
+def never_return(_: Any) -> None:
+    """
+    A utility function that is used to ensure that certain attributes are never returned
+    in the response. This is useful for attributes that should not be exposed or are not
+    relevant in the context of the response.
+    """
+    return None
+
+
+def return_if_not_empty(value: Any) -> Any:
+    """
+    A utility function that returns the value if it is not empty (i.e., not None, "", {}, or []),
+    otherwise returns None. This is useful for attributes that should only be included in the
+    response if they have a meaningful value.
+    """
+    return value if value not in [None, "", {}, []] else None
+
+
+def url_encode(value: Any) -> Any:
+    """A utility function that url encodes a value before inclusion in a response."""
+    from urllib.parse import quote
+
+    return quote(value) if isinstance(value, str) else value


Test patch:
diff --git a/tests/test_core/test_response_serialization.py b/tests/test_core/test_response_serialization.py
new file mode 100644
index 000000000..7d68bc9c7
--- /dev/null
+++ b/tests/test_core/test_response_serialization.py
@@ -0,0 +1,60 @@
+from botocore.model import ServiceModel
+
+from moto.core.responses import ActionContext, ActionResult, BaseResponse, EmptyResult
+from moto.core.serialize import QuerySerializer, never_return
+
+TEST_MODEL = {
+    "metadata": {"protocol": "query", "apiVersion": "2014-01-01"},
+    "documentation": "",
+    "operations": {
+        "TestOperation": {
+            "name": "TestOperation",
+            "http": {
+                "method": "POST",
+                "requestUri": "/",
+            },
+            "output": {"shape": "OutputShape"},
+        }
+    },
+    "shapes": {
+        "OutputShape": {
+            "type": "structure",
+            "members": {
+                "string": {"shape": "StringType"},
+                "TransformerTest": {"shape": "StringType"},
+            },
+        },
+        "StringType": {
+            "type": "string",
+        },
+    },
+}
+
+
+def test_response_result_execution() -> None:
+    """Exercise various response result execution paths."""
+
+    class TestResponseClass(BaseResponse):
+        RESPONSE_KEY_PATH_TO_TRANSFORMER = {"OutputShape.TransformerTest": never_return}
+
+    class TestResponseObject:
+        string = "test-string"
+        TransformerTest = "some data"
+
+    service_model = ServiceModel(TEST_MODEL)
+    operation_model = service_model.operation_model("TestOperation")
+    serializer_class = QuerySerializer
+    response_class = TestResponseClass
+
+    context = ActionContext(
+        service_model, operation_model, serializer_class, response_class
+    )
+
+    result = ActionResult(TestResponseObject())
+    _, __, body = result.execute_result(context)
+    assert "test-string" in body
+    assert "TransformerTest" not in body
+
+    result = EmptyResult()
+    _, __, body = result.execute_result(context)
+    assert "<OutputShapeResult/>" in body
diff --git a/tests/test_core/test_serialize.py b/tests/test_core/test_serialize.py
index 96fda31cb..a6e86a0f9 100644
--- a/tests/test_core/test_serialize.py
+++ b/tests/test_core/test_serialize.py
@@ -21,6 +21,9 @@ from moto.core.serialize import (
     AttributePickerContext,
     QuerySerializer,
     XFormedAttributePicker,
+    never_return,
+    return_if_not_empty,
+    url_encode,
 )
 
 
@@ -217,3 +220,25 @@ class TestXFormedAttributePicker:
         ctx = AttributePickerContext(DBInstance(), "DBInstanceClass", None)
         value = self.picker(ctx)
         assert value == "t2.medium"
+
+
+class TestResponseAttributeTransformers:
+    def test_never_return(self):
+        assert never_return(None) is None
+        assert never_return(True) is None
+        assert never_return(False) is None
+        assert never_return("Test") is None
+
+    def test_return_if_not_empty(self):
+        assert return_if_not_empty([]) is None
+        assert return_if_not_empty(None) is None
+        assert return_if_not_empty("Test") == "Test"
+        assert return_if_not_empty({}) is None
+        assert return_if_not_empty("") is None
+        assert return_if_not_empty({"key": "value"}) == {"key": "value"}
+        assert return_if_not_empty([0, 1, 2]) == [0, 1, 2]
+
+    def test_url_encode(self):
+        assert url_encode(None) is None
+        assert url_encode("Test") == "Test"
+        assert url_encode("Test String") == "Test%20String"