def __init__()

in Python/sqlmlutils/sqlbuilder.py [0:0]

• 43 lines of code
• 16 McCabe index (conditional complexity)

    def __init__(self, 
                name: str, func: Callable,
                input_params: dict = None, 
                output_params: dict = None,
                language_name: str = "Python"):
        """StoredProcedureBuilderFromFunction SQL stored procedures based on Python functions.

        :param name: name of the stored procedure
        :param func: function to base the stored procedure on
        :param input_params: input parameters type annotation dictionary for the stored procedure
        Can use function type annotations instead; if both, they must match
        :param output_params: output parameters type annotation dictionary from the stored procedure
        :param language_name: name of the language to be executed in sp_execute_external_script, if using EXTERNAL LANGUAGE
        """
        if input_params is None:
            input_params = {}
        if output_params is None:
            output_params = {}
            
        output_params[STDOUT_COLUMN_NAME] = str
        output_params[STDERR_COLUMN_NAME] = str

        self._func = func
        self._name = name
        self._output_params = output_params
        self._language_name = language_name

        # Get function text and escape single quotes
        function_text = textwrap.dedent(inspect.getsource(self._func)).replace("'","''")

        # Get function arguments and type annotations
        argspec = inspect.getfullargspec(self._func)
        names_of_input_args = argspec.args
        annotations = argspec.annotations

        if argspec.defaults is not None:
            warnings.warn("Default values are not supported")

        # Figure out input and output parameter dictionaries
        if input_params != {}:
            if annotations != {} and annotations != input_params:
                raise ValueError("Annotations and input_params do not match!")
            self._input_params = input_params
        elif annotations != {}:
            self._input_params = annotations
        elif len(names_of_input_args) == 0:
            self._input_params = {}

        names_of_output_args = list(self._output_params)
        
        if len(names_of_input_args) != len(self._input_params):
            raise ValueError("Number of argument annotations doesn't match the number of arguments!")
        if set(names_of_input_args) != set(self._input_params.keys()):
            raise ValueError("Names of arguments do not match the annotation keys!")
                
        calling_text = self.get_function_calling_text(self._func, names_of_input_args)

        output_data_set = None
        for name in names_of_output_args:
            if self._output_params[name] == DataFrame:
                names_of_output_args.remove(name)
                output_data_set = name
                break

        ending = self.get_ending(self._output_params, output_data_set)
        # Creates the base python script to put in the SPEES query.
        # Arguments to function are passed by name into script using SPEES @params argument.
        self._script = """