diff --git a/modules/openapi-generator/src/main/resources/python/rest.mustache b/modules/openapi-generator/src/main/resources/python/rest.mustache index 5d852b28d92b..f814e2334575 100644 --- a/modules/openapi-generator/src/main/resources/python/rest.mustache +++ b/modules/openapi-generator/src/main/resources/python/rest.mustache @@ -27,6 +27,21 @@ def is_socks_proxy_url(url): else: return split_section[0].lower() in SUPPORTED_SOCKS_PROXIES +def contenttype_matches(contenttype, maintype, subtype): + """Matches the given contenttype against the given type and subtype + + :param contenttype: the content type to match + :param maintype: the expected maintype + :param subtype: the expected subtype + :return: `true` when the given content type matches the given type and subtype, + regardless of the presence of mime type parameters, otherwise returns `false`. + :rtype: bool + """ + pattern = '{type}/(?:[^+;]+\\+)?{subtype}(?:[ \t]*;.*)?'.format( + type = re.escape(maintype), + subtype = re.escape(subtype), + ) + return re.fullmatch(pattern, contenttype, re.IGNORECASE) is not None def should_bypass_proxies(url: str, no_proxy: str) -> bool: """Return whether ``url`` matches the comma-separated ``no_proxy`` rules.""" @@ -212,14 +227,18 @@ class RESTClientObject: content_type = headers.get('Content-Type') is_json = ( not content_type - or re.search('json', content_type, re.IGNORECASE) + or contenttype_matches(content_type, 'application', 'json') ) # JSON is valid YAML 1.2, so structured YAML bodies can use # the existing JSON serializer: # https://yaml.org/spec/1.2.2/#13-relation-to-json is_structured_yaml = ( content_type - and re.search('yaml', content_type, re.IGNORECASE) + and ( + contenttype_matches(content_type, 'application', 'yaml') + or contenttype_matches(content_type, 'text', 'yaml') + or contenttype_matches(content_type, 'text', 'x-yaml') + ) and not isinstance(body, (str, bytes)) ) if is_json or is_structured_yaml: @@ -234,7 +253,7 @@ class RESTClientObject: headers=headers, preload_content=False ) - elif content_type == 'application/x-www-form-urlencoded': + elif contenttype_matches(content_type, 'application', 'x-www-form-urlencoded'): r = self.pool_manager.request( method, url, @@ -244,7 +263,7 @@ class RESTClientObject: headers=headers, preload_content=False ) - elif content_type == 'multipart/form-data': + elif contenttype_matches(content_type, 'multipart', 'form-data'): # must del headers['Content-Type'], or the correct # Content-Type which generated by urllib3 will be # overwritten. @@ -272,7 +291,7 @@ class RESTClientObject: headers=headers, preload_content=False ) - elif headers['Content-Type'].startswith('text/') and isinstance(body, bool): + elif content_type.startswith('text/') and isinstance(body, bool): request_body = "true" if body else "false" r = self.pool_manager.request( method, diff --git a/samples/client/echo_api/python-disallowAdditionalPropertiesIfNotPresent/openapi_client/rest.py b/samples/client/echo_api/python-disallowAdditionalPropertiesIfNotPresent/openapi_client/rest.py index 934eb3f8b794..ef3ed39f4679 100644 --- a/samples/client/echo_api/python-disallowAdditionalPropertiesIfNotPresent/openapi_client/rest.py +++ b/samples/client/echo_api/python-disallowAdditionalPropertiesIfNotPresent/openapi_client/rest.py @@ -37,6 +37,21 @@ def is_socks_proxy_url(url): else: return split_section[0].lower() in SUPPORTED_SOCKS_PROXIES +def contenttype_matches(contenttype, maintype, subtype): + """Matches the given contenttype against the given type and subtype + + :param contenttype: the content type to match + :param maintype: the expected maintype + :param subtype: the expected subtype + :return: `true` when the given content type matches the given type and subtype, + regardless of the presence of mime type parameters, otherwise returns `false`. + :rtype: bool + """ + pattern = '{type}/(?:[^+;]+\\+)?{subtype}(?:[ \t]*;.*)?'.format( + type = re.escape(maintype), + subtype = re.escape(subtype), + ) + return re.fullmatch(pattern, contenttype, re.IGNORECASE) is not None def should_bypass_proxies(url: str, no_proxy: str) -> bool: """Return whether ``url`` matches the comma-separated ``no_proxy`` rules.""" @@ -222,14 +237,18 @@ def request( content_type = headers.get('Content-Type') is_json = ( not content_type - or re.search('json', content_type, re.IGNORECASE) + or contenttype_matches(content_type, 'application', 'json') ) # JSON is valid YAML 1.2, so structured YAML bodies can use # the existing JSON serializer: # https://yaml.org/spec/1.2.2/#13-relation-to-json is_structured_yaml = ( content_type - and re.search('yaml', content_type, re.IGNORECASE) + and ( + contenttype_matches(content_type, 'application', 'yaml') + or contenttype_matches(content_type, 'text', 'yaml') + or contenttype_matches(content_type, 'text', 'x-yaml') + ) and not isinstance(body, (str, bytes)) ) if is_json or is_structured_yaml: @@ -244,7 +263,7 @@ def request( headers=headers, preload_content=False ) - elif content_type == 'application/x-www-form-urlencoded': + elif contenttype_matches(content_type, 'application', 'x-www-form-urlencoded'): r = self.pool_manager.request( method, url, @@ -254,7 +273,7 @@ def request( headers=headers, preload_content=False ) - elif content_type == 'multipart/form-data': + elif contenttype_matches(content_type, 'multipart', 'form-data'): # must del headers['Content-Type'], or the correct # Content-Type which generated by urllib3 will be # overwritten. @@ -282,7 +301,7 @@ def request( headers=headers, preload_content=False ) - elif headers['Content-Type'].startswith('text/') and isinstance(body, bool): + elif content_type.startswith('text/') and isinstance(body, bool): request_body = "true" if body else "false" r = self.pool_manager.request( method, diff --git a/samples/client/echo_api/python/openapi_client/rest.py b/samples/client/echo_api/python/openapi_client/rest.py index b5a114226b79..5bd90b34957e 100644 --- a/samples/client/echo_api/python/openapi_client/rest.py +++ b/samples/client/echo_api/python/openapi_client/rest.py @@ -37,6 +37,21 @@ def is_socks_proxy_url(url): else: return split_section[0].lower() in SUPPORTED_SOCKS_PROXIES +def contenttype_matches(contenttype, maintype, subtype): + """Matches the given contenttype against the given type and subtype + + :param contenttype: the content type to match + :param maintype: the expected maintype + :param subtype: the expected subtype + :return: `true` when the given content type matches the given type and subtype, + regardless of the presence of mime type parameters, otherwise returns `false`. + :rtype: bool + """ + pattern = '{type}/(?:[^+;]+\\+)?{subtype}(?:[ \t]*;.*)?'.format( + type = re.escape(maintype), + subtype = re.escape(subtype), + ) + return re.fullmatch(pattern, contenttype, re.IGNORECASE) is not None def should_bypass_proxies(url: str, no_proxy: str) -> bool: """Return whether ``url`` matches the comma-separated ``no_proxy`` rules.""" @@ -222,14 +237,18 @@ def request( content_type = headers.get('Content-Type') is_json = ( not content_type - or re.search('json', content_type, re.IGNORECASE) + or contenttype_matches(content_type, 'application', 'json') ) # JSON is valid YAML 1.2, so structured YAML bodies can use # the existing JSON serializer: # https://yaml.org/spec/1.2.2/#13-relation-to-json is_structured_yaml = ( content_type - and re.search('yaml', content_type, re.IGNORECASE) + and ( + contenttype_matches(content_type, 'application', 'yaml') + or contenttype_matches(content_type, 'text', 'yaml') + or contenttype_matches(content_type, 'text', 'x-yaml') + ) and not isinstance(body, (str, bytes)) ) if is_json or is_structured_yaml: @@ -244,7 +263,7 @@ def request( headers=headers, preload_content=False ) - elif content_type == 'application/x-www-form-urlencoded': + elif contenttype_matches(content_type, 'application', 'x-www-form-urlencoded'): r = self.pool_manager.request( method, url, @@ -254,7 +273,7 @@ def request( headers=headers, preload_content=False ) - elif content_type == 'multipart/form-data': + elif contenttype_matches(content_type, 'multipart', 'form-data'): # must del headers['Content-Type'], or the correct # Content-Type which generated by urllib3 will be # overwritten. @@ -282,7 +301,7 @@ def request( headers=headers, preload_content=False ) - elif headers['Content-Type'].startswith('text/') and isinstance(body, bool): + elif content_type.startswith('text/') and isinstance(body, bool): request_body = "true" if body else "false" r = self.pool_manager.request( method, diff --git a/samples/openapi3/client/petstore/python-lazyImports/petstore_api/rest.py b/samples/openapi3/client/petstore/python-lazyImports/petstore_api/rest.py index 5df0581723cc..075dca30faee 100644 --- a/samples/openapi3/client/petstore/python-lazyImports/petstore_api/rest.py +++ b/samples/openapi3/client/petstore/python-lazyImports/petstore_api/rest.py @@ -36,6 +36,21 @@ def is_socks_proxy_url(url): else: return split_section[0].lower() in SUPPORTED_SOCKS_PROXIES +def contenttype_matches(contenttype, maintype, subtype): + """Matches the given contenttype against the given type and subtype + + :param contenttype: the content type to match + :param maintype: the expected maintype + :param subtype: the expected subtype + :return: `true` when the given content type matches the given type and subtype, + regardless of the presence of mime type parameters, otherwise returns `false`. + :rtype: bool + """ + pattern = '{type}/(?:[^+;]+\\+)?{subtype}(?:[ \t]*;.*)?'.format( + type = re.escape(maintype), + subtype = re.escape(subtype), + ) + return re.fullmatch(pattern, contenttype, re.IGNORECASE) is not None def should_bypass_proxies(url: str, no_proxy: str) -> bool: """Return whether ``url`` matches the comma-separated ``no_proxy`` rules.""" @@ -221,14 +236,18 @@ def request( content_type = headers.get('Content-Type') is_json = ( not content_type - or re.search('json', content_type, re.IGNORECASE) + or contenttype_matches(content_type, 'application', 'json') ) # JSON is valid YAML 1.2, so structured YAML bodies can use # the existing JSON serializer: # https://yaml.org/spec/1.2.2/#13-relation-to-json is_structured_yaml = ( content_type - and re.search('yaml', content_type, re.IGNORECASE) + and ( + contenttype_matches(content_type, 'application', 'yaml') + or contenttype_matches(content_type, 'text', 'yaml') + or contenttype_matches(content_type, 'text', 'x-yaml') + ) and not isinstance(body, (str, bytes)) ) if is_json or is_structured_yaml: @@ -243,7 +262,7 @@ def request( headers=headers, preload_content=False ) - elif content_type == 'application/x-www-form-urlencoded': + elif contenttype_matches(content_type, 'application', 'x-www-form-urlencoded'): r = self.pool_manager.request( method, url, @@ -253,7 +272,7 @@ def request( headers=headers, preload_content=False ) - elif content_type == 'multipart/form-data': + elif contenttype_matches(content_type, 'multipart', 'form-data'): # must del headers['Content-Type'], or the correct # Content-Type which generated by urllib3 will be # overwritten. @@ -281,7 +300,7 @@ def request( headers=headers, preload_content=False ) - elif headers['Content-Type'].startswith('text/') and isinstance(body, bool): + elif content_type.startswith('text/') and isinstance(body, bool): request_body = "true" if body else "false" r = self.pool_manager.request( method, diff --git a/samples/openapi3/client/petstore/python/petstore_api/rest.py b/samples/openapi3/client/petstore/python/petstore_api/rest.py index 5df0581723cc..075dca30faee 100755 --- a/samples/openapi3/client/petstore/python/petstore_api/rest.py +++ b/samples/openapi3/client/petstore/python/petstore_api/rest.py @@ -36,6 +36,21 @@ def is_socks_proxy_url(url): else: return split_section[0].lower() in SUPPORTED_SOCKS_PROXIES +def contenttype_matches(contenttype, maintype, subtype): + """Matches the given contenttype against the given type and subtype + + :param contenttype: the content type to match + :param maintype: the expected maintype + :param subtype: the expected subtype + :return: `true` when the given content type matches the given type and subtype, + regardless of the presence of mime type parameters, otherwise returns `false`. + :rtype: bool + """ + pattern = '{type}/(?:[^+;]+\\+)?{subtype}(?:[ \t]*;.*)?'.format( + type = re.escape(maintype), + subtype = re.escape(subtype), + ) + return re.fullmatch(pattern, contenttype, re.IGNORECASE) is not None def should_bypass_proxies(url: str, no_proxy: str) -> bool: """Return whether ``url`` matches the comma-separated ``no_proxy`` rules.""" @@ -221,14 +236,18 @@ def request( content_type = headers.get('Content-Type') is_json = ( not content_type - or re.search('json', content_type, re.IGNORECASE) + or contenttype_matches(content_type, 'application', 'json') ) # JSON is valid YAML 1.2, so structured YAML bodies can use # the existing JSON serializer: # https://yaml.org/spec/1.2.2/#13-relation-to-json is_structured_yaml = ( content_type - and re.search('yaml', content_type, re.IGNORECASE) + and ( + contenttype_matches(content_type, 'application', 'yaml') + or contenttype_matches(content_type, 'text', 'yaml') + or contenttype_matches(content_type, 'text', 'x-yaml') + ) and not isinstance(body, (str, bytes)) ) if is_json or is_structured_yaml: @@ -243,7 +262,7 @@ def request( headers=headers, preload_content=False ) - elif content_type == 'application/x-www-form-urlencoded': + elif contenttype_matches(content_type, 'application', 'x-www-form-urlencoded'): r = self.pool_manager.request( method, url, @@ -253,7 +272,7 @@ def request( headers=headers, preload_content=False ) - elif content_type == 'multipart/form-data': + elif contenttype_matches(content_type, 'multipart', 'form-data'): # must del headers['Content-Type'], or the correct # Content-Type which generated by urllib3 will be # overwritten. @@ -281,7 +300,7 @@ def request( headers=headers, preload_content=False ) - elif headers['Content-Type'].startswith('text/') and isinstance(body, bool): + elif content_type.startswith('text/') and isinstance(body, bool): request_body = "true" if body else "false" r = self.pool_manager.request( method,