b2luigi_task_dependency_chain_with_json_output.py

1import b2luigi
2import json
3
4
5class TaskWithOutput(b2luigi.Task):
6    def output(self):
7        return b2luigi.LocalTarget("output.json")
8
9    def run(self):
10        with self.output().open("w") as f:
11            json.dump({"foo": "bar"}, f)
12
13
14class TaskWithParameters(b2luigi.Task):
15    some_parameter = b2luigi.Parameter()
16
17    def requires(self):
18        return TaskWithOutput()
19
20    def run(self):
21        with self.input().open("r") as f:
22            data = json.load(f)
23
24        print(f"Task with parameter {self.some_parameter} read data: {data}")
25
26
27if __name__ == "__main__":
28    b2luigi.process(TaskWithParameters(some_parameter="some_value"))