Add requestBody component and reference resolution

src/apispec/core.py

@@ -53,12 +53,14 @@ def __init__(
         self.examples: dict[str, dict] = {}
         self.links: dict[str, dict] = {}
         self.security_schemes: dict[str, dict] = {}
+        self.request_bodies: dict[str, dict] = {}
         self.schemas_lazy: dict[str, dict] = {}
         self.responses_lazy: dict[str, dict] = {}
         self.parameters_lazy: dict[str, dict] = {}
         self.headers_lazy: dict[str, dict] = {}
         self.examples_lazy: dict[str, dict] = {}
         self.links_lazy: dict[str, dict] = {}
+        self.request_bodies_lazy: dict[str, dict] = {}
 
         self._subsections = {
             "schema": self.schemas,
@@ -68,6 +70,7 @@ def __init__(
             "example": self.examples,
             "link": self.links,
             "security_scheme": self.security_schemes,
+            "request_body": self.request_bodies,
         }
         self._subsections_lazy = {
             "schema": self.schemas_lazy,
@@ -76,13 +79,14 @@ def __init__(
             "header": self.headers_lazy,
             "example": self.examples_lazy,
             "link": self.links_lazy,
+            "request_body": self.request_bodies_lazy,
         }
 
     def to_dict(self) -> dict[str, dict]:
         return {
             COMPONENT_SUBSECTIONS[self.openapi_version.major][k]: v
             for k, v in self._subsections.items()
-            if v != {}
+            if v != {} and k in COMPONENT_SUBSECTIONS[self.openapi_version.major]
         }
 
     def _register_component(
@@ -280,6 +284,38 @@ def header(
         self._register_component("header", component_id, ret, lazy=lazy)
         return self
 
+    def request_body(
+        self,
+        component_id: str,
+        component: dict | None = None,
+        *,
+        lazy: bool = False,
+        **kwargs: typing.Any,
+    ) -> Components:
+        """Add a request body which can be referenced.
+
+        :param str component_id: identifier by which request body may be referenced
+        :param dict component: request body fields
+        :param bool lazy: register component only when referenced in the spec
+        :param kwargs: plugin-specific arguments
+
+        https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#requestBodyObject
+        """
+        if component_id in self.request_bodies:
+            raise DuplicateComponentNameError(
+                f'Another request body with name "{component_id}" is already registered.'
+            )
+        ret = deepcopy(component) or {}
+        # Execute all helpers from plugins
+        for plugin in self._plugins:
+            try:
+                ret.update(plugin.request_body_helper(ret, **kwargs) or {})
+            except PluginMethodNotImplementedError:
+                continue
+        self._resolve_refs_in_request_body(ret)
+        self._register_component("request_body", component_id, ret, lazy=lazy)
+        return self
+
     def example(
         self, component_id: str, component: dict, *, lazy: bool = False
     ) -> Components:
@@ -371,7 +407,7 @@ def _resolve_refs_in_parameter_or_header(self, parameter_or_header) -> None:
 
     def _resolve_refs_in_request_body(self, request_body) -> None:
         # requestBody is OpenAPI v3+
-        for media_type in request_body["content"].values():
+        for media_type in request_body.get("content", {}).values():
             self._resolve_schema(media_type)
             self._resolve_examples(media_type)
 
@@ -402,6 +438,9 @@ def _resolve_refs_in_operation(self, operation) -> None:
                     for path in callback.values():
                         self.resolve_refs_in_path(path)
         if "requestBody" in operation:
+            operation["requestBody"] = self.get_ref(
+                "request_body", operation["requestBody"]
+            )
             self._resolve_refs_in_request_body(operation["requestBody"])
         if "responses" in operation:
             responses = {}

src/apispec/ext/marshmallow/__init__.py

@@ -222,6 +222,17 @@ def header_helper(self, header: dict, **kwargs: typing.Any):
         self.resolver.resolve_schema(header)
         return header
 
+    def request_body_helper(self, request_body: dict, **kwargs: typing.Any):
+        """RequestBody component helper that allows using a marshmallow
+        :class:`Schema <marshmallow.Schema>` in request_body definition.
+
+        :param dict request_body: request_body fields. May contain a marshmallow
+            Schema class or instance.
+        """
+        assert self.resolver  # needed for mypy
+        self.resolver.resolve_schema(request_body)
+        return request_body
+
     def operation_helper(
         self,
         path: str | None = None,

src/apispec/plugin.py

@@ -52,6 +52,16 @@ def header_helper(self, header: dict, **kwargs: typing.Any) -> dict | None:
         """
         raise PluginMethodNotImplementedError
 
+    def request_body_helper(
+        self, request_body: dict, **kwargs: typing.Any
+    ) -> dict | None:
+        """May return requestBody component description as a dict.
+
+        :param dict request_body: RequestBody fields
+        :param kwargs: All additional keywords arguments sent to `APISpec.request_body()`
+        """
+        raise PluginMethodNotImplementedError
+
     def path_helper(
         self,
         path: str | None = None,

src/apispec/utils.py

@@ -21,6 +21,7 @@
         "example": "examples",
         "link": "links",
         "security_scheme": "securitySchemes",
+        "request_body": "requestBodies",
     },
 }
 

tests/test_core.py

@@ -19,6 +19,7 @@
     get_links,
     get_parameters,
     get_paths,
+    get_request_bodies,
     get_responses,
     get_schemas,
     get_security_schemes,
@@ -318,6 +319,34 @@ def test_header_duplicate_name(self, spec):
         ):
             spec.components.header("test_header", {"schema": {"type": "integer"}})
 
+    # Referenced request bodies are only supported in OAS 3.x
+    @pytest.mark.parametrize("spec", ("3.0.0",), indirect=True)
+    def test_request_body(self, spec):
+        body = {"content": {"application/json": {"schema": {"type": "string"}}}}
+        spec.components.request_body("test_body", body.copy())
+        bodies = get_request_bodies(spec)
+        assert bodies["test_body"] == body
+
+    # Referenced request bodies are only supported in OAS 3.x
+    @pytest.mark.parametrize("spec", ("3.0.0",), indirect=True)
+    def test_request_body_is_chainable(self, spec):
+        body = {"content": {"application/json": {"schema": {"type": "string"}}}}
+        spec.components.request_body("body1", body).request_body("body2", body)
+        bodies = get_request_bodies(spec)
+        assert "body1" in bodies
+        assert "body2" in bodies
+
+    # Referenced request bodies are only supported in OAS 3.x
+    @pytest.mark.parametrize("spec", ("3.0.0",), indirect=True)
+    def test_request_body_duplicate_name(self, spec):
+        body = {"content": {"application/json": {"schema": {"type": "string"}}}}
+        spec.components.request_body("test_body", body)
+        with pytest.raises(
+            DuplicateComponentNameError,
+            match='Another request body with name "test_body" is already registered.',
+        ):
+            spec.components.request_body("test_body", body)
+
     # Referenced examples are only supported in OAS 3.x
     @pytest.mark.parametrize("spec", ("3.0.0",), indirect=True)
     def test_example(self, spec):
@@ -538,6 +567,40 @@ def test_components_resolve_refs_in_header_schema(self, spec):
         spec.components.header("header", header)
         self.assert_schema_refs(spec, get_headers(spec)["header"]["schema"])
 
+    # "requestBodies" components section only exists in OAS 3
+    @pytest.mark.parametrize("spec", ("3.0.0",), indirect=True)
+    def test_components_resolve_request_body_schema(self, spec):
+        body = {"content": {"application/json": {"schema": "PetSchema"}}}
+        spec.components.request_body("body", body)
+        res = get_request_bodies(spec)["body"]
+        assert res["content"]["application/json"]["schema"] == build_ref(
+            spec, "schema", "PetSchema"
+        )
+
+    # "requestBodies" components section only exists in OAS 3
+    @pytest.mark.parametrize("spec", ("3.0.0",), indirect=True)
+    def test_components_resolve_request_body_examples(self, spec):
+        body = {
+            "content": {"application/json": {"examples": {"example_1": "Example_1"}}}
+        }
+        spec.components.request_body("body", body)
+        res = get_request_bodies(spec)["body"]
+        assert res["content"]["application/json"]["examples"]["example_1"] == build_ref(
+            spec, "example", "Example_1"
+        )
+
+    # "requestBodies" components section only exists in OAS 3
+    @pytest.mark.parametrize("spec", ("3.0.0",), indirect=True)
+    def test_components_resolve_refs_in_request_body_schema(self, spec):
+        body = {
+            "content": {"application/json": {"schema": copy.deepcopy(self.REFS_SCHEMA)}}
+        }
+        spec.components.request_body("body", body)
+        self.assert_schema_refs(
+            spec,
+            get_request_bodies(spec)["body"]["content"]["application/json"]["schema"],
+        )
+
     def test_schema_lazy(self, spec):
         spec.components.schema("Pet_1", {"properties": self.properties}, lazy=False)
         spec.components.schema("Pet_2", {"properties": self.properties}, lazy=True)
@@ -642,6 +705,22 @@ def test_links_lazy(self, spec):
         assert "GetUser" in links
         assert links["GetUser"] == link
 
+    # Referenced request bodies are only supported in OAS 3.x
+    @pytest.mark.parametrize("spec", ("3.0.0",), indirect=True)
+    def test_request_body_lazy(self, spec):
+        body = {"content": {"application/json": {"schema": {"type": "string"}}}}
+        spec.components.request_body("Body_1", body, lazy=False)
+        spec.components.request_body("Body_2", body, lazy=True)
+        bodies = get_request_bodies(spec)
+        assert "Body_1" in bodies
+        assert "Body_2" not in bodies
+        spec.path(
+            "/path",
+            operations={"post": {"requestBody": "Body_2"}},
+        )
+        bodies = get_request_bodies(spec)
+        assert "Body_2" in bodies
+
 
 class TestPath(RefsSchemaTestMixin):
     paths = {
@@ -1004,6 +1083,17 @@ def test_path_resolve_request_body(self, spec):
             "application/json"
         ]["schema"] == build_ref(spec, "schema", "PetSchema")
 
+    # requestBody only exists in OAS 3
+    @pytest.mark.parametrize("spec", ("3.0.0",), indirect=True)
+    def test_path_resolve_request_body_ref(self, spec):
+        spec.path(
+            "/pet/{petId}",
+            operations={"post": {"requestBody": "PetRequestBody"}},
+        )
+        assert get_paths(spec)["/pet/{petId}"]["post"]["requestBody"] == build_ref(
+            spec, "request_body", "PetRequestBody"
+        )
+
     # "headers" components section only exists in OAS 3
     @pytest.mark.parametrize("spec", ("3.0.0",), indirect=True)
     def test_path_resolve_response_header(self, spec):
@@ -1139,6 +1229,11 @@ def header_helper(self, header, **kwargs):
                 if not return_none:
                     return {"description": "some header"}
 
+            def request_body_helper(self, request_body, **kwargs):
+                request_body.pop("dummy", None)
+                if not return_none:
+                    return {"description": "some request body"}
+
             def path_helper(self, path, operations, parameters, **kwargs):
                 if not return_none:
                     if path == "/path_1":
@@ -1234,6 +1329,27 @@ def test_plugin_header_helper_is_used(self, openapi_version, return_none):
         # Check original header is not modified
         assert header == {"dummy": "dummy"}
 
+    @pytest.mark.parametrize("openapi_version", ("3.0.0",))
+    @pytest.mark.parametrize("return_none", (True, False))
+    def test_plugin_request_body_helper_is_used(self, openapi_version, return_none):
+        spec = APISpec(
+            title="Swagger Petstore",
+            version="1.0.0",
+            openapi_version=openapi_version,
+            plugins=(self.make_test_plugin(return_none),),
+        )
+        body = {"dummy": "dummy"}
+        spec.components.request_body("Pet", body)
+        bodies = get_request_bodies(spec)
+        if return_none:
+            assert bodies["Pet"] == {}
+        else:
+            assert bodies["Pet"] == {
+                "description": "some request body",
+            }
+        # Check original request_body is not modified
+        assert body == {"dummy": "dummy"}
+
     @pytest.mark.parametrize("openapi_version", ("2.0", "3.0.0"))
     @pytest.mark.parametrize("return_none", (True, False))
     def test_plugin_path_helper_is_used(self, openapi_version, return_none):

tests/test_ext_marshmallow.py

@@ -34,6 +34,7 @@
     get_headers,
     get_parameters,
     get_paths,
+    get_request_bodies,
     get_responses,
     get_schemas,
 )
@@ -316,6 +317,20 @@ def test_can_use_schema_in_header(self, spec, schema):
         assert resolved_schema["properties"]["password"]["type"] == "string"
         assert resolved_schema["properties"]["id"]["type"] == "integer"
 
+    @pytest.mark.parametrize("spec", ("3.0.0",), indirect=True)
+    @pytest.mark.parametrize("schema", [PetSchema, PetSchema()])
+    def test_can_use_schema_in_request_body(self, spec, schema):
+        body = {"content": {"application/json": {"schema": schema}}}
+        spec.components.request_body("Pet", body)
+        rb = get_request_bodies(spec)["Pet"]
+        reference = rb["content"]["application/json"]["schema"]
+        assert reference == build_ref(spec, "schema", "Pet")
+
+        resolved_schema = spec.components.schemas["Pet"]
+        assert resolved_schema["properties"]["name"]["type"] == "string"
+        assert resolved_schema["properties"]["password"]["type"] == "string"
+        assert resolved_schema["properties"]["id"]["type"] == "integer"
+
 
 class TestCustomField:
     def test_can_use_custom_field_decorator(self, spec_fixture):

tests/utils.py

@@ -40,6 +40,10 @@ def get_links(spec):
     return spec.to_dict()["components"].get("links", {})
 
 
+def get_request_bodies(spec):
+    return spec.to_dict()["components"].get("requestBodies", {})
+
+
 def get_security_schemes(spec):
     if spec.openapi_version.major < 3:
         return spec.to_dict().get("securityDefinitions", {})