Skip to content

basic_example.py

basic_example

repo = CoSyLuigiRepo(TaskA, TaskB) module-attribute

maestro = Maestro(repo.cls_repo, repo.taxonomy) module-attribute

results = list(maestro.query(TaskB.target())) module-attribute

TaskA 


              flowchart TD
              getting_started.basic_example.TaskA[TaskA]

              

              click getting_started.basic_example.TaskA href "" "getting_started.basic_example.TaskA"
Source code in examples/getting_started/basic_example.py 
 9
10
11
12
13
14
15
class TaskA(CoSyLuigiTask):
    def output(self):
        return {"a_artifact": luigi.LocalTarget("output/task_a_output.txt")}

    def run(self):
        with self.output()["a_artifact"].open("w") as f:
            f.write("Task A completed")

output()

Source code in examples/getting_started/basic_example.py 
10
11
def output(self):
    return {"a_artifact": luigi.LocalTarget("output/task_a_output.txt")}

run()

Source code in examples/getting_started/basic_example.py 
13
14
15
def run(self):
    with self.output()["a_artifact"].open("w") as f:
        f.write("Task A completed")

TaskB 


              flowchart TD
              getting_started.basic_example.TaskB[TaskB]

              

              click getting_started.basic_example.TaskB href "" "getting_started.basic_example.TaskB"
Source code in examples/getting_started/basic_example.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
class TaskB(CoSyLuigiTask):
    task_a = CoSyLuigiTaskParameter(TaskA)

    def output(self):
        return {"b_artifact": luigi.LocalTarget("output/task_b_output.txt")}

    def run(self):
        with (
            self.input()["task_a"]["a_artifact"].open() as input_file,
            self.output()["b_artifact"].open("w") as output_file,
        ):
            data = input_file.read()
            output_file.write("Task B completed with input: " + data)

task_a = CoSyLuigiTaskParameter(TaskA) class-attribute instance-attribute

output()

Source code in examples/getting_started/basic_example.py 
21
22
def output(self):
    return {"b_artifact": luigi.LocalTarget("output/task_b_output.txt")}

run()

Source code in examples/getting_started/basic_example.py 
24
25
26
27
28
29
30
def run(self):
    with (
        self.input()["task_a"]["a_artifact"].open() as input_file,
        self.output()["b_artifact"].open("w") as output_file,
    ):
        data = input_file.read()
        output_file.write("Task B completed with input: " + data)