wip

examples/simple/worker.py

@@ -12,11 +12,14 @@
 
 @hatchet.workflow(on_events=["user:create"])
 class MyWorkflow:
-    @hatchet.step(timeout="2s", retries=3)
+    @hatchet.step(timeout="9s", retries=3)
     def step1(self, context: Context):
         print("executed step1")
-        time.sleep(10)
-        pass
+        time.sleep(2)
+        raise Exception("test")
+        return {
+            "step1": "step1",
+        }
 
 
 def main():

hatchet_sdk/client.py

@@ -6,6 +6,7 @@
 from hatchet_sdk.clients.run_event_listener import RunEventListenerClient
 from hatchet_sdk.clients.workflow_listener import PooledWorkflowRunListener
 from hatchet_sdk.connection import new_conn
+from hatchet_sdk.logger import logger
 
 from .clients.admin import AdminClient, new_admin
 from .clients.dispatcher import DispatcherClient, new_dispatcher
@@ -28,7 +29,7 @@ def from_environment(
         cls,
         defaults: ClientConfig = ClientConfig(),
         debug: bool = False,
-        *opts_functions: Callable[[ClientConfig], None]
+        *opts_functions: Callable[[ClientConfig], None],
     ):
         config: ClientConfig = ConfigLoader(".").load_client_config(defaults)
         for opt_function in opts_functions:

hatchet_sdk/clients/dispatcher.py

@@ -4,9 +4,10 @@
 import threading
 import time
 from dataclasses import dataclass, field
-from typing import Any, AsyncGenerator, List, Optional
+from typing import Any, AsyncGenerator, List, Optional, overload
 
 import grpc
+from google.protobuf.timestamp_pb2 import Timestamp
 from grpc._cython import cygrpc
 
 from hatchet_sdk.clients.event_ts import Event_ts, read_with_interrupt
@@ -16,11 +17,13 @@
     ActionType,
     AssignedAction,
     GroupKeyActionEvent,
+    GroupKeyActionEventType,
     HeartbeatRequest,
     OverridesData,
     RefreshTimeoutRequest,
     ReleaseSlotRequest,
     StepActionEvent,
+    StepActionEventType,
     WorkerListenRequest,
     WorkerRegisterRequest,
     WorkerRegisterResponse,
@@ -356,12 +359,52 @@ async def send_step_action_event(self, in_: StepActionEvent):
             metadata=get_metadata(self.token),
         )
 
+    async def send_step_action_event_simple(
+        self, action: Action, event_type: StepActionEventType, payload: str
+    ):
+        eventTimestamp = Timestamp()
+        eventTimestamp.GetCurrentTime()
+
+        event = StepActionEvent(
+            workerId=action.worker_id,
+            jobId=action.job_id,
+            jobRunId=action.job_run_id,
+            stepId=action.step_id,
+            stepRunId=action.step_run_id,
+            actionId=action.action_id,
+            eventTimestamp=eventTimestamp,
+            eventType=event_type,
+            eventPayload=payload,
+        )
+
+        return await self.send_step_action_event(event)
+
     async def send_group_key_action_event(self, in_: GroupKeyActionEvent):
         await self.aio_client.SendGroupKeyActionEvent(
             in_,
             metadata=get_metadata(self.token),
         )
 
+    async def send_group_key_action_event_simple(
+        self, action: Action, event_type: GroupKeyActionEventType, payload: str
+    ):
+        eventTimestamp = Timestamp()
+        eventTimestamp.GetCurrentTime()
+
+        event = GroupKeyActionEvent(
+            workerId=action.worker_id,
+            jobId=action.job_id,
+            jobRunId=action.job_run_id,
+            stepId=action.step_id,
+            stepRunId=action.step_run_id,
+            actionId=action.action_id,
+            eventTimestamp=eventTimestamp,
+            eventType=event_type,
+            eventPayload=payload,
+        )
+
+        return await self.send_group_key_action_event(event)
+
     def put_overrides_data(self, data: OverridesData):
         response: ActionEventResponse = self.client.PutOverridesData(
             data,

hatchet_sdk/logger.py

@@ -6,7 +6,7 @@
 logger.setLevel(logging.ERROR)
 
 handler = logging.StreamHandler(sys.stdout)
-formatter = logging.Formatter("[%(levelname)s] 🪓 -- %(asctime)s - %(message)s")
+formatter = logging.Formatter("[%(levelname)s]\t🪓 -- %(asctime)s - %(message)s")
 handler.setFormatter(formatter)
 logger.addHandler(handler)