Skip to content

variation_example.py

variation_example

WriteTemplateTask 


              flowchart TD
              getting_started.variation_example.WriteTemplateTask[WriteTemplateTask]

              

              click getting_started.variation_example.WriteTemplateTask href "" "getting_started.variation_example.WriteTemplateTask"
Source code in examples/getting_started/variation_example.py 
11
12
13
14
15
16
17
class WriteTemplateTask(CoSyLuigiTask):
    def output(self):
        return {"template": luigi.LocalTarget("hello_world_template.txt")}

    def run(self):
        with self.output()["template"].open("w") as result:
            result.write("Hello World $name")

output()

Source code in examples/getting_started/variation_example.py 
12
13
def output(self):
    return {"template": luigi.LocalTarget("hello_world_template.txt")}

run()

Source code in examples/getting_started/variation_example.py 
15
16
17
def run(self):
    with self.output()["template"].open("w") as result:
        result.write("Hello World $name")

SubstituteNameTask 


              flowchart TD
              getting_started.variation_example.SubstituteNameTask[SubstituteNameTask]

              

              click getting_started.variation_example.SubstituteNameTask href "" "getting_started.variation_example.SubstituteNameTask"
Source code in examples/getting_started/variation_example.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
class SubstituteNameTask(CoSyLuigiTask, ABC):
    template_task = CoSyLuigiTaskParameter(WriteTemplateTask)
    name: str = None

    def output(self):
        return {"filled_template": luigi.LocalTarget(self.__class__.__name__ + "_filled_template.txt")}

    def run(self):
        with self.input()["template_task"]["template"].open() as input_template:
            template = Template(input_template.read())
            result = template.substitute(name=self.name)
            with self.output()["filled_template"].open("w") as outfile:
                outfile.write(result)

template_task = CoSyLuigiTaskParameter(WriteTemplateTask) class-attribute instance-attribute

name: str = None class-attribute instance-attribute

output()

Source code in examples/getting_started/variation_example.py 
24
25
def output(self):
    return {"filled_template": luigi.LocalTarget(self.__class__.__name__ + "_filled_template.txt")}

run()

Source code in examples/getting_started/variation_example.py 
27
28
29
30
31
32
def run(self):
    with self.input()["template_task"]["template"].open() as input_template:
        template = Template(input_template.read())
        result = template.substitute(name=self.name)
        with self.output()["filled_template"].open("w") as outfile:
            outfile.write(result)

SubstituteNameByJohnDoeTask 


              flowchart TD
              getting_started.variation_example.SubstituteNameByJohnDoeTask[SubstituteNameByJohnDoeTask]
              getting_started.variation_example.SubstituteNameTask[SubstituteNameTask]

                              getting_started.variation_example.SubstituteNameTask --> getting_started.variation_example.SubstituteNameByJohnDoeTask
                


              click getting_started.variation_example.SubstituteNameByJohnDoeTask href "" "getting_started.variation_example.SubstituteNameByJohnDoeTask"
              click getting_started.variation_example.SubstituteNameTask href "" "getting_started.variation_example.SubstituteNameTask"
Source code in examples/getting_started/variation_example.py 
35
36
class SubstituteNameByJohnDoeTask(SubstituteNameTask):
    name = "John Doe"

name = 'John Doe' class-attribute instance-attribute

SubstituteNameByJaneDoeTask 


              flowchart TD
              getting_started.variation_example.SubstituteNameByJaneDoeTask[SubstituteNameByJaneDoeTask]
              getting_started.variation_example.SubstituteNameTask[SubstituteNameTask]

                              getting_started.variation_example.SubstituteNameTask --> getting_started.variation_example.SubstituteNameByJaneDoeTask
                


              click getting_started.variation_example.SubstituteNameByJaneDoeTask href "" "getting_started.variation_example.SubstituteNameByJaneDoeTask"
              click getting_started.variation_example.SubstituteNameTask href "" "getting_started.variation_example.SubstituteNameTask"
Source code in examples/getting_started/variation_example.py 
39
40
class SubstituteNameByJaneDoeTask(SubstituteNameTask):
    name = "Jane Doe"

name = 'Jane Doe' class-attribute instance-attribute

main()

Source code in examples/getting_started/variation_example.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
def main():
    repo = CoSyLuigiRepo(WriteTemplateTask, SubstituteNameTask)
    maestro = Maestro(repo.cls_repo, repo.taxonomy)
    results = list(maestro.query(SubstituteNameTask.target()))
    luigi.build(results, local_scheduler=True, detailed_summary=True)
    print(
        textwrap.dedent(
            f"""
                ===============================================
                    There are a total of {len(results)} results
                ==============================================="""
        )
    )