Async SDK Clients for Orkes Conductor

docs/metadata/README.md

@@ -8,7 +8,7 @@ In order to define a workflow, you must provide a `MetadataClient` and a `Workfl
 
 ```python
 from conductor.client.configuration.configuration import Configuration
-from conductor.client.configuration.settings.authentication_settings import AuthenticationSettings
+from conductor.shared.configuration.settings.authentication_settings import AuthenticationSettings
 from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClie
 from conductor.client.workflow.conductor_workflow import ConductorWorkflow
 from conductor.client.workflow.executor.workflow_executor import WorkflowExecutor

docs/schedule/README.md

@@ -3,9 +3,10 @@
 ## Scheduler Client
 
 ### Initialization
+
 ```python
 from conductor.client.configuration.configuration import Configuration
-from conductor.client.configuration.settings.authentication_settings import AuthenticationSettings
+from conductor.shared.configuration.settings.authentication_settings import AuthenticationSettings
 from conductor.client.orkes.orkes_scheduler_client import OrkesSchedulerClient
 
 configuration = Configuration(

docs/secret/README.md

@@ -3,9 +3,10 @@
 ## Secret Client
 
 ### Initialization
+
 ```python
 from conductor.client.configuration.configuration import Configuration
-from conductor.client.configuration.settings.authentication_settings import AuthenticationSettings
+from conductor.shared.configuration.settings.authentication_settings import AuthenticationSettings
 from conductor.client.orkes.orkes_secret_client import OrkesSecretClient
 
 configuration = Configuration(

docs/task/README.md

@@ -3,9 +3,10 @@
 ## Task Client
 
 ### Initialization
+
 ```python
 from conductor.client.configuration.configuration import Configuration
-from conductor.client.configuration.settings.authentication_settings import AuthenticationSettings
+from conductor.shared.configuration.settings.authentication_settings import AuthenticationSettings
 from conductor.client.orkes.orkes_task_client import OrkesTaskClient
 
 configuration = Configuration(

docs/testing/README.md

@@ -14,7 +14,7 @@ A sample unit test code snippet is provided below.
 
 ```python
 import json
-from conductor.client.configuration.settings.authentication_settings import AuthenticationSettings
+from conductor.shared.configuration.settings.authentication_settings import AuthenticationSettings
 from conductor.client.configuration.configuration import Configuration
 from conductor.client.http.models.workflow_test_request import WorkflowTestRequest
 from conductor.client.orkes.orkes_workflow_client import OrkesWorkflowClient

docs/worker/README.md

@@ -38,7 +38,8 @@ Quick example below:
 
 ```python
 from conductor.client.http.models import Task, TaskResult
-from conductor.client.http.models.task_result_status import TaskResultStatus
+from conductor.shared.http.enums import TaskResultStatus
+
 
 def execute(task: Task) -> TaskResult:
     task_result = TaskResult(
@@ -59,7 +60,7 @@ The class must implement `WorkerInterface` class, which requires an `execute` me
 
 ```python
 from conductor.client.http.models import Task, TaskResult
-from conductor.client.http.models.task_result_status import TaskResultStatus
+from conductor.shared.http.enums import TaskResultStatus
 from conductor.client.worker.worker_interface import WorkerInterface
 
 class SimplePythonWorker(WorkerInterface):
@@ -99,13 +100,14 @@ def python_annotated_task(input) -> object:
 Now you can run your workers by calling a `TaskHandler`, example:
 
 ```python
-from conductor.client.configuration.settings.authentication_settings import AuthenticationSettings
+from conductor.shared.configuration.settings.authentication_settings import AuthenticationSettings
 from conductor.client.configuration.configuration import Configuration
 from conductor.client.automator.task_handler import TaskHandler
 from conductor.client.worker.worker import Worker
 
 #### Add these lines if running on a mac####
 from multiprocessing import set_start_method
+
 set_start_method('fork')
 ############################################
 
@@ -347,7 +349,7 @@ and [simple_cpp_worker.py](src/example/worker/cpp/simple_cpp_worker.py) for comp
 ```python
 from conductor.client.http.models.task import Task
 from conductor.client.http.models.task_result import TaskResult
-from conductor.client.http.models.task_result_status import TaskResultStatus
+from conductor.shared.http.enums import TaskResultStatus
 from conductor.client.worker.worker_interface import WorkerInterface
 from ctypes import cdll
 

docs/workflow/README.md

@@ -3,9 +3,10 @@
 ## Workflow Client
 
 ### Initialization
+
 ```python
 from conductor.client.configuration.configuration import Configuration
-from conductor.client.configuration.settings.authentication_settings import AuthenticationSettings
+from conductor.shared.configuration.settings.authentication_settings import AuthenticationSettings
 from conductor.client.orkes.orkes_workflow_client import OrkesWorkflowClient
 
 configuration = Configuration(

examples/async/dynamic_workflow.py

@@ -0,0 +1,70 @@
+"""
+This is a dynamic workflow that can be created and executed at run time.
+dynamic_workflow will run worker tasks get_user_email and send_email in the same order.
+For use cases in which the workflow cannot be defined statically, dynamic workflows is a useful approach.
+For detailed explanation, https://github.com/conductor-sdk/conductor-python/blob/main/workflows.md
+"""
+
+import asyncio
+
+from conductor.asyncio_client.automator.task_handler import TaskHandler
+from conductor.asyncio_client.configuration.configuration import Configuration
+from conductor.asyncio_client.adapters import ApiClient
+from conductor.asyncio_client.orkes.orkes_clients import OrkesClients
+from conductor.asyncio_client.worker.worker_task import worker_task
+from conductor.asyncio_client.workflow.conductor_workflow import AsyncConductorWorkflow
+
+
+@worker_task(task_definition_name="get_user_email")
+def get_user_email(userid: str) -> str:
+    return f"{userid}@example.com"
+
+
+@worker_task(task_definition_name="send_email")
+def send_email(email: str, subject: str, body: str):
+    print(f"sending email to {email} with subject {subject} and body {body}")
+
+
+async def main():
+    # defaults to reading the configuration using following env variables
+    # CONDUCTOR_SERVER_URL : conductor server e.g. https://play.orkes.io/api
+    # CONDUCTOR_AUTH_KEY : API Authentication Key
+    # CONDUCTOR_AUTH_SECRET: API Auth Secret
+    api_config = Configuration()
+    task_handler = TaskHandler(configuration=api_config)
+    task_handler.start_processes()
+
+    async with ApiClient(api_config) as api_client:
+        clients = OrkesClients(api_client=api_client, configuration=api_config)
+        workflow_executor = clients.get_workflow_executor()
+        workflow = AsyncConductorWorkflow(
+            name="dynamic_workflow", version=1, executor=workflow_executor
+        )
+        get_email = get_user_email(
+            task_ref_name="get_user_email_ref", userid=workflow.input("userid")
+        )
+        sendmail = send_email(
+            task_ref_name="send_email_ref",
+            email=get_email.output("result"),
+            subject="Hello from Orkes",
+            body="Test Email",
+        )
+
+        workflow >> get_email >> sendmail
+
+        # Configure the output of the workflow
+        workflow.output_parameters(
+            output_parameters={"email": get_email.output("result")}
+        )
+
+        workflow_run = await workflow.execute(workflow_input={"userid": "user_a"})
+        print(f"\nworkflow output:  {workflow_run.output}\n")
+        print(
+            f"check the workflow execution here: {api_config.ui_host}/execution/{workflow_run.workflow_id}"
+        )
+
+    task_handler.stop_processes()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/async/helloworld/greetings_worker.py

@@ -0,0 +1,11 @@
+"""
+This file contains a Simple Worker that can be used in any workflow.
+For detailed information https://github.com/conductor-sdk/conductor-python/blob/main/README.md#step-2-write-worker
+"""
+
+from conductor.asyncio_client.worker.worker_task import worker_task
+
+
+@worker_task(task_definition_name="greet")
+def greet(name: str) -> str:
+    return f"Hello {name}"

examples/async/helloworld/greetings_workflow.json

@@ -0,0 +1,17 @@
+{
+  "name": "greetings",
+  "description": "Sample greetings workflow",
+  "version": 1,
+  "tasks": [
+    {
+      "name": "greet",
+      "taskReferenceName": "greet_ref",
+      "type": "SIMPLE",
+      "inputParameters": {
+        "name": "${workflow.input.name}"
+      }
+    }
+  ],
+  "timeoutPolicy": "TIME_OUT_WF",
+  "timeoutSeconds": 60
+}

examples/async/helloworld/greetings_workflow.py

@@ -0,0 +1,20 @@
+"""
+For detailed explanation https://github.com/conductor-sdk/conductor-python/blob/main/README.md#step-1-create-a-workflow
+"""
+
+from greetings_worker import greet
+
+from conductor.asyncio_client.workflow.conductor_workflow import AsyncConductorWorkflow
+from conductor.asyncio_client.workflow.executor.workflow_executor import (
+    AsyncWorkflowExecutor,
+)
+
+
+def greetings_workflow(
+    workflow_executor: AsyncWorkflowExecutor,
+) -> AsyncConductorWorkflow:
+    name = "greetings"
+    workflow = AsyncConductorWorkflow(name=name, executor=workflow_executor)
+    workflow.version = 1
+    workflow >> greet(task_ref_name="greet_ref", name=workflow.input("name"))
+    return workflow

examples/async/helloworld/helloworld.py

@@ -0,0 +1,50 @@
+import asyncio
+
+from greetings_workflow import greetings_workflow
+
+from conductor.asyncio_client.automator.task_handler import TaskHandler
+from conductor.asyncio_client.configuration import Configuration
+from conductor.asyncio_client.adapters import ApiClient
+from conductor.asyncio_client.workflow.conductor_workflow import AsyncConductorWorkflow
+from conductor.asyncio_client.workflow.executor.workflow_executor import (
+    AsyncWorkflowExecutor,
+)
+
+
+async def register_workflow(
+    workflow_executor: AsyncWorkflowExecutor,
+) -> AsyncConductorWorkflow:
+    workflow = greetings_workflow(workflow_executor=workflow_executor)
+    await workflow.register(True)
+    return workflow
+
+
+async def main():
+    # points to http://localhost:8080/api by default
+    api_config = Configuration()
+    async with ApiClient(api_config) as api_client:
+        workflow_executor = AsyncWorkflowExecutor(
+            configuration=api_config, api_client=api_client
+        )
+        # Needs to be done only when registering a workflow one-time
+        workflow = await register_workflow(workflow_executor)
+
+        task_handler = TaskHandler(configuration=api_config)
+        task_handler.start_processes()
+
+        workflow_run = await workflow_executor.execute(
+            name=workflow.name,
+            version=workflow.version,
+            workflow_input={"name": "World"},
+        )
+
+    print(f"\nworkflow result: {workflow_run.output}\n")
+    print(
+        f"see the workflow execution here: {api_config.ui_host}/execution/{workflow_run.workflow_id}\n"
+    )
+
+    task_handler.stop_processes()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())