Feat: V2 Workflows and Steps

conftest.py

@@ -1,9 +1,10 @@
 import logging
+import os
 import subprocess
 import time
 from io import BytesIO
 from threading import Thread
-from typing import AsyncGenerator, Callable, cast
+from typing import AsyncGenerator, Callable, Generator, cast
 
 import psutil
 import pytest
@@ -23,7 +24,9 @@ def hatchet() -> Hatchet:
 
 
 @pytest.fixture()
-def worker(request: pytest.FixtureRequest):
+def worker(
+    request: pytest.FixtureRequest,
+) -> Generator[subprocess.Popen[bytes], None, None]:
     example = cast(str, request.param)
 
     command = ["poetry", "run", example]

examples/affinity-workers/event.py

@@ -1,5 +1,6 @@
 from dotenv import load_dotenv
 
+from hatchet_sdk.clients.events import PushEventOptions
 from hatchet_sdk.hatchet import Hatchet
 
 load_dotenv()
@@ -9,5 +10,5 @@
 hatchet.event.push(
     "affinity:run",
     {"test": "test"},
-    options={"additional_metadata": {"hello": "moon"}},
+    options=PushEventOptions(additional_metadata={"hello": "moon"}),
 )

examples/affinity-workers/worker.py

@@ -1,6 +1,7 @@
 from dotenv import load_dotenv
 
 from hatchet_sdk import Context, Hatchet, WorkerLabelComparator
+from hatchet_sdk.labels import DesiredWorkerLabel
 
 load_dotenv()
 
@@ -11,12 +12,12 @@
 class AffinityWorkflow:
     @hatchet.step(
         desired_worker_labels={
-            "model": {"value": "fancy-ai-model-v2", "weight": 10},
-            "memory": {
-                "value": 256,
-                "required": True,
-                "comparator": WorkerLabelComparator.LESS_THAN,
-            },
+            "model": DesiredWorkerLabel(value="fancy-ai-model-v2", weight=10),
+            "memory": DesiredWorkerLabel(
+                value=256,
+                required=True,
+                comparator=WorkerLabelComparator.LESS_THAN,
+            ),
         },
     )
     async def step(self, context: Context) -> dict[str, str | None]:

examples/blocked_async/event.py

@@ -6,7 +6,8 @@
 
 client = new_client()
 
-# client.event.push("user:create", {"test": "test"})
 client.event.push(
-    "user:create", {"test": "test"}, options={"additional_metadata": {"hello": "moon"}}
+    "user:create",
+    {"test": "test"},
+    options=PushEventOptions(additional_metadata={"hello": "moon"}),
 )

examples/bulk_fanout/bulk_trigger.py

@@ -7,7 +7,7 @@
 from dotenv import load_dotenv
 
 from hatchet_sdk import new_client
-from hatchet_sdk.clients.admin import TriggerWorkflowOptions
+from hatchet_sdk.clients.admin import TriggerWorkflowOptions, WorkflowRunDict
 from hatchet_sdk.clients.rest.models.workflow_run import WorkflowRun
 from hatchet_sdk.clients.run_event_listener import StepRunEventType
 
@@ -16,25 +16,22 @@ async def main() -> None:
     load_dotenv()
     hatchet = new_client()
 
