Merge pull request #31 from hatchet-dev/belanger/fix-admin-aio

fix: admin aio client config

hatchet_sdk/clients/admin.py

@@ -122,6 +122,7 @@ def _prepare_schedule_workflow_request(
 class AdminClientAioImpl(AdminClientBase):
     def __init__(self, config: ClientConfig):
         aio_conn = new_conn(config, True)
+        self.config = config
         self.aio_client = WorkflowServiceStub(aio_conn)
         self.token = config.token
         self.listener_client = new_listener(config)

hatchet_sdk/context.py

@@ -48,24 +48,23 @@ def __init__(
         admin_client: AdminClientImpl,
         event_client: EventClientImpl,
         workflow_listener: PooledWorkflowRunListener,
+        workflow_run_event_listener: RunEventListenerClient,
         namespace: str = "",
     ):
         self.action = action
         self.dispatcher_client = dispatcher_client
         self.admin_client = admin_client
         self.event_client = event_client
         self.workflow_listener = workflow_listener
+        self.workflow_run_event_listener = workflow_run_event_listener
         self.namespace = namespace
         self.spawn_index = -1
 
     async def spawn_workflow(
         self, workflow_name: str, input: dict = {}, key: str = None
-    ):
+    ) -> WorkflowRunRef:
         options = self._prepare_workflow_options(key)
-        child_workflow_run_id = await self.admin_client.aio.run_workflow(
-            workflow_name, input, options
-        )
-        return WorkflowRunRef(child_workflow_run_id, self.workflow_listener)
+        return await self.admin_client.aio.run_workflow(workflow_name, input, options)
 
 
 class Context(BaseContext):
@@ -78,6 +77,7 @@ def __init__(
         admin_client: AdminClientImpl,
         event_client: EventClientImpl,
         workflow_listener: PooledWorkflowRunListener,
+        workflow_run_event_listener: RunEventListenerClient,
         namespace: str = "",
     ):
         self.aio = ContextAioImpl(
@@ -86,6 +86,7 @@ def __init__(
             admin_client,
             event_client,
             workflow_listener,
+            workflow_run_event_listener,
             namespace,
         )
 
@@ -110,6 +111,7 @@ def __init__(
         self.admin_client = admin_client
         self.event_client = event_client
         self.workflow_listener = workflow_listener
+        self.workflow_run_event_listener = workflow_run_event_listener
         self.namespace = namespace
 
         # FIXME: this limits the number of concurrent log requests to 1, which means we can do about
@@ -171,10 +173,8 @@ def spawn_workflow(self, workflow_name: str, input: dict = {}, key: str = None):
         workflow_name = f"{self.namespace}{workflow_name}"
 
         options = self._prepare_workflow_options(key)
-        child_workflow_run_id = self.admin_client.run_workflow(
-            workflow_name, input, options
-        )
-        return WorkflowRunRef(child_workflow_run_id, self.workflow_listener)
+
+        return self.admin_client.run_workflow(workflow_name, input, options)
 
     def _log(self, line: str):
         try:

hatchet_sdk/worker.py

@@ -16,6 +16,7 @@
 from google.protobuf.timestamp_pb2 import Timestamp
 
 from hatchet_sdk.clients.admin import new_admin
+from hatchet_sdk.clients.run_event_listener import new_listener
 from hatchet_sdk.clients.workflow_listener import PooledWorkflowRunListener
 from hatchet_sdk.loader import ClientConfig
 
@@ -153,6 +154,7 @@ async def handle_start_step_run(self, action: Action):
             self.admin_client,
             self.client.event,
             self.client.workflow_listener,
+            self.workflow_run_event_listener,
             self.client.config.namespace,
         )
         self.contexts[action.step_run_id] = context
@@ -193,6 +195,7 @@ async def handle_start_group_key_run(self, action: Action):
             self.admin_client,
             self.client.event,
             self.client.workflow_listener,
+            self.workflow_run_event_listener,
             self.client.config.namespace,
         )
 
@@ -482,6 +485,7 @@ async def async_start(self, retry_count=1):
             # otherwise the grpc.aio methods will use a different event loop and we'll get a bunch of errors.
             self.dispatcher_client = new_dispatcher(self.config)
             self.admin_client = new_admin(self.config)
+            self.workflow_run_event_listener = new_listener(self.config)
             self.client.workflow_listener = PooledWorkflowRunListener(self.config)
 
             self.listener: ActionListenerImpl = (

hatchet_sdk/workflow_run.py

@@ -24,5 +24,5 @@ def __str__(self):
     def stream(self) -> RunEventListener:
         return self.workflow_run_event_listener.stream(self.workflow_run_id)
 
-    async def result(self):
-        return await self.workflow_listener.result(self.workflow_run_id)
+    def result(self):
+        return self.workflow_listener.result(self.workflow_run_id)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "hatchet-sdk"
-version = "0.25.1"
+version = "0.25.2"
 description = ""
 authors = ["Alexander Belanger <[email protected]>"]
 readme = "README.md"