# -*- encoding: utf-8 -*-
"""
Class for communication with an H2O server.
`H2OConnection` is the main class of this module, and it handles the connection itself:
hc = H2OConnection.open() : open a new connection
hc.request(endpoint, [data|json|filename]) : make a REST API request to the server
hc.info() : return information about the current connection
hc.close() : close the connection
hc.session_id : current session id
:copyright: (c) 2016 H2O.ai
:license: Apache License Version 2.0 (see LICENSE for details)
"""
from h2o.utils.compatibility import * # NOQA
import atexit
from collections import defaultdict
import os
import re
import sys
import tempfile
import time
import types
from warnings import warn
from weakref import ref
import requests
from requests.auth import AuthBase
from h2o.backend import H2OCluster, H2OLocalServer
from h2o.display import print2
from h2o.exceptions import H2OConnectionError, H2OServerError, H2OResponseError, H2OValueError
from h2o.model.metrics import make_metrics
from h2o.schemas import H2OMetadataV3, H2OErrorV3, H2OModelBuilderErrorV3, define_classes_from_schema
from h2o.two_dim_table import H2OTwoDimTable
from h2o.utils.metaclass import CallableString, backwards_compatibility, h2o_meta
from h2o.utils.shared_utils import stringify_list, stringify_dict, as_resource
from h2o.utils.typechecks import (assert_is_type, assert_matches, assert_satisfies, is_type, numeric)
__all__ = ("H2OConnection", "H2OConnectionConf", )
if tuple(int(x) for x in requests.__version__.split('.')) < (2, 10):
print("[WARNING] H2O requires requests module of version 2.10 or newer. You have version %s.\n"
"You can upgrade to the newest version of the module running from the command line\n"
" $ pip%s install --upgrade requests" % (requests.__version__, sys.version_info[0]))
_session_hooks_ = defaultdict(list)
def apply_session_hooks(event='open'):
handler = 'on_{}'.format(event)
if handler not in _session_hooks_:
return
for h in _session_hooks_[handler]:
try:
h()
except Exception:
pass
def register_session_hook(event='open', hook=None):
assert hook is not None
handler = 'on_{}'.format(event)
_session_hooks_[handler].append(hook)
class H2OConnectionConf(object):
"""
Configuration of connection to H2O.
The main goal of this class is to specify location of running
H2O instance and properties of connection. The location of running instance
is given by schema, ip, port, context_path parameters forming connection URL in
the following way: ``schema://ip:port/context_path``
"""
def __init__(self, config=None):
self._ip = None
self._port = None
self._https = None
self._context_path = ''
self._verify_ssl_certificates = True
self._cacert = None
self._proxy = None
self._auth = None
self._cookies = None
self._verbose = True
# Fill from config if it is specified
if config is not None:
self._fill_from_config(config)
"""List of allowed property names exposed by this class"""
allowed_properties = ["ip", "port", "https", "context_path", "verify_ssl_certificates", "cacert",
"proxy", "auth", "cookies", "verbose"]
def _fill_from_config(self, config):
"""
Fill this instance from given dictionary.
The method only uses keys which corresponds to properties
this class, throws exception on unknown property name.
:param conf: dictionary of parameters
:return: a new instance of this class filled with values from given dictionary
:raises H2OValueError: if input config contains unknown property name.
"""
for k,v in config.items():
if k in H2OConnectionConf.allowed_properties:
setattr(self, k, v)
else:
raise H2OValueError(message="Unsupported name of property: %s!" % k, var_name="config")
@property
def ip(self):
return self._ip
@ip.setter
def ip(self, value):
assert_is_type(value, str)
self._ip = value
@property
def port(self):
return self._port
@port.setter
def port(self, value):
assert_is_type(value, int)
self._port = value
@property
def https(self):
return self._https
@https.setter
def https(self, value):
assert_is_type(value, bool)
self._https = value
@property
def context_path(self):
return self._context_path
@context_path.setter
def context_path(self, value):
assert_is_type(value, str)
self._context_path = value
@property
def verify_ssl_certificates(self):
return self._verify_ssl_certificates
@verify_ssl_certificates.setter
def verify_ssl_certificates(self, value):
assert_is_type(value, bool)
self._verify_ssl_certificates = value
@property
def cacert(self):
return self._cacert
@cacert.setter
def cacert(self, value):
assert_is_type(value, None, str)
self._cacert = value
@property
def proxy(self):
return self._proxy
@proxy.setter
def proxy(self, value):
assert_is_type(value, str, None)
self._proxy = value
@property
def auth(self):
return self._auth
@auth.setter
def auth(self, value):
assert_is_type(value, AuthBase, (str, str), None)
self._auth = value
@property
def cookies(self):
return self._cookies
@cookies.setter
def cookies(self, value):
assert_is_type(value, [str], None)
self._cookies = value
@property
def verbose(self):
return self._verbose
@verbose.setter
def verbose(self, value):
assert_is_type(value, bool)
self._verbose = value
@property
def url(self):
if self.https:
schema = "https"
else:
schema = "http"
curl = "{}://{}:{}/{}".format(schema, self.ip, self.port, self.context_path)
return curl
[docs]@backwards_compatibility(
class_attrs=dict(
__ENCODING__="utf-8",
__ENCODING_ERROR__="replace",
default=lambda: _deprecated_default(),
jar_paths=lambda: list(getattr(H2OLocalServer, "_jar_paths")()),
rest_version=lambda: 3,
https=lambda: __H2OCONN__.base_url.split(":")[0] == "https",
ip=lambda: __H2OCONN__.base_url.split(":")[1][2:],
port=lambda: __H2OCONN__.base_url.split(":")[2],
username=lambda: _deprecated_username(),
password=lambda: _deprecated_password(),
insecure=lambda: not getattr(__H2OCONN__, "_verify_ssl_cert"),
current_connection=lambda: __H2OCONN__,
check_conn=lambda: _deprecated_check_conn(),
make_url=lambda url_suffix, _rest_version=3: __H2OCONN__.make_url(url_suffix, _rest_version),
get=lambda url_suffix, **kwargs: _deprecated_get(__H2OCONN__, url_suffix, **kwargs),
post=lambda url_suffix, file_upload_info=None, **kwa: _deprecated_post(__H2OCONN__, url_suffix, file_upload_info=file_upload_info, **kwa),
delete=lambda url_suffix, **kwargs: _deprecated_delete(__H2OCONN__, url_suffix, **kwargs),
get_json=lambda url_suffix, **kwargs: _deprecated_get(__H2OCONN__, url_suffix, **kwargs),
post_json=lambda url_suffix, file_upload_info=None, **kwa: _deprecated_post(__H2OCONN__, url_suffix, file_upload_info=file_upload_info, **kwa),
rest_ctr=lambda: __H2OCONN__.requests_count,
),
instance_attrs=dict(
cluster_is_up=lambda self: self.cluster.is_running(),
info=lambda self, refresh=True: self.cluster,
shutdown_server=lambda self, prompt=True: self.cluster.shutdown(prompt),
make_url=lambda self, url_suffix, _rest_version=3: "/".join([self._base_url, str(_rest_version), url_suffix]),
get=lambda *args, **kwargs: _deprecated_get(*args, **kwargs),
post=lambda *args, **kwargs: _deprecated_post(*args, **kwargs),
delete=lambda *args, **kwargs: _deprecated_delete(*args, **kwargs),
get_json=lambda *args, **kwargs: _deprecated_get(*args, **kwargs),
post_json=lambda *args, **kwargs: _deprecated_post(*args, **kwargs),
)
)
class H2OConnection(h2o_meta()):
"""
Connection handle to an H2O cluster.
Typically, you don't need to access this class directly. Instead, use :func:`h2o.connect` to
establish a connection, and :func:`h2o.api` to make requests to the backend H2O server. However, if your
use-case is not typical, then read on.
Instances of this class may only be created through the static method :meth:`open`::
hc = H2OConnection.open(...)
Once opened, the connection remains active until the script exits (or until you explicitly :meth:`close` it).
If the script exits with an exception, then the connection will fail to close, and the backend server will.
keep all the temporary frames and the open session.
Alternatively, you can use this class as a context manager, which will ensure that the connection gets closed
at the end of the ``with ...`` block even if an exception occurs::
with H2OConnection.open() as hc:
hc.info().pprint()
Once the connection is established, you can send REST API requests to the server using :meth:`request`.
"""
"""
Defines pattern matching URL in the following form ``schema://ip:port/context_path``.
"""
url_pattern = r"^(https?)://((?:[\w-]+\.)*[\w-]+):(\d+)/?((/[\w-]+)+)?$"
[docs] @staticmethod
def open(server=None, url=None, ip=None, port=None, name=None, https=None, auth=None,
verify_ssl_certificates=True, cacert=None,
proxy=None, cookies=None, verbose=True, msgs=None, strict_version_check=True):
r"""
Establish connection to an existing H2O server.
The connection is not kept alive, so what this method actually does is attempt to connect to the
specified server, and check that the server is healthy and responds to REST API requests. If the H2O server
cannot be reached, an :class:`H2OConnectionError` will be raised. On a success, this method returns a new
:class:`H2OConnection` object, and it is the only "official" way to create instances of this class.
There are 3 ways to specify the target to connect to (these settings are mutually exclusive):
* pass a ``server`` option,
* pass the full ``url`` for the connection,
* provide a triple of parameters ``ip``, ``port``, ``https``.
:param H2OLocalServer server: connect to the specified local server instance. There is a slight difference
between connecting to a local server by specifying its ip and address, and connecting through
an H2OLocalServer instance: if the server becomes unresponsive, then having access to its process handle
will allow us to query the server status through OS, and potentially provide snapshot of the server's
error log in the exception information.
:param url: full url of the server to connect to.
:param ip: target server's IP address or hostname (default "localhost").
:param port: H2O server's port (default 54321).
:param name: H2O cluster name.
:param https: if True then connect using https instead of http (default False).
:param verify_ssl_certificates: if False then SSL certificate checking will be disabled (default True). This
setting should rarely be disabled, as it makes your connection vulnerable to man-in-the-middle attacks. When
used, it will generate a warning from the requests library. Has no effect when ``https`` is False.
:param cacert: Path to a CA bundle file or a directory with certificates of trusted CAs (optional).
:param auth: authentication token for connecting to the remote server. This can be either a
(username, password) tuple, or an authenticator (AuthBase) object. Please refer to the documentation in
the ``requests.auth`` module.
:param proxy: url address of a proxy server. If you do not specify the proxy, then the requests module
will attempt to use a proxy specified in the environment (in HTTP_PROXY / HTTPS_PROXY variables). We
check for the presence of these variables and issue a warning if they are found. In order to suppress
that warning and use proxy from the environment, pass ``proxy="(default)"``.
:param cookies: Cookie (or list of) to add to requests.
:param verbose: if True, then connection progress info will be printed to the stdout.
:param strict_version_check: If True, an error will be raised if the client and server versions don't match.
:param msgs: custom messages to display during connection. This is a tuple (initial message, success message,
failure message).
:returns: A new :class:`H2OConnection` instance.
:raises H2OConnectionError: if the server cannot be reached.
:raises H2OServerError: if the server is in an unhealthy state (although this might be a recoverable error, the
client itself should decide whether it wants to retry or not).
"""
if server is not None:
assert_is_type(server, H2OLocalServer)
assert_is_type(ip, None, "`ip` should be None when `server` parameter is supplied")
assert_is_type(url, None, "`url` should be None when `server` parameter is supplied")
assert_is_type(name, None, "`name` should be None when `server` parameter is supplied")
if not server.is_running():
raise H2OConnectionError("Unable to connect to server because it is not running")
ip = server.ip
port = server.port
scheme = server.scheme
context_path = ''
elif url is not None:
assert_is_type(url, str)
assert_is_type(ip, None, "`ip` should be None when `url` parameter is supplied")
assert_is_type(name, str, None)
# We don't allow any Unicode characters in the URL. Maybe some day we will...
match = assert_matches(url, H2OConnection.url_pattern)
scheme = match.group(1)
ip = match.group(2)
port = int(match.group(3))
context_path = '' if match.group(4) is None else "%s" % (match.group(4))
else:
if ip is None: ip = str("localhost")
if port is None: port = 54321
if https is None: https = False
if is_type(port, str) and port.isdigit(): port = int(port)
assert_is_type(ip, str)
assert_is_type(port, int)
assert_is_type(name, str, None)
assert_is_type(https, bool)
assert_matches(ip, r"(?:[\w-]+\.)*[\w-]+")
assert_satisfies(port, 1 <= port <= 65535)
scheme = "https" if https else "http"
context_path = ''
if verify_ssl_certificates is None: verify_ssl_certificates = True
assert_is_type(verify_ssl_certificates, bool)
assert_is_type(cacert, str, None)
assert_is_type(proxy, str, None)
assert_is_type(auth, AuthBase, (str, str), None)
assert_is_type(cookies, str, [str], None)
assert_is_type(msgs, None, (str, str, str))
conn = H2OConnection()
conn._verbose = bool(verbose)
conn._local_server = server
conn._base_url = "%s://%s:%d%s" % (scheme, ip, port, context_path)
conn._name = server.name if server else name
conn._verify_ssl_cert = bool(verify_ssl_certificates)
conn._cacert = cacert
conn._auth = auth
conn._cookies = cookies
conn._proxies = None
if proxy and proxy != "(default)":
conn._proxies = {scheme: proxy}
elif not proxy:
# Give user a warning if there are any "*_proxy" variables in the environment.
# GH issue: https://github.com/h2oai/h2o-3/issues/15409
# To suppress the warning pass proxy = "(default)".
for name in os.environ:
if name.lower() == scheme + "_proxy":
warn("Proxy is defined in the environment: %s. "
"This may interfere with your H2O Connection." % name)
if "localhost" in conn.ip() or "127.0.0.1" in conn.ip():
# Empty list will cause requests library to respect the default behavior.
# Thus a non-existing proxy is inserted.
conn._proxies = {
"http": None,
"https": None,
}
try:
retries = 20 if server else 5
conn._stage = 1
conn._timeout = 3.0
conn._cluster = conn._test_connection(retries, messages=msgs)
# If a server is unable to respond within 1s, it should be considered a bug. However we disable this
# setting for now, for no good reason other than to ignore all those bugs :(
conn._timeout = None
# create a weakref to prevent the atexit callback from keeping hard ref
# to the connection even after manual close.
conn_ref = ref(conn)
apply_session_hooks('open')
def exit_close():
con = conn_ref()
if con and con.connected:
print("Closing connection %s at exit" % con.session_id)
con.close()
atexit.register(exit_close)
except Exception:
# Reset _session_id so that we know the connection was not initialized properly.
conn._stage = 0
raise
conn._cluster.check_version(strict=strict_version_check)
return conn
[docs] def request(self, endpoint, data=None, json=None, filename=None, save_to=None):
"""
Perform a REST API request to the backend H2O server.
:param endpoint: (str) The endpoint's URL, for example "GET /4/schemas/KeyV4"
:param data: data payload for POST (and sometimes GET) requests. This should be a dictionary of simple
key/value pairs (values can also be arrays), which will be sent over in x-www-form-encoded format.
:param json: also data payload, but it will be sent as a JSON body. Cannot be used together with `data`.
:param filename: file to upload to the server. Cannot be used with `data` or `json`.
:param save_to: if provided, will write the response to that file (additionally, the response will be
streamed, so large files can be downloaded seamlessly). This parameter can be either a file name,
or a folder name. If the folder doesn't exist, it will be created automatically.
:returns: an H2OResponse object representing the server's response (unless ``save_to`` parameter is
provided, in which case the output file's name will be returned).
:raises H2OConnectionError: if the H2O server cannot be reached (or connection is not initialized).
:raises H2OServerError: if there was a server error (http 500), or server returned malformed JSON.
:raises H2OResponseError: if the server returned an H2OErrorV3 response (e.g. if the parameters were invalid).
"""
if self._stage == 0: raise H2OConnectionError("Connection not initialized; run .connect() first.")
if self._stage == -1: raise H2OConnectionError("Connection was closed, and can no longer be used.")
# Prepare URL
assert_is_type(endpoint, str)
match = assert_matches(str(endpoint), r"^(GET|POST|PUT|DELETE|PATCH|HEAD|TRACE) (/.*)$")
method = match.group(1)
urltail = match.group(2)
url = self._base_url + urltail
# Prepare data
if filename is not None:
assert_is_type(filename, str)
assert_is_type(json, None, "Argument `json` should be None when `filename` is used.")
assert_is_type(data, None, "Argument `data` should be None when `filename` is used.")
assert_satisfies(method, method == "POST",
"File uploads can only be done via POST method, got %s" % method)
elif data is not None:
assert_is_type(data, dict)
assert_is_type(json, None, "Argument `json` should be None when `data` is used.")
elif json is not None:
assert_is_type(json, dict)
request_data = self._prepare_data_payload(data) if filename is None else self._prepare_file_payload(filename)
params = None
if (method == "GET" or method == "DELETE") and data:
params = request_data
request_data = None
stream = False
if save_to is not None:
assert_is_type(save_to, str, types.FunctionType)
stream = True
if self._cookies is not None and isinstance(self._cookies, list):
self._cookies = ";".join(self._cookies)
# Make the request
with as_resource(request_data) as rd:
start_time = time.time()
try:
self._log_start_transaction(endpoint, rd, json, filename, params)
args = self._request_args()
resp = requests.request(method=method, url=url, data=rd, json=json, params=params,
stream=stream, **args)
if isinstance(save_to, types.FunctionType):
save_to = save_to(resp)
self._log_end_transaction(start_time, resp)
return self._process_response(resp, save_to)
except (requests.exceptions.ConnectionError, requests.exceptions.HTTPError) as e:
if self._local_server and not self._local_server.is_running():
self._log_end_exception("Local server has died.")
raise H2OConnectionError("Local server has died unexpectedly. RIP.")
else:
self._log_end_exception(e)
raise H2OConnectionError("Unexpected HTTP error: %s" % e)
except requests.exceptions.Timeout as e:
self._log_end_exception(e)
elapsed_time = time.time() - start_time
raise H2OConnectionError("Timeout after %.3fs" % elapsed_time)
except H2OResponseError as e:
err = e.args[0]
if isinstance(err, H2OErrorV3):
err.endpoint = endpoint
err.payload = (rd, json, filename, params)
raise
def _request_args(self):
headers = {"User-Agent": "H2O Python client/" + sys.version.replace("\n", ""),
"X-Cluster": self._cluster_id,
"Cookie": self._cookies}
verify = self._cacert if self._verify_ssl_cert and self._cacert else self._verify_ssl_cert
return {
'headers': headers,
'timeout': self._timeout,
'auth': self._auth,
'verify': verify,
'proxies': self._proxies
}
@staticmethod
def save_to_detect(resp):
disposition = resp.headers['Content-Disposition']
return disposition.split("filename=")[1].strip()
[docs] def close(self):
"""
Close an existing connection; once closed it cannot be used again.
Strictly speaking, it is not necessary to close all connections that you opened -- we have several mechanisms
in place that will do so automatically (``__del__()``, ``__exit__()``, and ``atexit()`` handlers), however there is also
no good reason to make this method private.
"""
if self._session_id:
try:
# If the server gone bad, we don't want to wait forever...
if self._timeout is None: self._timeout = 1
self.request("DELETE /4/sessions/%s" % self._session_id)
self._print("H2O session %s closed." % self._session_id)
except Exception as e:
self._print("H2O session %s was not closed properly." % self._session_id)
self._log_end_exception(e)
self._session_id = None
self._stage = -1
apply_session_hooks('close')
@property
def connected(self):
return self._stage > 0
@property
def session_id(self):
"""
Return the session id of the current connection.
The session id is issued (through an API request) the first time it is requested, but no sooner. This is
because generating a session id puts it into the DKV on the server, which effectively locks the cluster. Once
issued, the session id will stay the same until the connection is closed.
"""
if self._session_id is None:
req = self.request("POST /4/sessions")
self._session_id = req.get("session_key") or req.get("session_id")
return CallableString(self._session_id)
@property
def cluster(self):
"""H2OCluster object describing the underlying cluster."""
return self._cluster
@property
def base_url(self):
"""Base URL of the server, without trailing ``"/"``. For example: ``"https://example.com:54321"``."""
return self._base_url
@property
def name(self):
return self._name
@property
def proxy(self):
"""URL of the proxy server used for the connection (or None if there is no proxy)."""
return self._proxies
@property
def local_server(self):
"""Handler to the H2OLocalServer instance (if connected to one)."""
return self._local_server
@property
def requests_count(self):
"""Total number of request requests made since the connection was opened (used for debug purposes)."""
return self._requests_counter
@property
def timeout_interval(self):
"""Timeout length for each request, in seconds."""
return self._timeout
@timeout_interval.setter
def timeout_interval(self, v):
assert_is_type(v, numeric, None)
self._timeout = v
[docs] def start_logging(self, dest=None):
"""
Start logging all API requests to the provided destination.
:param dest: Where to write the log: either a filename (str), or an open file handle (file). If not given,
then a new temporary file will be created.
"""
assert_is_type(dest, None, str, type(sys.stdout))
if dest is None:
dest = os.path.join(tempfile.mkdtemp(), "h2o-connection.log")
self._print("Now logging all API requests to file %r" % dest)
self._is_logging = True
self._logging_dest = dest
[docs] def stop_logging(self):
"""Stop logging API requests."""
if self._is_logging:
self._print("Logging stopped.")
self._is_logging = False
# ------------------------------------------------------------------------------------------------------------------
# PRIVATE
# ------------------------------------------------------------------------------------------------------------------
def __init__(self):
"""[Private] Please use H2OConnection.connect() to create H2OConnection objects."""
globals()["__H2OCONN__"] = self # for backward-compatibility: __H2OCONN__ is the latest instantiated object
self._stage = 0 # 0 = not connected, 1 = connected, -1 = disconnected
self._session_id = None # Rapids session id; issued upon request only
self._base_url = None # "{scheme}://{ip}:{port}"
self._name = None
self._verify_ssl_cert = None
self._cacert = None
self._auth = None # Authentication token
self._proxies = None # `proxies` dictionary in the format required by the requests module
self._cluster_id = None
self._cookies = None
self._cluster = None # H2OCluster object
self._verbose = None # Print detailed information about connection status
self._requests_counter = 0 # how many API requests were made
self._timeout = None # timeout for a single request (in seconds)
self._is_logging = False # when True, log every request
self._logging_dest = None # where the log messages will be written, either filename or open file handle
self._local_server = None # H2OLocalServer instance to which we are connected (if known)
# self.start_logging(sys.stdout)
def _test_connection(self, max_retries=5, messages=None):
"""
Test that the H2O cluster can be reached, and retrieve basic cluster status info.
:param max_retries: Number of times to try to connect to the cluster (with 0.2s intervals).
:returns: Cluster information (an H2OCluster object)
:raises H2OConnectionError, H2OServerError:
"""
if messages is None:
messages = ("Connecting to H2O server at {url} ..", "successful.", "failed.")
self._print(messages[0].format(url=self._base_url), end="")
cld = None
errors = []
for _ in range(max_retries):
self._print(".", end="", flush=True)
if self._local_server and not self._local_server.is_running():
raise H2OServerError("Local server was unable to start")
try:
define_classes_from_schema(_classes_defined_from_schema_, self)
cld = self.request("GET /3/Cloud")
if self.name and cld.cloud_name != self.name:
raise H2OConnectionError(
"Connected to cloud %s but requested %s." % (cld.cloud_name, self.name)
)
if cld.consensus and cld.cloud_healthy:
self._print(" " + messages[1])
return cld
else:
if cld.consensus and not cld.cloud_healthy:
msg = "in consensus but not healthy"
elif not cld.consensus and cld.cloud_healthy:
msg = "not in consensus but healthy"
else:
msg = "not in consensus and not healthy"
errors.append("Cloud is in a bad shape: %s (size = %d, bad nodes = %d)"
% (msg, cld.cloud_size, cld.bad_nodes))
except (H2OConnectionError, H2OServerError) as e:
message = str(e)
if "\n" in message: message = message[:message.index("\n")]
errors.append("[%s.%02d] %s: %s" %
(time.strftime("%M:%S"), int(time.time() * 100) % 100, e.__class__.__name__, message))
# Cloud too small, or voting in progress, or server is not up yet; sleep then try again
time.sleep(0.2)
self._print(" " + messages[2])
if cld and not cld.cloud_healthy:
raise H2OServerError("Cluster reports unhealthy status")
if cld and not cld.consensus:
raise H2OServerError("Cluster cannot reach consensus")
else:
raise H2OConnectionError("Could not establish link to the H2O cloud %s after %d retries\n%s"
% (self._base_url, max_retries, "\n".join(errors)))
@staticmethod
def _prepare_data_payload(data):
"""
Make a copy of the `data` object, preparing it to be sent to the server.
The data will be sent via x-www-form-urlencoded or multipart/form-data mechanisms. Both of them work with
plain lists of key/value pairs, so this method converts the data into such format.
"""
if not data: return None
res = {}
for key, value in data.items():
if value is None: continue # don't send args set to None so backend defaults take precedence
if isinstance(value, list):
value = stringify_list(value)
elif isinstance(value, dict):
if "__meta" in value and value["__meta"]["schema_name"].endswith("KeyV3"):
value = value["name"]
else:
value = stringify_dict(value)
else:
value = str(value)
res[key] = value
return res
@staticmethod
def _prepare_file_payload(filename):
"""
Prepare `filename` to be sent to the server.
The "preparation" consists of creating a data structure suitable
for passing to requests.request().
"""
if not filename: return None
absfilename = os.path.abspath(filename)
if not os.path.exists(absfilename):
raise H2OValueError("File %s does not exist" % filename, skip_frames=1)
return open(absfilename, "rb")
def _log_start_transaction(self, endpoint, data, json, filename, params):
"""Log the beginning of an API request."""
# TODO: add information about the caller, i.e. which module + line of code called the .request() method
# This can be done by fetching current traceback and then traversing it until we find the request function
self._requests_counter += 1
if not self._is_logging: return
msg = "\n---- %d --------------------------------------------------------\n" % self._requests_counter
msg += "[%s] %s\n" % (time.strftime("%H:%M:%S"), endpoint)
if params is not None: msg += " params: {%s}\n" % ", ".join("%s:%s" % item for item in params.items())
if json is not None:
import json as j
msg += " json: %s\n" % j.dumps(json)
if filename is not None:
msg += " file: %s\n" % filename
elif data is not None:
msg += " body: {%s}\n" % ", ".join("%s:%s" % item for item in data.items())
self._log_message(msg + "\n")
def _log_end_transaction(self, start_time, response):
"""Log response from an API request."""
if not self._is_logging: return
elapsed_time = int((time.time() - start_time) * 1000)
msg = "<<< HTTP %d %s (%d ms)\n" % (response.status_code, response.reason, elapsed_time)
if "Content-Type" in response.headers:
msg += " Content-Type: %s\n" % response.headers["Content-Type"]
msg += response.text
self._log_message(msg + "\n\n")
def _log_end_exception(self, exception):
"""Log API request that resulted in an exception."""
if not self._is_logging: return
self._log_message(">>> %s\n\n" % str(exception))
def _log_message(self, msg):
"""
Log the message `msg` to the destination `self._logging_dest`.
If this destination is a file name, then we append the message to the file and then close the file
immediately. If the destination is an open file handle, then we simply write the message there and do not
attempt to close it.
"""
if is_type(self._logging_dest, str):
with open(self._logging_dest, "at", encoding="utf-8") as f:
f.write(msg)
else:
self._logging_dest.write(msg)
@staticmethod
def _process_response(response, save_to):
"""
Given a response object, prepare it to be handed over to the external caller.
Preparation steps include:
* detect if the response has error status, and convert it to an appropriate exception;
* detect Content-Type, and based on that either parse the response as JSON or return as plain text.
"""
status_code = response.status_code
if status_code == 200 and save_to:
if save_to.startswith("~"): save_to = os.path.expanduser(save_to)
if os.path.isdir(save_to) or save_to.endswith(os.path.sep):
dirname = os.path.join(os.path.abspath(save_to), '')
filename = H2OConnection._find_file_name(response)
else:
dirname, filename = os.path.split(os.path.abspath(save_to))
fullname = os.path.join(dirname, filename)
try:
if not os.path.exists(dirname):
os.makedirs(dirname)
with open(fullname, "wb") as f:
for chunk in response.iter_content(chunk_size=65536):
if chunk: # Empty chunks may occasionally happen
f.write(chunk)
except OSError as e:
raise H2OValueError("Cannot write to file %s: %s" % (fullname, e))
return fullname
content_type = response.headers.get("Content-Type", "")
if ";" in content_type: # Remove a ";charset=..." part
content_type = content_type[:content_type.index(";")]
# this is needed so that response.text() works correctly
response.encoding = response.headers.get("Character-Encoding", response.encoding)
# Auto-detect response type by its content-type. Decode JSON, all other responses pass as-is.
if content_type == "application/json":
try:
data = response.json(object_pairs_hook=H2OResponse)
except (JSONDecodeError, requests.exceptions.ContentDecodingError) as e:
raise H2OServerError("Malformed JSON from server (%s):\n%s" % (str(e), response.text))
else:
data = response.text
# Success (200 = "Ok", 201 = "Created", 202 = "Accepted", 204 = "No Content")
if status_code in {200, 201, 202, 204}:
return data
# Client errors (400 = "Bad Request", 404 = "Not Found", 412 = "Precondition Failed")
if status_code in {400, 404, 412} and isinstance(data, H2OErrorV3):
data.show_stacktrace = False
raise H2OResponseError(data)
# Server errors (notably 500 = "Server Error")
# Note that it is possible to receive valid H2OErrorV3 object in this case, however it merely means the server
# did not provide the correct status code.
raise H2OServerError("HTTP %d %s:\n%s" % (status_code, response.reason, data))
@staticmethod
def _find_file_name(response):
cd = response.headers.get("Content-Disposition", "")
mm = re.search(r'filename="(.*)"$', cd)
return mm.group(1) if mm else "unknown"
def _print(self, msg, flush=False, end="\n"):
"""Helper function to print connection status messages when in verbose mode."""
if self._verbose:
print2(msg, end=end, flush=flush)
def __repr__(self):
if self._stage == 0:
return "<H2OConnection uninitialized>"
elif self._stage == 1:
sess = "session %s" % self._session_id if self._session_id else "no session"
return "<H2OConnection to %s, %s>" % (self._base_url, sess)
else:
return "<H2OConnection closed>"
def __enter__(self):
"""Called when an H2OConnection object is created within the ``with ...`` statement."""
return self
def __exit__(self, *args):
"""Called at the end of the ``with ...`` statement."""
self.close()
assert len(args) == 3 # Avoid warning about unused args...
return False # ensure that any exception will be re-raised
class H2OResponse(dict):
"""Temporary..."""
def __new__(cls, keyvals):
# This method is called by the simplejson.json(object_pairs_hook=<this>)
# `keyvals` is a list of (key,value) tuples. For example:
# [("schema_version", 3), ("schema_name", "InitIDV3"), ("schema_type", "Iced")]
schema = None
for k, v in keyvals:
if k == "__meta" and isinstance(v, dict):
schema = v["schema_name"]
break
if k == "__schema" and is_type(v, str):
schema = v
break
if schema == "MetadataV3": return H2OMetadataV3.make(keyvals)
if schema == "CloudV3": return H2OCluster.make(keyvals)
if schema == "H2OErrorV3": return H2OErrorV3.make(keyvals)
if schema == "H2OModelBuilderErrorV3": return H2OModelBuilderErrorV3.make(keyvals)
if schema == "TwoDimTableV3": return H2OTwoDimTable.make(keyvals)
if schema and schema.startswith("ModelMetrics"):
metrics = make_metrics(schema, keyvals)
if metrics is not None:
return metrics
return super(H2OResponse, cls).__new__(cls, keyvals)
# def __getattr__(self, key):
# """This gets invoked for any attribute "key" that is NOT yet defined on the object."""
# if key in self:
# return self[key]
# return None
_classes_defined_from_schema_ = [H2OCluster, H2OErrorV3, H2OModelBuilderErrorV3]
# Find the exception that occurs on invalid JSON input
JSONDecodeError, _r = None, None
try:
_r = requests.Response()
_r._content = b"haha"
_r.json()
except Exception as exc:
JSONDecodeError = type(exc)
del _r
# ----------------------------------------------------------------------------------------------------------------------
# Deprecated method implementations
# ----------------------------------------------------------------------------------------------------------------------
__H2OCONN__ = H2OConnection() # Latest instantiated H2OConnection object. Do not use in any new code!
__H2O_REST_API_VERSION__ = 3 # Has no actual meaning
def _deprecated_default():
H2OConnection.__ENCODING__ = "utf-8"
H2OConnection.__ENCODING_ERROR__ = "replace"
def _deprecated_username():
auth = getattr(__H2OCONN__, "_auth")
return auth[0] if isinstance(auth, tuple) else None
def _deprecated_password():
auth = getattr(__H2OCONN__, "_auth")
return auth[1] if isinstance(auth, tuple) else None
def _deprecated_check_conn():
if not __H2OCONN__:
raise H2OConnectionError("No active connection to an H2O cluster. Try calling `h2o.connect()`")
return __H2OCONN__
def _deprecated_get(self, url_suffix, **kwargs):
restver = kwargs.pop("_rest_version") if "_rest_version" in kwargs else 3
endpoint = "GET /%d/%s" % (restver, url_suffix)
return self.request(endpoint, data=kwargs)
def _deprecated_post(self, url_suffix, **kwargs):
restver = kwargs.pop("_rest_version") if "_rest_version" in kwargs else 3
endpoint = "POST /%d/%s" % (restver, url_suffix)
filename = None
if "file_upload_info" in kwargs:
filename = next(iter(kwargs.pop("file_upload_info").values()))
return self.request(endpoint, data=kwargs, filename=filename)
def _deprecated_delete(self, url_suffix, **kwargs):
restver = kwargs.pop("_rest_version") if "_rest_version" in kwargs else 3
endpoint = "DELETE /%d/%s" % (restver, url_suffix)
return self.request(endpoint, data=kwargs)
def end_session():
"""Deprecated, use connection.close() instead."""
print("Warning: end_session() is deprecated")
__H2OCONN__.close()