-    workflowRuns: list[dict[str, Any]] = []
-
-    # we are going to run the BulkParent workflow 20 which will trigger the Child workflows n times for each n in range(20)
-    for i in range(20):
-        workflowRuns.append(
-            {
-                "workflow_name": "BulkParent",
-                "input": {"n": i},
-                "options": {
-                    "additional_metadata": {
-                        "bulk-trigger": i,
-                        "hello-{i}": "earth-{i}",
-                    },
-                },
-            }
+    workflow_runs = [
+        WorkflowRunDict(
+            workflow_name="BulkParent",
+            input={"n": i},
+            options=TriggerWorkflowOptions(
+                additional_metadata={
+                    "bulk-trigger": i,
+                    "hello-{i}": "earth-{i}",
+                }
+            ),
         )
+        for i in range(20)
+    ]
 
     workflowRunRefs = hatchet.admin.run_workflows(
-        workflowRuns,
+        workflow_runs,
     )
 
     results = await asyncio.gather(

examples/bulk_fanout/stream.py

@@ -6,10 +6,9 @@
 
 from dotenv import load_dotenv
 
-from hatchet_sdk import new_client
+from hatchet_sdk import Hatchet, new_client
 from hatchet_sdk.clients.admin import TriggerWorkflowOptions
 from hatchet_sdk.clients.run_event_listener import StepRunEventType
-from hatchet_sdk.v2.hatchet import Hatchet
 
 
 async def main() -> None:
@@ -31,7 +30,7 @@ async def main() -> None:
     workflowRun = hatchet.admin.run_workflow(
         "Parent",
         {"n": 2},
-        options={"additional_metadata": {streamKey: streamVal}},
+        options=TriggerWorkflowOptions(additional_metadata={streamKey: streamVal}),
     )
 
     # Stream all events for the additional meta key value

examples/bulk_fanout/trigger.py

@@ -7,6 +7,7 @@
 
 from hatchet_sdk import new_client
 from hatchet_sdk.clients.admin import TriggerWorkflowOptions
+from hatchet_sdk.clients.events import PushEventOptions
 from hatchet_sdk.clients.rest.models.workflow_run import WorkflowRun
 from hatchet_sdk.clients.run_event_listener import StepRunEventType
 
@@ -15,10 +16,10 @@ async def main() -> None:
     load_dotenv()
     hatchet = new_client()
 
-    workflowRuns: WorkflowRun = []  # type: ignore[assignment]
-
-    event = hatchet.event.push(
-        "parent:create", {"n": 999}, {"additional_metadata": {"no-dedupe": "world"}}
+    hatchet.event.push(
+        "parent:create",
+        {"n": 999},
+        PushEventOptions(additional_metadata={"no-dedupe": "world"}),
     )
 
 

examples/bulk_fanout/worker.py

@@ -4,7 +4,7 @@
 from dotenv import load_dotenv
 
 from hatchet_sdk import Context, Hatchet
-from hatchet_sdk.clients.admin import ChildWorkflowRunDict
+from hatchet_sdk.clients.admin import ChildTriggerWorkflowOptions, ChildWorkflowRunDict
 
 load_dotenv()
 
@@ -22,18 +22,17 @@ async def spawn(self, context: Context) -> dict[str, list[Any]]:
 
         n = context.workflow_input().get("n", 100)
 
-        child_workflow_runs: list[ChildWorkflowRunDict] = []
-
-        for i in range(n):
-
-            child_workflow_runs.append(
-                {
-                    "workflow_name": "BulkChild",
-                    "input": {"a": str(i)},
-                    "key": f"child{i}",
-                    "options": {"additional_metadata": {"hello": "earth"}},
-                }
+        child_workflow_runs = [
+            ChildWorkflowRunDict(
+                workflow_name="BulkChild",
+                input={"a": str(i)},
+                key=f"child{i}",
+                options=ChildTriggerWorkflowOptions(
+                    additional_metadata={"hello": "earth"}
+                ),
             )
+            for i in range(n)
+        ]
 
         if len(child_workflow_runs) == 0:
             return {}

examples/dedupe/worker.py

@@ -4,7 +4,7 @@
 
 from dotenv import load_dotenv
 
-from hatchet_sdk import Context, Hatchet
+from hatchet_sdk import ChildTriggerWorkflowOptions, Context, Hatchet
 from hatchet_sdk.clients.admin import DedupeViolationErr
 from hatchet_sdk.loader import ClientConfig
 
@@ -29,7 +29,9 @@ async def spawn(self, context: Context) -> dict[str, list[Any]]:
                             "DedupeChild",
                             {"a": str(i)},
                             key=f"child{i}",
-                            options={"additional_metadata": {"dedupe": "test"}},
+                            options=ChildTriggerWorkflowOptions(
+                                additional_metadata={"dedupe": "test"}
+                            ),
                         )
                     ).result()
                 )