Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 24 additions & 5 deletions modules/openapi-generator/src/main/resources/python/rest.mustache
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Comment thread
MarcelKonrad marked this conversation as resolved.
"""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."""
Expand Down Expand Up @@ -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:
Expand All @@ -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,
Expand All @@ -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.
Expand Down Expand Up @@ -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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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:
Expand All @@ -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,
Expand All @@ -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.
Expand Down Expand Up @@ -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,
Expand Down
29 changes: 24 additions & 5 deletions samples/client/echo_api/python/openapi_client/rest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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:
Expand All @@ -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,
Expand All @@ -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.
Expand Down Expand Up @@ -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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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:
Expand All @@ -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,
Expand All @@ -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.
Expand Down Expand Up @@ -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,
Expand Down
29 changes: 24 additions & 5 deletions samples/openapi3/client/petstore/python/petstore_api/rest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
"""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."""
Expand Down Expand Up @@ -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:
Expand All @@ -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,
Expand All @@ -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.
Expand Down Expand Up @@ -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,
Expand Down
Loading