def validate_file_path()

in source/deploy_dag.py [0:0]

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

    def validate_file_path(self):
        """This function validates the file path for given dag and tasks_variables """
        """
        :param self
        :return str dag_filename: DAG filename
        :return str dag_directory_path: DAG local file path
        :return str dag_file_validation: DAG File validation flag Success or Fail
        :return str tasks_variables_filename: Task Variables filename
        :return str tasks_variables_directory_path: Tasks Variables directory file path
        :return str tasks_variables_file_validation: Tasks Variables directory file validation flag Success or Fail
        """
        try:
            logging.info("Reading dag_file and tasks_variables_file to validate file path")
            dag_directory = str(self.dag_file).split("/")
            dag_directory_path = '/'.join(dag_directory[:-1])
            dag_filename = dag_directory[-1]
            dag_files = [ file for file in  os.listdir(dag_directory_path)]

            if dag_filename in dag_files:
                dag_file_validation = "success"
                logging.info("DAG file path successfully validated")
            else:
                dag_file_validation = "fail"
                logging.error(" Please check DAG filename or file path as file:{}\
                            not found at the given path {}".format(dag_filename,dag_directory_path))

            if self.tasks_variables_flag == "True":
                tasks_variables_directory = str(self.tasks_variables_file).split("/")
                tasks_variables_directory_path = '/'.join(tasks_variables_directory[:-1])
                tasks_variables_filename = tasks_variables_directory[-1]
                tasks_variables_files = [ file for file in  os.listdir(tasks_variables_directory_path)]
                if tasks_variables_filename in tasks_variables_files:
                    tasks_variables_file_validation = "success"
                    logging.info("Tasks Variables file path successfully validated")
                else:
                    tasks_variables_file_validation = "fail"
                    logging.error(" Check tasks variables filename or file path as file:{}\
                            not found at the given path {}".format(tasks_variables_filename,tasks_variables_directory))
                    raise Exception(" Check tasks variables filename or file path as file:{}\
                            not found at the given path {}".format(tasks_variables_filename,tasks_variables_directory))
                    
            elif self.tasks_variables_flag == "False":
                tasks_variables_file_validation = "pass"
                tasks_variables_filename = ""
                tasks_variables_directory_path = ""
                logging.warning("Skipping to validate tasks variables file path as tasks_variables_flag value is False ")
            else:
                tasks_variables_file_validation = "fail"
                raise Exception("Check tasks variables filename or file path as file:{}\
                            not found at the given path {}".format(tasks_variables_filename,tasks_variables_directory))
                
        except Exception as error:
            logging.error(" File validation failed due to following errors : {}".format(str(type(error).__name__)+" --> "+ str(error)))
            return dag_file_validation,tasks_variables_file_validation
        else:
            return dag_filename,dag_directory_path,dag_file_validation,\
                tasks_variables_filename,tasks_variables_directory_path,tasks_variables_file_validation