mirror of
https://github.com/hatchet-dev/hatchet.git
synced 2026-01-08 09:49:41 -06:00
d01736c15a
* feat(py-sdk): add put_workflow method and extend from base class * feat: add cron_input to create workflow API
47 lines
1.1 KiB
Python
47 lines
1.1 KiB
Python
import json
|
|
from hatchet_sdk import Hatchet, Context, CreateWorkflowVersionOpts
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
hatchet = Hatchet(debug=True)
|
|
|
|
@hatchet.workflow(on_events=["user:create"])
|
|
class MyWorkflow:
|
|
def __init__(self):
|
|
self.my_value = "test"
|
|
|
|
@hatchet.step()
|
|
def step1(self, context: Context):
|
|
test = context.playground("test", "test")
|
|
test2 = context.playground("test2", 100)
|
|
test3 = context.playground("test3", None)
|
|
|
|
print(test)
|
|
print(test2)
|
|
|
|
print("executed step1")
|
|
pass
|
|
|
|
@hatchet.step(parents=["step1"],timeout='4s')
|
|
def step2(self, context):
|
|
print("started step2")
|
|
context.sleep(1)
|
|
print("finished step2")
|
|
|
|
workflow = MyWorkflow()
|
|
worker = hatchet.worker('test-worker', max_runs=4)
|
|
worker.register_workflow(workflow)
|
|
|
|
# workflow1 = hatchet.client.admin.put_workflow(
|
|
# "workflow-copy-2",
|
|
# MyWorkflow(),
|
|
# overrides=CreateWorkflowVersionOpts(
|
|
# cron_triggers=["* * * * *"],
|
|
# cron_input=json.dumps({"test": "test"}),
|
|
# ),
|
|
# )
|
|
|
|
# print(workflow1)
|
|
|
|
worker.start() |