Module aiolirest.rest
HPE Machine Learning Inference Software (MLIS/Aioli)
HPE MLIS is Aioli – The AI On-line Inference Platform that enables easy deployment, tracking, and serving of your packaged models regardless of your preferred AI framework.
The version of the OpenAPI document: 1.0.0 Contact: community@determined-ai Generated by OpenAPI Generator (https://openapi-generator.tech)
Do not edit the class manually.
Expand source code
# coding: utf-8
"""
HPE Machine Learning Inference Software (MLIS/Aioli)
HPE MLIS is *Aioli* -- The AI On-line Inference Platform that enables easy deployment, tracking, and serving of your packaged models regardless of your preferred AI framework.
The version of the OpenAPI document: 1.0.0
Contact: community@determined-ai
Generated by OpenAPI Generator (https://openapi-generator.tech)
Do not edit the class manually.
""" # noqa: E501
import io
import json
import re
import ssl
import urllib3
from aiolirest.exceptions import ApiException, ApiValueError
RESTResponseType = urllib3.HTTPResponse
class RESTResponse(io.IOBase):
def __init__(self, resp) -> None:
self.response = resp
self.status = resp.status
self.reason = resp.reason
self.data = None
def read(self):
if self.data is None:
self.data = self.response.data
return self.data
def getheaders(self):
"""Returns a dictionary of the response headers."""
return self.response.headers
def getheader(self, name, default=None):
"""Returns a given response header."""
return self.response.headers.get(name, default)
class RESTClientObject:
def __init__(self, configuration) -> None:
# urllib3.PoolManager will pass all kw parameters to connectionpool
# https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/poolmanager.py#L75 # noqa: E501
# https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/connectionpool.py#L680 # noqa: E501
# Custom SSL certificates and client certificates: http://urllib3.readthedocs.io/en/latest/advanced-usage.html # noqa: E501
# cert_reqs
if configuration.verify_ssl:
cert_reqs = ssl.CERT_REQUIRED
else:
cert_reqs = ssl.CERT_NONE
addition_pool_args = {}
if configuration.assert_hostname is not None:
addition_pool_args['assert_hostname'] = (
configuration.assert_hostname
)
if configuration.retries is not None:
addition_pool_args['retries'] = configuration.retries
if configuration.tls_server_name:
addition_pool_args['server_hostname'] = configuration.tls_server_name
if configuration.socket_options is not None:
addition_pool_args['socket_options'] = configuration.socket_options
# https pool manager
if configuration.proxy:
self.pool_manager = urllib3.ProxyManager(
cert_reqs=cert_reqs,
ca_certs=configuration.ssl_ca_cert,
cert_file=configuration.cert_file,
key_file=configuration.key_file,
proxy_url=configuration.proxy,
proxy_headers=configuration.proxy_headers,
**addition_pool_args
)
else:
self.pool_manager = urllib3.PoolManager(
cert_reqs=cert_reqs,
ca_certs=configuration.ssl_ca_cert,
cert_file=configuration.cert_file,
key_file=configuration.key_file,
**addition_pool_args
)
def request(
self,
method,
url,
headers=None,
body=None,
post_params=None,
_request_timeout=None
):
"""Perform requests.
:param method: http request method
:param url: http request url
:param headers: http request headers
:param body: request json body, for `application/json`
:param post_params: request post parameters,
`application/x-www-form-urlencoded`
and `multipart/form-data`
:param _request_timeout: timeout setting for this request. If one
number provided, it will be total request
timeout. It can also be a pair (tuple) of
(connection, read) timeouts.
"""
method = method.upper()
assert method in [
'GET',
'HEAD',
'DELETE',
'POST',
'PUT',
'PATCH',
'OPTIONS'
]
if post_params and body:
raise ApiValueError(
"body parameter cannot be used with post_params parameter."
)
post_params = post_params or {}
headers = headers or {}
timeout = None
if _request_timeout:
if isinstance(_request_timeout, (int, float)):
timeout = urllib3.Timeout(total=_request_timeout)
elif (
isinstance(_request_timeout, tuple)
and len(_request_timeout) == 2
):
timeout = urllib3.Timeout(
connect=_request_timeout[0],
read=_request_timeout[1]
)
try:
# For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE`
if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']:
# no content type provided or payload is json
content_type = headers.get('Content-Type')
if (
not content_type
or re.search('json', content_type, re.IGNORECASE)
):
request_body = None
if body is not None:
request_body = json.dumps(body)
r = self.pool_manager.request(
method,
url,
body=request_body,
timeout=timeout,
headers=headers,
preload_content=False
)
elif content_type == 'application/x-www-form-urlencoded':
r = self.pool_manager.request(
method,
url,
fields=post_params,
encode_multipart=False,
timeout=timeout,
headers=headers,
preload_content=False
)
elif content_type == 'multipart/form-data':
# must del headers['Content-Type'], or the correct
# Content-Type which generated by urllib3 will be
# overwritten.
del headers['Content-Type']
r = self.pool_manager.request(
method,
url,
fields=post_params,
encode_multipart=True,
timeout=timeout,
headers=headers,
preload_content=False
)
# Pass a `string` parameter directly in the body to support
# other content types than Json when `body` argument is
# provided in serialized form
elif isinstance(body, str) or isinstance(body, bytes):
request_body = body
r = self.pool_manager.request(
method,
url,
body=request_body,
timeout=timeout,
headers=headers,
preload_content=False
)
else:
# Cannot generate the request from given parameters
msg = """Cannot prepare a request message for provided
arguments. Please check that your arguments match
declared content type."""
raise ApiException(status=0, reason=msg)
# For `GET`, `HEAD`
else:
r = self.pool_manager.request(
method,
url,
fields={},
timeout=timeout,
headers=headers,
preload_content=False
)
except urllib3.exceptions.SSLError as e:
msg = "\n".join([type(e).__name__, str(e)])
raise ApiException(status=0, reason=msg)
return RESTResponse(r)Classes
class RESTClientObject (configuration)-
Expand source code
class RESTClientObject: def __init__(self, configuration) -> None: # urllib3.PoolManager will pass all kw parameters to connectionpool # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/poolmanager.py#L75 # noqa: E501 # https://github.com/shazow/urllib3/blob/f9409436f83aeb79fbaf090181cd81b784f1b8ce/urllib3/connectionpool.py#L680 # noqa: E501 # Custom SSL certificates and client certificates: http://urllib3.readthedocs.io/en/latest/advanced-usage.html # noqa: E501 # cert_reqs if configuration.verify_ssl: cert_reqs = ssl.CERT_REQUIRED else: cert_reqs = ssl.CERT_NONE addition_pool_args = {} if configuration.assert_hostname is not None: addition_pool_args['assert_hostname'] = ( configuration.assert_hostname ) if configuration.retries is not None: addition_pool_args['retries'] = configuration.retries if configuration.tls_server_name: addition_pool_args['server_hostname'] = configuration.tls_server_name if configuration.socket_options is not None: addition_pool_args['socket_options'] = configuration.socket_options # https pool manager if configuration.proxy: self.pool_manager = urllib3.ProxyManager( cert_reqs=cert_reqs, ca_certs=configuration.ssl_ca_cert, cert_file=configuration.cert_file, key_file=configuration.key_file, proxy_url=configuration.proxy, proxy_headers=configuration.proxy_headers, **addition_pool_args ) else: self.pool_manager = urllib3.PoolManager( cert_reqs=cert_reqs, ca_certs=configuration.ssl_ca_cert, cert_file=configuration.cert_file, key_file=configuration.key_file, **addition_pool_args ) def request( self, method, url, headers=None, body=None, post_params=None, _request_timeout=None ): """Perform requests. :param method: http request method :param url: http request url :param headers: http request headers :param body: request json body, for `application/json` :param post_params: request post parameters, `application/x-www-form-urlencoded` and `multipart/form-data` :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. """ method = method.upper() assert method in [ 'GET', 'HEAD', 'DELETE', 'POST', 'PUT', 'PATCH', 'OPTIONS' ] if post_params and body: raise ApiValueError( "body parameter cannot be used with post_params parameter." ) post_params = post_params or {} headers = headers or {} timeout = None if _request_timeout: if isinstance(_request_timeout, (int, float)): timeout = urllib3.Timeout(total=_request_timeout) elif ( isinstance(_request_timeout, tuple) and len(_request_timeout) == 2 ): timeout = urllib3.Timeout( connect=_request_timeout[0], read=_request_timeout[1] ) try: # For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE` if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']: # no content type provided or payload is json content_type = headers.get('Content-Type') if ( not content_type or re.search('json', content_type, re.IGNORECASE) ): request_body = None if body is not None: request_body = json.dumps(body) r = self.pool_manager.request( method, url, body=request_body, timeout=timeout, headers=headers, preload_content=False ) elif content_type == 'application/x-www-form-urlencoded': r = self.pool_manager.request( method, url, fields=post_params, encode_multipart=False, timeout=timeout, headers=headers, preload_content=False ) elif content_type == 'multipart/form-data': # must del headers['Content-Type'], or the correct # Content-Type which generated by urllib3 will be # overwritten. del headers['Content-Type'] r = self.pool_manager.request( method, url, fields=post_params, encode_multipart=True, timeout=timeout, headers=headers, preload_content=False ) # Pass a `string` parameter directly in the body to support # other content types than Json when `body` argument is # provided in serialized form elif isinstance(body, str) or isinstance(body, bytes): request_body = body r = self.pool_manager.request( method, url, body=request_body, timeout=timeout, headers=headers, preload_content=False ) else: # Cannot generate the request from given parameters msg = """Cannot prepare a request message for provided arguments. Please check that your arguments match declared content type.""" raise ApiException(status=0, reason=msg) # For `GET`, `HEAD` else: r = self.pool_manager.request( method, url, fields={}, timeout=timeout, headers=headers, preload_content=False ) except urllib3.exceptions.SSLError as e: msg = "\n".join([type(e).__name__, str(e)]) raise ApiException(status=0, reason=msg) return RESTResponse(r)Methods
def request(self, method, url, headers=None, body=None, post_params=None)-
Perform requests.
:param method: http request method :param url: http request url :param headers: http request headers :param body: request json body, for
application/json:param post_params: request post parameters,application/x-www-form-urlencodedandmultipart/form-data:param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts.Expand source code
def request( self, method, url, headers=None, body=None, post_params=None, _request_timeout=None ): """Perform requests. :param method: http request method :param url: http request url :param headers: http request headers :param body: request json body, for `application/json` :param post_params: request post parameters, `application/x-www-form-urlencoded` and `multipart/form-data` :param _request_timeout: timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts. """ method = method.upper() assert method in [ 'GET', 'HEAD', 'DELETE', 'POST', 'PUT', 'PATCH', 'OPTIONS' ] if post_params and body: raise ApiValueError( "body parameter cannot be used with post_params parameter." ) post_params = post_params or {} headers = headers or {} timeout = None if _request_timeout: if isinstance(_request_timeout, (int, float)): timeout = urllib3.Timeout(total=_request_timeout) elif ( isinstance(_request_timeout, tuple) and len(_request_timeout) == 2 ): timeout = urllib3.Timeout( connect=_request_timeout[0], read=_request_timeout[1] ) try: # For `POST`, `PUT`, `PATCH`, `OPTIONS`, `DELETE` if method in ['POST', 'PUT', 'PATCH', 'OPTIONS', 'DELETE']: # no content type provided or payload is json content_type = headers.get('Content-Type') if ( not content_type or re.search('json', content_type, re.IGNORECASE) ): request_body = None if body is not None: request_body = json.dumps(body) r = self.pool_manager.request( method, url, body=request_body, timeout=timeout, headers=headers, preload_content=False ) elif content_type == 'application/x-www-form-urlencoded': r = self.pool_manager.request( method, url, fields=post_params, encode_multipart=False, timeout=timeout, headers=headers, preload_content=False ) elif content_type == 'multipart/form-data': # must del headers['Content-Type'], or the correct # Content-Type which generated by urllib3 will be # overwritten. del headers['Content-Type'] r = self.pool_manager.request( method, url, fields=post_params, encode_multipart=True, timeout=timeout, headers=headers, preload_content=False ) # Pass a `string` parameter directly in the body to support # other content types than Json when `body` argument is # provided in serialized form elif isinstance(body, str) or isinstance(body, bytes): request_body = body r = self.pool_manager.request( method, url, body=request_body, timeout=timeout, headers=headers, preload_content=False ) else: # Cannot generate the request from given parameters msg = """Cannot prepare a request message for provided arguments. Please check that your arguments match declared content type.""" raise ApiException(status=0, reason=msg) # For `GET`, `HEAD` else: r = self.pool_manager.request( method, url, fields={}, timeout=timeout, headers=headers, preload_content=False ) except urllib3.exceptions.SSLError as e: msg = "\n".join([type(e).__name__, str(e)]) raise ApiException(status=0, reason=msg) return RESTResponse(r)
class RESTResponse (resp)-
The abstract base class for all I/O classes.
This class provides dummy implementations for many methods that derived classes can override selectively; the default implementations represent a file that cannot be read, written or seeked.
Even though IOBase does not declare read, readinto, or write because their signatures will vary, implementations and clients should consider those methods part of the interface. Also, implementations may raise UnsupportedOperation when operations they do not support are called.
The basic type used for binary data read from or written to a file is bytes. Other bytes-like objects are accepted as method arguments too. In some cases (such as readinto), a writable object is required. Text I/O classes work with str data.
Note that calling any method (except additional calls to close(), which are ignored) on a closed stream should raise a ValueError.
IOBase (and its subclasses) support the iterator protocol, meaning that an IOBase object can be iterated over yielding the lines in a stream.
IOBase also supports the :keyword:
withstatement. In this example, fp is closed after the suite of the with statement is complete:with open('spam.txt', 'r') as fp: fp.write('Spam and eggs!')
Expand source code
class RESTResponse(io.IOBase): def __init__(self, resp) -> None: self.response = resp self.status = resp.status self.reason = resp.reason self.data = None def read(self): if self.data is None: self.data = self.response.data return self.data def getheaders(self): """Returns a dictionary of the response headers.""" return self.response.headers def getheader(self, name, default=None): """Returns a given response header.""" return self.response.headers.get(name, default)Ancestors
- io.IOBase
- _io._IOBase
Methods
def getheader(self, name, default=None)-
Returns a given response header.
Expand source code
def getheader(self, name, default=None): """Returns a given response header.""" return self.response.headers.get(name, default) def getheaders(self)-
Returns a dictionary of the response headers.
Expand source code
def getheaders(self): """Returns a dictionary of the response headers.""" return self.response.headers def read(self)-
Expand source code
def read(self): if self.data is None: self.data = self.response.data return self.data