def create_sproc_from_script()

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

• 21 lines of code
• 7 McCabe index (conditional complexity)

    def create_sproc_from_script(self, name: str, path_to_script: str,
                                 input_params: dict = None, output_params: dict = None):
        """Create a SQL Server stored procedure based on a Python script

        :param name: name of stored procedure.
        :param path_to_script: file path to Python script to create a sproc from.
        :param input_params: optional dictionary of type annotations for inputs in the script
        :param output_params optional dictionary of type annotations for each output variable
        :return: True if creation succeeded

        >>> from sqlmlutils import ConnectionInfo, SQLPythonExecutor
        >>>
        >>>
        >>> sqlpy = SQLPythonExecutor(ConnectionInfo("localhost", database="AutoRegressTestDB"))
        >>> sqlpy.create_sproc_from_script(name="script_sproc", path_to_script="path/to/script")
        >>>
        >>> # This will execute the script in sql; with no inputs or outputs it will just run and return nothing
        >>> sqlpy.execute_sproc(name="script_sproc")
        >>> sqlpy.drop_sproc(name="script_sproc")

        """
        if input_params is None:
            input_params = {}
        if output_params is None:
            output_params = {}
        # Save the stored procedure in database
        try:
            with open(path_to_script, 'r') as script_file:
                content = script_file.read()
            print("File does exist, using " + path_to_script)
        except FileNotFoundError:
            raise FileNotFoundError("File does not exist!")

        # We modify input_params/output_params because we add stdout and stderr as params. 
        # We copy here to avoid modifying the underlying contents.
        #
        in_copy = input_params.copy() if input_params is not None else None
        out_copy = output_params.copy() if output_params is not None else None
        
        execute_query(StoredProcedureBuilder(name=name, 
                                            script=content, 
                                            input_params=in_copy,
                                            output_params=out_copy, 
                                            language_name=self._language_name),
                        self._connection_info)
        return True