From 57f74a8bcb90a946ca91b3e082bc4365eb5b6029 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Tue, 13 Oct 2026 11:36:56 +0000 Subject: [PATCH 1/2] Route.params reads form and multipart request bodies; Route.body_content_type (#23) Co-Authored-By: Claude Sonnet 5.5 --- ju/oas.py | 33 ++++++++++++++++++++++++++++++++- ju/tests/oas_test.py | 22 ++++++++++++++++++++++ 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/ju/oas.py b/ju/oas.py index 8aaa873..3e3794a 100644 --- a/ju/oas.py +++ b/ju/oas.py @@ -127,6 +127,9 @@ 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). @@ -238,6 +241,34 @@ 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", {}).get("content", {}) + for media_type in content: + if media_type.split(";")[0].strip().lower().endswith("json"): + return media_type + for media_type in BODY_MEDIA_TYPES: + if media_type in content: + return media_type + return None + @cached_property def output_specs(self): return self.method_data.get("responses", {}) @@ -252,7 +283,7 @@ def params(self): parameters = self.method_data.get("parameters", []) request_body = self.method_data.get("requestBody", {}) - content = path_get(request_body, "content.application/json") + content = request_body.get("content", {}).get(self.body_content_type, {}) body_schema = content.get("schema") body_properties = properties_of_schema(body_schema) if body_schema else {} diff --git a/ju/tests/oas_test.py b/ju/tests/oas_test.py index 083fe86..db9135e 100644 --- a/ju/tests/oas_test.py +++ b/ju/tests/oas_test.py @@ -270,3 +270,25 @@ 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"] == {} From 95e22021ca3927e2f2157cd652ced9dba9885897 Mon Sep 17 00:00:00 2001 From: Thor Whalen <1906276+thorwhalen@users.noreply.github.com> Date: Tue, 13 Oct 2026 11:39:28 +0000 Subject: [PATCH 2/2] Review fixes: media-type preference (plain JSON first, +json only, parameters/case ignored, schema-bearing first), null content (#23) Co-Authored-By: Claude Sonnet 5.5 --- ju/oas.py | 52 +++++++++++++++++++++++++++++++++++--------- ju/tests/oas_test.py | 38 ++++++++++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 10 deletions(-) diff --git a/ju/oas.py b/ju/oas.py index 3e3794a..42dea1b 100644 --- a/ju/oas.py +++ b/ju/oas.py @@ -136,6 +136,41 @@ def properties_of_schema(schema: dict) -> dict: 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") @@ -260,14 +295,11 @@ def body_content_type(self): >>> route.body_content_type, list(route.params['properties']) ('multipart/form-data', ['a']) """ - content = self.method_data.get("requestBody", {}).get("content", {}) - for media_type in content: - if media_type.split(";")[0].strip().lower().endswith("json"): - return media_type - for media_type in BODY_MEDIA_TYPES: - if media_type in content: - return media_type - return None + 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): @@ -282,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 = request_body.get("content", {}).get(self.body_content_type, {}) + 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 {} diff --git a/ju/tests/oas_test.py b/ju/tests/oas_test.py index db9135e..23a87b9 100644 --- a/ju/tests/oas_test.py +++ b/ju/tests/oas_test.py @@ -292,3 +292,41 @@ def test_form_and_multipart_bodies_are_read_and_json_is_preferred(): 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