tasks.ref(...)

Function used to retrieve the output of other tasks.

tasks.ref(task_name: str, local: bool = False)

• Parameters

  • task: str: the name of the task whose output you'd like to retrieve. This name should be in the form <module_name>, or <module_name>.<task_name>, where <task_name> is the name of the Prism task class / function.

  • local: bool: whether task lives in the same module. The default is False.

Examples:

Non-local tasks:

Class-based tasks

import prism.task
import prism.target
import prism.decorators

class ExampleTask(prism.task.PrismTask):
    
    @prism.decorators.target(
        type=prism.target.Txt, 
        loc="/Users/hello_world.txt",
        **kwargs
    )
    def run(self, tasks, hooks):
        test_str = "Hello, world!"

        # Get output from task in first_task.py. If this module
        # contains multiple tasks, then we'll need to specify
        # the task via <module_name>.<task_name>
        previous_task_output = tasks.ref("first_task")
        return test_str + previous_task_output

Function-based tasks

from prism.decorators import task

@task()
def example_function(tasks, hooks):
    test_str = "Hello, world!"
    
    # Get output from task in first_task.py. If this module
    # contains multiple tasks, then we'll need to specify
    # the task via <module_name>.<task_name>
    previous_task_output = tasks.ref("first_task")

    return test_str + previous_task_output

Local tasks:

Class-based tasks

import prism.task
import prism.target
import prism.decorators

class FirstTask(prism.task.PrismTask):
    
    def run(self, tasks, hooks):
        # do something


class ExampleTask(prism.task.PrismTask):
    
    @prism.decorators.target(
        type=prism.target.Txt, 
        loc="/Users/hello_world.txt",
        **kwargs
    )
    def run(self, tasks, hooks):
        test_str = "Hello, world!"
        previous_task_output = tasks.ref("FirstTask", local=True)
        return test_str + previous_task_output

Function-based tasks

from prism.decorators import task

@task()
def first_task(tasks, hooks):
    # do something


@task()
def example_function(tasks, hooks):
    test_str = "Hello, world!"
    previous_task_output = tasks.ref("first_task", local=True)
    return test_str + previous_task_output