Add live_task_id to base task.
@@ -15,6 +15,7 @@ import logging
from collections import OrderedDict
from contextlib import contextmanager
from abc import abstractmethod
+from inspect import getargspec
import luigi
import luigi.util
@@ -137,6 +138,24 @@ class BaseTask(luigi.Task):
return all(t.exists() for t in outputs)
+ @property
+ def live_task_id(self):
+ """
+ The task id depends on the task family and parameters, and is generated by luigi once in the
+ constructor. As the latter may change, this property returns to the id with the current set
+ of parameters.
+ """
+ # create string params (only_public was introduced in 2.8.0, so check if that arg exists)
+ str_params_kwargs = {"only_significant": True}
+ if "only_public" in getargspec(self.to_str_params).args:
+ str_params_kwargs["only_public"] = True
+ str_params = self.to_str_params(**str_params_kwargs)
+
+ # create the task id
+ task_id = luigi.task.task_id_str(self.get_task_family(), str_params)
+
+ return task_id
+
def walk_deps(self, max_depth=-1, order="level"):
# see https://en.wikipedia.org/wiki/Tree_traversal
if order not in ("level", "pre"):
