code: def delegates(task_that_delegates):
      
      if not hasattr(task_that_delegates, 'subtasks'):
          # This method can (optionally) define a couple of delegate tasks that
          # will be accessible as interfaces, meaning that the task can access
          # those tasks and run methods defined on them, etc
          raise AttributeError('%s needs to implement the method "subtasks"' % task_that_delegates)
  
      @task._task_wraps(task_that_delegates)
      class Wrapped(task_that_delegates):
  
          def deps(self):
              # Overrides method in base class
              return task.flatten(self.requires()) + task.flatten([t.deps() for t in task.flatten(self.subtasks())])
  
          def run(self):
              for t in task.flatten(self.subtasks()):
                  t.run()
              task_that_delegates.run(self)
  
      return Wrapped
docs: Lets a task call methods on subtask(s).
  
      The way this works is that the subtask is run as a part of the task, but
      the task itself doesn't have to care about the requirements of the subtasks.
      The subtask doesn't exist from the scheduler's point of view, and
      its dependencies are instead required by the main task.
  
      Example:
  
      .. code-block:: python
  
          class PowersOfN(luigi.Task):
              n = luigi.IntParameter()
              def f(self, x): return x ** self.n
  
          @delegates
          class T(luigi.Task):
              def subtasks(self): return PowersOfN(5)
              def run(self): print self.subtasks().f(42)