Async SDK Clients for Orkes Conductor and Conductor OSS (#320)

Author: IgorChvyrov-sm

docs/authorization/README.md

@@ -225,8 +225,10 @@ authorization_client.remove_user_from_group(group_id, user_id)
 Grants a set of accesses to the specified Subject for a given Target.
 
 ```python
-from conductor.client.http.models.target_ref import TargetRef, TargetType
-from conductor.client.http.models.subject_ref import SubjectRef, SubjectType
+from conductor.client.http.models.target_ref import TargetRef
+from conductor.shared.http.enums.target_type import TargetType
+from conductor.client.http.models.subject_ref import SubjectRef
+from conductor.shared.http.enums.subject_type import SubjectType
 from conductor.client.orkes.models.access_type import AccessType
 
 target = TargetRef(TargetType.WORKFLOW_DEF, "TEST_WORKFLOW")
@@ -245,7 +247,8 @@ Given the target, returns all permissions associated with it as a Dict[str, List
 In the returned dictionary, key is AccessType and value is a list of subjects.
 
 ```python
-from conductor.client.http.models.target_ref import TargetRef, TargetType
+from conductor.client.http.models.target_ref import TargetRef
+from conductor.shared.http.enums.target_type import TargetType
 
 target = TargetRef(TargetType.WORKFLOW_DEF, WORKFLOW_NAME)
 target_permissions = authorization_client.get_permissions(target)
@@ -273,8 +276,10 @@ user_permissions = authorization_client.get_granted_permissions_for_user(user_id
 Removes a set of accesses from a specified Subject for a given Target.
 
 ```python
-from conductor.client.http.models.target_ref import TargetRef, TargetType
-from conductor.client.http.models.subject_ref import SubjectRef, SubjectType
+from conductor.client.http.models.target_ref import TargetRef
+from conductor.shared.http.enums.target_type import TargetType
+from conductor.client.http.models.subject_ref import SubjectRef
+from conductor.shared.http.enums.subject_type import SubjectType
 from conductor.client.orkes.models.access_type import AccessType
 
 target = TargetRef(TargetType.WORKFLOW_DEF, "TEST_WORKFLOW")

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(

src/__init__.py examples/async/__init__.pysrc/__init__.py renamed to examples/async/__init__.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())