Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_log_batch_nonexistent_run(self):
fs = FileStore(self.test_root)
nonexistent_uuid = uuid.uuid4().hex
with self.assertRaises(MlflowException) as e:
fs.log_batch(nonexistent_uuid, [], [], [])
assert e.exception.error_code == ErrorCode.Name(RESOURCE_DOES_NOT_EXIST)
assert ("Run '%s' not found" % nonexistent_uuid) in e.exception.message
def test_rest_exception_error_code_and_no_message():
exc = RestException({"error_code": ErrorCode.Name(RESOURCE_DOES_NOT_EXIST),
"messages": "something important."})
assert "something important." in str(exc)
assert "RESOURCE_DOES_NOT_EXIST" in str(exc)
json.loads(exc.serialize_as_json())
def test_scoring_server_responds_to_malformed_json_input_with_stacktrace_and_error_code(
sklearn_model, model_path):
mlflow.sklearn.save_model(sk_model=sklearn_model.model, path=model_path)
malformed_json_content = "this is,,,, not valid json"
response = pyfunc_serve_and_score_model(
model_uri=os.path.abspath(model_path),
data=malformed_json_content,
content_type=pyfunc_scoring_server.CONTENT_TYPE_JSON_SPLIT_ORIENTED)
response_json = json.loads(response.content)
assert "error_code" in response_json
assert response_json["error_code"] == ErrorCode.Name(MALFORMED_REQUEST)
assert "message" in response_json
assert "stack_trace" in response_json
def test_validate_metric_name():
for good_name in GOOD_METRIC_OR_PARAM_NAMES:
_validate_metric_name(good_name)
for bad_name in BAD_METRIC_OR_PARAM_NAMES:
with pytest.raises(MlflowException, match="Invalid metric name") as e:
_validate_metric_name(bad_name)
assert e.value.error_code == ErrorCode.Name(INVALID_PARAMETER_VALUE)
def test_attempting_to_deploy_in_asynchronous_mode_without_archiving_throws_exception(
pretrained_model):
with pytest.raises(MlflowException) as exc:
mfs.deploy(app_name="test-app",
model_uri=pretrained_model.model_uri,
mode=mfs.DEPLOYMENT_MODE_CREATE,
archive=False,
synchronous=False)
assert "Resources must be archived" in exc.value.message
assert exc.value.error_code == ErrorCode.Name(INVALID_PARAMETER_VALUE)
import json
from mlflow.protos.databricks_pb2 import INTERNAL_ERROR, TEMPORARILY_UNAVAILABLE, \
ENDPOINT_NOT_FOUND, PERMISSION_DENIED, REQUEST_LIMIT_EXCEEDED, BAD_REQUEST, \
INVALID_PARAMETER_VALUE, RESOURCE_DOES_NOT_EXIST, INVALID_STATE, RESOURCE_ALREADY_EXISTS, \
ErrorCode
ERROR_CODE_TO_HTTP_STATUS = {
ErrorCode.Name(INTERNAL_ERROR): 500,
ErrorCode.Name(INVALID_STATE): 500,
ErrorCode.Name(TEMPORARILY_UNAVAILABLE): 503,
ErrorCode.Name(REQUEST_LIMIT_EXCEEDED): 429,
ErrorCode.Name(ENDPOINT_NOT_FOUND): 404,
ErrorCode.Name(RESOURCE_DOES_NOT_EXIST): 404,
ErrorCode.Name(PERMISSION_DENIED): 403,
ErrorCode.Name(BAD_REQUEST): 400,
ErrorCode.Name(RESOURCE_ALREADY_EXISTS): 400,
ErrorCode.Name(INVALID_PARAMETER_VALUE): 400
}
class MlflowException(Exception):
"""
Generic exception thrown to surface failure information about external-facing operations.
The error message associated with this exception may be exposed to clients in HTTP responses
for debugging purposes. If the error text is sensitive, raise a generic `Exception` object
instead.
"""
def get_experiment_by_name(self, experiment_name):
try:
req_body = message_to_json(GetExperimentByName(experiment_name=experiment_name))
response_proto = self._call_endpoint(GetExperimentByName, req_body)
return Experiment.from_proto(response_proto.experiment)
except MlflowException as e:
if e.error_code == databricks_pb2.ErrorCode.Name(
databricks_pb2.RESOURCE_DOES_NOT_EXIST):
return None
elif e.error_code == databricks_pb2.ErrorCode.Name(
databricks_pb2.REQUEST_LIMIT_EXCEEDED):
raise e
# Fall back to using ListExperiments-based implementation.
for experiment in self.list_experiments(ViewType.ALL):
if experiment.name == experiment_name:
return experiment
return None
from mlflow.protos.databricks_pb2 import INTERNAL_ERROR, TEMPORARILY_UNAVAILABLE, \
ENDPOINT_NOT_FOUND, PERMISSION_DENIED, REQUEST_LIMIT_EXCEEDED, BAD_REQUEST, \
INVALID_PARAMETER_VALUE, RESOURCE_DOES_NOT_EXIST, INVALID_STATE, RESOURCE_ALREADY_EXISTS, \
ErrorCode
ERROR_CODE_TO_HTTP_STATUS = {
ErrorCode.Name(INTERNAL_ERROR): 500,
ErrorCode.Name(INVALID_STATE): 500,
ErrorCode.Name(TEMPORARILY_UNAVAILABLE): 503,
ErrorCode.Name(REQUEST_LIMIT_EXCEEDED): 429,
ErrorCode.Name(ENDPOINT_NOT_FOUND): 404,
ErrorCode.Name(RESOURCE_DOES_NOT_EXIST): 404,
ErrorCode.Name(PERMISSION_DENIED): 403,
ErrorCode.Name(BAD_REQUEST): 400,
ErrorCode.Name(RESOURCE_ALREADY_EXISTS): 400,
ErrorCode.Name(INVALID_PARAMETER_VALUE): 400
}
class MlflowException(Exception):
"""
Generic exception thrown to surface failure information about external-facing operations.
The error message associated with this exception may be exposed to clients in HTTP responses
for debugging purposes. If the error text is sensitive, raise a generic `Exception` object
instead.
"""
def __init__(self, message, error_code=INTERNAL_ERROR, **kwargs):
"""
:param message: The message describing the error that occured. This will be included in the
exception's serialized JSON representation.
:param error_code: An appropriate error code for the error that occured; it will be included
in the exception's serialized JSON representation. This should be one of
def get_experiment_by_name(self, experiment_name):
try:
req_body = message_to_json(GetExperimentByName(experiment_name=experiment_name))
response_proto = self._call_endpoint(GetExperimentByName, req_body)
return Experiment.from_proto(response_proto.experiment)
except MlflowException as e:
if e.error_code == databricks_pb2.ErrorCode.Name(
databricks_pb2.RESOURCE_DOES_NOT_EXIST):
return None
elif e.error_code == databricks_pb2.ErrorCode.Name(
databricks_pb2.REQUEST_LIMIT_EXCEEDED):
raise e
# Fall back to using ListExperiments-based implementation.
for experiment in self.list_experiments(ViewType.ALL):
if experiment.name == experiment_name:
return experiment
return None
import json
from mlflow.protos.databricks_pb2 import INTERNAL_ERROR, TEMPORARILY_UNAVAILABLE, \
ENDPOINT_NOT_FOUND, PERMISSION_DENIED, REQUEST_LIMIT_EXCEEDED, BAD_REQUEST, \
INVALID_PARAMETER_VALUE, RESOURCE_DOES_NOT_EXIST, INVALID_STATE, RESOURCE_ALREADY_EXISTS, \
ErrorCode
ERROR_CODE_TO_HTTP_STATUS = {
ErrorCode.Name(INTERNAL_ERROR): 500,
ErrorCode.Name(INVALID_STATE): 500,
ErrorCode.Name(TEMPORARILY_UNAVAILABLE): 503,
ErrorCode.Name(REQUEST_LIMIT_EXCEEDED): 429,
ErrorCode.Name(ENDPOINT_NOT_FOUND): 404,
ErrorCode.Name(RESOURCE_DOES_NOT_EXIST): 404,
ErrorCode.Name(PERMISSION_DENIED): 403,
ErrorCode.Name(BAD_REQUEST): 400,
ErrorCode.Name(RESOURCE_ALREADY_EXISTS): 400,
ErrorCode.Name(INVALID_PARAMETER_VALUE): 400
}
class MlflowException(Exception):
"""
Generic exception thrown to surface failure information about external-facing operations.
The error message associated with this exception may be exposed to clients in HTTP responses
for debugging purposes. If the error text is sensitive, raise a generic `Exception` object
instead.
"""
def __init__(self, message, error_code=INTERNAL_ERROR, **kwargs):
"""
:param message: The message describing the error that occured. This will be included in the
exception's serialized JSON representation.