def get_parameter_type()

in assets/inference/environments/mlflow-py312-inference/context/mlflow_score_script.py [0:0]

• 47 lines of code
• 19 McCabe index (conditional complexity)

def get_parameter_type(sample_input_ex, sample_output_ex=None, sample_param_ex=None):
    """Get parameter type."""
    if sample_input_ex is None:
        _logger.info("sample input is none, returning NoSampleParameterType")
        input_param = NoSampleParameterType()
    else:
        try:
            # schema = _infer_schema(sample_input_ex)
            # schema_types = schema.input_types
            _infer_schema(sample_input_ex)
        except MlflowException:
            pass
        finally:
            if isinstance(sample_input_ex, np.ndarray):
                _logger.info("sample input is a numpy array")
                # Unnamed tensor input
                input_param = NumpyParameterType(sample_input_ex, enforce_shape=False)
            elif pandas_installed and isinstance(sample_input_ex, pd.DataFrame):
                _logger.info("sample input is a dataframe")
                # TODO check with OSS about pd.Series
                input_param = PandasParameterType(sample_input_ex, enforce_shape=False, orient='split')
            # elif schema_types and isinstance(sample_input_ex, dict) and \
            #     not all(stype == DataType.string for stype in schema_types) and \
            #     all(isinstance(value, list) for value in sample_input_ex.values()):
            #     # for dictionaries where there is any non-string type, named tensor
            #     param_arg = {}
            #     for key, value in sample_input_ex.items():
            #         param_arg[key] = NumpyParameterType(value, enforce_shape=False)
            #     input_param = StandardPythonParameterType(param_arg)
            elif isinstance(sample_input_ex, dict) and is_transformers:
                input_param = StandardPythonParameterType(sample_input_ex)
            elif isinstance(sample_input_ex, dict):
                _logger.info("sample input is a dict")
                # TODO keeping this around while _infer_schema doesn't work on dataframe string signatures
                param_arg = {}
                for key, value in sample_input_ex.items():
                    param_arg[key] = NumpyParameterType(value, enforce_shape=False)
                input_param = StandardPythonParameterType(param_arg)
            elif isinstance(sample_input_ex, list) and is_transformers:
                _logger.info("transformers sample input is a list")
                input_param = StandardPythonParameterType(sample_input_ex)
            else:
                _logger.info("sample input is string, bytes, or non-transformers list")
                # strings, bytes, lists and dictionaries with only strings as base type
                input_param = NoSampleParameterType()

    if sample_output_ex is None:
        output_param = NoSampleParameterType()
    else:
        if isinstance(sample_output_ex, np.ndarray):
            # Unnamed tensor input
            output_param = NumpyParameterType(sample_output_ex, enforce_shape=False)
        elif isinstance(sample_output_ex, dict):
            param_arg = {}
            for key, value in sample_output_ex.items():
                param_arg[key] = NumpyParameterType(value, enforce_shape=False)
            output_param = StandardPythonParameterType(param_arg)
        else:
            output_param = PandasParameterType(sample_output_ex, enforce_shape=False, orient='records')

    if sample_param_ex is None:
        param_param = NoSampleParameterType()
    else:
        param_param = StandardPythonParameterType(sample_param_ex)

    return input_param, output_param, param_param