in docker_images/sklearn/app/pipelines/common.py [0:0]
def __call__(self, inputs: Any) -> Any:
"""Handle call for getting the model prediction
This method is responsible for handling all possible errors and
warnings. To get the actual prediction, implement the `_get_output`
method.
The types of the inputs and output depend on the specific task being
implemented.
"""
if self._load_exception:
# there has been an error while loading the model. We need to raise
# that, and can't call predict on the model.
raise ValueError(
"An error occurred while loading the model: "
f"{str(self._load_exception)}"
)
_warnings = []
if self.columns:
# TODO: we should probably warn if columns are not configured, we
# really do need them.
given_cols = set(inputs["data"].keys())
expected = set(self.columns)
extra = given_cols - expected
missing = expected - given_cols
if extra:
_warnings.append(
f"The following columns were given but not expected: {extra}"
)
if missing:
_warnings.append(
f"The following columns were expected but not given: {missing}"
)
exception = None
try:
with warnings.catch_warnings(record=True) as record:
res = self._get_output(inputs)
except Exception as e:
exception = e
for warning in record:
_warnings.append(f"{warning.category.__name__}({warning.message})")
for warning in self._load_warnings:
_warnings.append(f"{warning.category.__name__}({warning.message})")
if _warnings:
for warning in _warnings:
logger.warning(warning)
if not exception:
# we raise an error if there are any warnings, so that routes.py
# can catch and return a non 200 status code.
error = {
"error": "There were warnings while running the model.",
"output": res,
"warnings": _warnings, # see issue #96
}
raise ValueError(json.dumps(error))
else:
# if there was an exception, we raise it so that routes.py can
# catch and return a non 200 status code.
raise exception
if exception:
raise exception
return res