Route.params reads form and multipart bodies; body_content_type (#23)

ju/oas.py

@@ -127,12 +127,50 @@ def properties_of_schema(schema: dict) -> dict:
         return path_get(schema, "properties")
 
 
+#: Non-JSON request body media types `Route.params` reads, in order of preference.
+BODY_MEDIA_TYPES = ("application/x-www-form-urlencoded", "multipart/form-data")
+
 #: The `OAS_IN_KEY` of a property that comes from the request body.
 BODY_IN = "body"
 #: The `OAS_IN_KEY` of a parameter that does not say where it goes (invalid OpenAPI).
 PARAM_IN = "param"
 
 
+def _media_type(media_type: str) -> str:
+    """A media type without its parameters, lower-cased.
+
+    >>> _media_type('Application/JSON; charset=utf-8')
+    'application/json'
+    """
+    return media_type.split(";")[0].strip().lower()
+
+
+def _preferred_media_type(media_types):
+    """The media type to read a request body from: plain JSON, any other JSON
+    (``application/vnd.api+json``), then form-urlencoded, then multipart.
+
+    >>> _preferred_media_type(['application/json-patch+json', 'application/json'])
+    'application/json'
+    >>> _preferred_media_type(['text/plain', 'Multipart/Form-Data'])
+    'Multipart/Form-Data'
+    >>> _preferred_media_type(['application/x-ndjson']) is None
+    True
+    """
+    by_type = {}
+    for media_type in media_types:
+        by_type.setdefault(_media_type(media_type), media_type)
+    if "application/json" in by_type:
+        return by_type["application/json"]
+    for normalized, media_type in by_type.items():
+        subtype = normalized.partition("/")[2]
+        if subtype == "json" or subtype.endswith("+json"):
+            return media_type
+    for normalized in BODY_MEDIA_TYPES:
+        if normalized in by_type:
+            return by_type[normalized]
+    return None
+
+
 def _param_id(param: dict) -> tuple:
     """What identifies an OpenAPI parameter: its name *and* where it goes."""
     return param.get("name"), param.get("in")
@@ -238,6 +276,31 @@ def input_specs(self):
             "requestBody": self.method_data.get("requestBody", {}),
         }
 
+    @cached_property
+    def body_content_type(self):
+        """The media type of the request body `params` reads, or None for no body.
+
+        JSON is preferred; then form-urlencoded, then multipart.
+
+        >>> from yaml import safe_load
+        >>> spec = safe_load('''
+        ... paths:
+        ...   /f:
+        ...     post:
+        ...       requestBody:
+        ...         content:
+        ...           multipart/form-data: {schema: {properties: {a: {type: string}}}}
+        ... ''')
+        >>> route = Route('post', '/f', spec)
+        >>> route.body_content_type, list(route.params['properties'])
+        ('multipart/form-data', ['a'])
+        """
+        content = (self.method_data.get("requestBody") or {}).get("content") or {}
+        with_schema = [
+            m for m, v in content.items() if isinstance(v, dict) and v.get("schema")
+        ]
+        return _preferred_media_type(with_schema) or _preferred_media_type(content)
+
     @cached_property
     def output_specs(self):
         return self.method_data.get("responses", {})
@@ -251,8 +314,8 @@ def params(self):
         schema = {"type": "object", "properties": {}, "required": []}
 
         parameters = self.method_data.get("parameters", [])
-        request_body = self.method_data.get("requestBody", {})
-        content = path_get(request_body, "content.application/json")
+        request_body = self.method_data.get("requestBody") or {}
+        content = (request_body.get("content") or {}).get(self.body_content_type) or {}
         body_schema = content.get("schema")
         body_properties = properties_of_schema(body_schema) if body_schema else {}
 

ju/tests/oas_test.py

@@ -270,3 +270,63 @@ def test_a_distinct_key_never_takes_a_name_the_spec_uses():
         "other": ("other", "query"),
         "other_cookie": ("other", "cookie"),
     }
+
+
+def test_form_and_multipart_bodies_are_read_and_json_is_preferred():
+    from yaml import safe_load
+
+    spec = safe_load(
+        """
+paths:
+  /a: {post: {requestBody: {content: {application/x-www-form-urlencoded:
+         {schema: {required: [f], properties: {f: {type: string}}}}}}}}
+  /b: {post: {requestBody: {content: {multipart/form-data: {schema: {properties: {m: {}}}},
+         application/json: {schema: {properties: {j: {}}}}}}}}
+  /c: {post: {responses: {}}}
+"""
+    )
+    routes = Routes(spec)
+    a, b, c = (routes["post", p] for p in ("/a", "/b", "/c"))
+    assert a.body_content_type == "application/x-www-form-urlencoded"
+    assert list(a.params["properties"]) == ["f"] and a.params["required"] == ["f"]
+    assert b.body_content_type == "application/json"
+    assert list(b.params["properties"]) == ["j"]
+    assert c.body_content_type is None and c.params["properties"] == {}
+
+
+@pytest.mark.parametrize(
+    "content, expected",
+    [
+        (  # plain JSON beats other JSON types, whatever the order
+            ["application/json-patch+json", "application/json", "text/json"],
+            "application/json",
+        ),
+        (["application/vnd.api+json; charset=utf-8", "text/plain"], "application/vnd.api+json; charset=utf-8"),
+        (["application/x-ndjson", "application/notjson"], None),
+        (["application/X-WWW-Form-Urlencoded; charset=utf-8"], "application/X-WWW-Form-Urlencoded; charset=utf-8"),
+        (["Multipart/Form-Data"], "Multipart/Form-Data"),
+        (["text/plain", "*/*"], None),
+    ],
+)
+def test_body_content_type_preference(content, expected):
+    from ju.oas import _preferred_media_type
+
+    assert _preferred_media_type(content) == expected
+
+
+def test_a_media_type_with_a_schema_beats_one_without():
+    from yaml import safe_load
+
+    spec = safe_load(
+        """
+paths:
+  /a: {post: {requestBody: {content: {application/json: {},
+         application/x-www-form-urlencoded: {schema: {properties: {f: {}}}}}}}}
+  /b: {post: {requestBody: {content: }}}
+  /c: {post: {requestBody: }}
+"""
+    )
+    routes = Routes(spec)
+    assert list(routes["post", "/a"].params["properties"]) == ["f"]
+    assert routes["post", "/b"].params["properties"] == {}  # content: null
+    assert routes["post", "/c"].body_content_type is None