Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def exists(self):
if not super().exists():
return False
try:
import ROOT
except ImportError:
raise ImportError("Can not import ROOT. The ROOTLocalTarget can not be used!")
path = self.path
tfile = ROOT.TFile.Open(path)
return tfile and len(tfile.GetListOfKeys()) > 0
class Task(luigi.Task, TaskWithDataSupport):
git_hash = luigi.Parameter(default=utils.get_basf2_git_hash())
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
filename = os.path.realpath(sys.argv[0])
log_folder = os.path.join(os.path.dirname(filename), "logs")
stdout_file_name = self.get_output_file_name(self.get_task_family() + "_stdout", create_folder=True,
result_path=log_folder)
stderr_file_name = self.get_output_file_name(self.get_task_family() + "_stderr", create_folder=True,
result_path=log_folder)
self.log_files = {"stdout": stdout_file_name,
"stderr": stderr_file_name,
"log_folder": log_folder}
def _transform_input(input_generator, key=None):
input_list = utils.flatten_to_list_of_dicts(input_generator)
file_paths = utils.flatten_to_file_paths(input_list)
if key is not None:
return file_paths[key]
else:
return file_paths