feat: workflow implementation, declaration, and config

Author: mrkaye97

examples/simple/worker.py

@@ -1,17 +1,18 @@
-import time
-
 from dotenv import load_dotenv
 
 from hatchet_sdk import Context
-from hatchet_sdk.v2 import Hatchet, Workflow, WorkflowConfig
+from hatchet_sdk.v2 import BaseWorkflowImpl, Hatchet
 
 load_dotenv()
 
 hatchet = Hatchet(debug=True)
 
 
-class MyWorkflow(Workflow):
-    config = WorkflowConfig(name="foobar", on_events=["user:create"])
+class MyWorkflow(BaseWorkflowImpl):
+    declaration = hatchet.declare_workflow(
+        name="foobar",
+        on_events=["user:create"],
+    )
 
     @hatchet.step(timeout="11s", retries=3)
     def step1(self, context: Context) -> dict[str, str]:

hatchet_sdk/v2/__init__.py

@@ -1,3 +1,3 @@
 from .hatchet import Hatchet as Hatchet
-from .workflows import Workflow as Workflow
+from .workflows import BaseWorkflowImpl as BaseWorkflowImpl
 from .workflows import WorkflowConfig as WorkflowConfig

hatchet_sdk/v2/hatchet.py

@@ -1,6 +1,21 @@
 import asyncio
 import logging
-from typing import Any, Callable, Optional, TypeVar
+from enum import Enum
+from typing import (
+    Any,
+    Awaitable,
+    Callable,
+    Generic,
+    Optional,
+    ParamSpec,
+    Type,
+    TypeGuard,
+    TypeVar,
+    Union,
+    cast,
+)
+
+from pydantic import BaseModel, ConfigDict
 
 from hatchet_sdk.client import Client, new_client, new_client_raw
 from hatchet_sdk.clients.admin import AdminClient
@@ -16,7 +31,16 @@
 from hatchet_sdk.loader import ClientConfig
 from hatchet_sdk.logger import logger
 from hatchet_sdk.rate_limit import RateLimit
-from hatchet_sdk.v2.workflows import Step, StepType
+from hatchet_sdk.v2.workflows import (
+    ConcurrencyExpression,
+    EmptyModel,
+    Step,
+    StepType,
+    StickyStrategy,
+    TWorkflowInput,
+    WorkflowConfig,
+    WorkflowDeclaration,
+)
 from hatchet_sdk.worker.worker import Worker
 
 R = TypeVar("R")
@@ -192,3 +216,33 @@ def worker(
             debug=self._client.debug,
             owned_loop=loop is None,
         )
+
+    def declare_workflow(
+        self,
+        name: str = "",
+        on_events: list[str] = [],
+        on_crons: list[str] = [],
+        version: str = "",
+        timeout: str = "60m",
+        schedule_timeout: str = "5m",
+        sticky: StickyStrategy | None = None,
+        default_priority: int = 1,
+        concurrency: ConcurrencyExpression | None = None,
+        input_validator: Type[TWorkflowInput] | None = None,
+    ) -> WorkflowDeclaration[TWorkflowInput]:
+        return WorkflowDeclaration[TWorkflowInput](
+            WorkflowConfig(
+                name=name,
+                on_events=on_events,
+                on_crons=on_crons,
+                version=version,
+                timeout=timeout,
+                schedule_timeout=schedule_timeout,
+                sticky=sticky,
+                default_priority=default_priority,
+                concurrency=concurrency,
+                input_validator=input_validator
+                or cast(Type[TWorkflowInput], EmptyModel),
+            ),
+            self,
+        )

hatchet_sdk/v2/workflows.py

@@ -1,6 +1,7 @@
 import asyncio
 from enum import Enum
 from typing import (
+    TYPE_CHECKING,
     Any,
     Awaitable,
     Callable,
@@ -30,6 +31,9 @@
 from hatchet_sdk.contracts.workflows_pb2 import WorkflowConcurrencyOpts, WorkflowKind
 from hatchet_sdk.logger import logger
 
+if TYPE_CHECKING:
+    from hatchet_sdk.v2 import Hatchet
+
 R = TypeVar("R")
 P = ParamSpec("P")
 
@@ -69,6 +73,9 @@ class ConcurrencyExpression(BaseModel):
     limit_strategy: ConcurrencyLimitStrategy
 
 
+TWorkflowInput = TypeVar("TWorkflowInput", bound=BaseModel, default=EmptyModel)
+
+
 class WorkflowConfig(BaseModel):
     model_config = ConfigDict(extra="forbid", arbitrary_types_allowed=True)
 
@@ -158,8 +165,25 @@ async def acall(self, ctx: Context) -> R:
         raise TypeError(f"{self.name} is not an async function. Use `call` instead.")
 
 
-class Workflow:
-    config: WorkflowConfig = WorkflowConfig()
+class WorkflowDeclaration(Generic[TWorkflowInput]):
+    def __init__(self, config: WorkflowConfig, hatchet: "Hatchet"):
+        self.config = config
+        self.hatchet = hatchet
+
+    def run(self, input: TWorkflowInput | None = None) -> Any:
+        return self.hatchet.admin.run_workflow(
+            workflow_name=self.config.name, input=input.model_dump() if input else {}
+        )
+
+
+class BaseWorkflowImpl:
+    """
+    A Hatchet workflow implementation base. This class should be inherited by all workflow implementations.
+
+    Configuration is passed to the workflow implementation via the `config` attribute.
+    """
+
+    declaration: WorkflowDeclaration
 
     def get_service_name(self, namespace: str) -> str:
         return f"{namespace}{self.config.name.lower()}"
@@ -191,6 +215,7 @@ def create_action_name(self, namespace: str, step: Step[Any]) -> str:
         return self.get_service_name(namespace) + ":" + step.name
 
     def __init__(self) -> None:
+        self.config = self.declaration.config
         self.config.name = self.config.name or str(self.__class__.__name__)
 
     def get_name(self, namespace: str) -> str:

hatchet_sdk/worker/worker.py

@@ -36,7 +36,7 @@
 )
 
 if TYPE_CHECKING:
-    from hatchet_sdk.v2 import Workflow
+    from hatchet_sdk.v2 import BaseWorkflowImpl
 
 T = TypeVar("T")
 
@@ -110,7 +110,7 @@ def register_workflow_from_opts(
             logger.error(e)
             sys.exit(1)
 
-    def register_workflow(self, workflow: Union["Workflow", Any]) -> None:
+    def register_workflow(self, workflow: Union["BaseWorkflowImpl", Any]) -> None:
         namespace = self.client.config.namespace
 
         try: