fix: clean up step instantiation

Author: mrkaye97

hatchet_sdk/v2/workflows.py

@@ -116,7 +116,6 @@ def __init__(
         self,
         fn: Callable[[Any, Context], R] | Callable[[Any, Context], Awaitable[R]],
         type: StepType,
-        workflow: Union["BaseWorkflow", None] = None,
         name: str = "",
         timeout: str = "60m",
         parents: list[str] = [],
@@ -130,7 +129,7 @@ def __init__(
     ) -> None:
         self.fn = fn
         self.is_async_function = is_async_fn(fn)
-        self.workflow = workflow
+        self.workflow: Union["BaseWorkflow", None] = None
 
         self.type = type
         self.timeout = timeout
@@ -146,7 +145,9 @@ def __init__(
 
     def call(self, ctx: Context) -> R:
         if not self.is_registered:
-            raise ValueError("Only steps that have been registered can be called.")
+            raise ValueError(
+                "Only steps that have been registered can be called. To register this step, instantiate its corresponding workflow."
+            )
 
         if self.is_async_function:
             raise TypeError(f"{self.name} is not a sync function. Use `acall` instead.")
@@ -159,7 +160,9 @@ def call(self, ctx: Context) -> R:
 
     async def acall(self, ctx: Context) -> R:
         if not self.is_registered:
-            raise ValueError("Only steps that have been registered can be called.")
+            raise ValueError(
+                "Only steps that have been registered can be called. To register this step, instantiate its corresponding workflow."
+            )
 
         if not self.is_async_function:
             raise TypeError(
@@ -207,6 +210,9 @@ class BaseWorkflow:
     def __init__(self) -> None:
         self.config.name = self.config.name or str(self.__class__.__name__)
 
+        for step in self.steps:
+            step.workflow = self
+
     def get_service_name(self, namespace: str) -> str:
         return f"{namespace}{self.config.name.lower()}"
 

hatchet_sdk/worker/worker.py

@@ -123,8 +123,6 @@ def register_workflow(self, workflow: Union["BaseWorkflow", Any]) -> None:
             sys.exit(1)
 
         for step in workflow.steps:
-            step.workflow = workflow
-
             action_name = workflow.create_action_name(namespace, step)
             self.action_registry[action_name] = step
             return_type = get_type_hints(step.fn).get("return")