feat(runner): add skip_if_no_inputs to workflows

secator/runners/command.py

@@ -124,9 +124,12 @@ def __init__(self, inputs=[], **run_opts):
 		caller = run_opts.get('caller', None)
 		results = run_opts.pop('results', [])
 		context = run_opts.pop('context', {})
+		self.skip_if_no_inputs = run_opts.pop('skip_if_no_inputs', False)
 
 		# Prepare validators
-		input_validators = [self._validate_input_nonempty]
+		input_validators = []
+		if not self.skip_if_no_inputs:
+			input_validators.append(self._validate_input_nonempty)
 		if not caller:
 			input_validators.append(self._validate_chunked_input)
 		validators = {'validate_input': input_validators}
@@ -176,22 +179,6 @@ def __init__(self, inputs=[], **run_opts):
 			item_loaders.append(instance_func)
 		self.item_loaders = item_loaders
 
-		# Print built cmd
-		if not self.has_children:
-			if self.sync:
-				if self.caller and self.description:
-					self._print(f'\n:wrench: {self.description} ...', color='bold gold3', rich=True)
-				elif self.print_cmd:
-					self._print('')
-		if self.print_cmd:
-			cmd_str = self.cmd.replace('[', '\\[')
-			if self.sync and self.chunk and self.chunk_count:
-				cmd_str += f' [dim gray11]({self.chunk}/{self.chunk_count})[/]'
-			self._print(cmd_str, color='bold cyan', rich=True)
-
-		# Debug
-		self.debug('Command', obj={'cmd': self.cmd}, sub='init')
-
 	def toDict(self):
 		res = super().toDict()
 		res.update({
@@ -347,6 +334,8 @@ def yielder(self):
 			if self.has_children:
 				return
 
+			self.print_description()
+
 			# Yield targets
 			for input in self.inputs:
 				yield Target(name=input, _source=self.unique_name, _uuid=str(uuid.uuid4()))
@@ -379,6 +368,7 @@ def yielder(self):
 				shell=self.shell,
 				env=env,
 				cwd=self.cwd)
+			self.print_command()
 
 			# If sudo password is provided, send it to stdin
 			if sudo_password:
@@ -447,6 +437,23 @@ def process_line(self, line):
 		yield from self.stats()
 		self.last_updated_stat = time()
 
+	def print_description(self):
+		"""Print description"""
+		if self.sync and not self.has_children:
+			if self.caller and self.description:
+				self._print(f'\n[bold gold3]:wrench: {self.description} [dim cyan]({self.config.name})[/][/] ...', rich=True)
+			elif self.print_cmd:
+				self._print('')
+
+	def print_command(self):
+		"""Print command."""
+		if self.print_cmd:
+			cmd_str = self.cmd.replace('[', '\\[')
+			if self.sync and self.chunk and self.chunk_count:
+				cmd_str += f' [dim gray11]({self.chunk}/{self.chunk_count})[/]'
+			self._print(cmd_str, color='bold cyan', rich=True)
+		self.debug('Command', obj={'cmd': self.cmd}, sub='init')
+
 	def handle_file_not_found(self, exc):
 		"""Handle case where binary is not found.
 

secator/runners/workflow.py

@@ -123,6 +123,7 @@ def get_tasks(self, config, inputs, workflow_opts, run_opts):
 				opts['context'] = self.context.copy()
 				opts['name'] = task_name
 				opts['has_parent'] = True
+				opts['skip_if_no_inputs'] = True
 
 				# Create task signature
 				task_id = str(uuid.uuid4())