hatchet_sdk_basic_workflow_with_sequential_steps.py

1from hatchet_sdk import Hatchet
2from dotenv import load_dotenv
3
4load_dotenv()
5
6hatchet = Hatchet()
7
8@hatchet.workflow(on_events=["user:create"])
9class MyWorkflow:
10    @hatchet.step()
11    def step1(self, context):
12        print("started step1")
13        return {
14            "step1": "completed"
15        }
16
17    @hatchet.step(parents=["step1"])
18    def step2(self, context):
19        print("started step2")
20        return {
21            "step2": "completed"
22        }
23
24workflow = MyWorkflow()
25
26if __name__ == "__main__":
27    worker = hatchet.worker("example-worker")
28    worker.register_workflow(workflow)
29    worker.start()