Workflow management python SDK

dapr/aio/clients/grpc/client.py

@@ -1103,6 +1103,216 @@ async def unlock(
         return UnlockResponse(status=UnlockResponseStatus(response.status),
                               headers=await call.initial_metadata())
 
+    async def start_workflow(
+            self,
+            instance_id: str,
+            workflow_component: str,
+            workflow_name: str,
+            input: bytes,
+            workflow_options: dict) -> api_v1.StartWorkflowResponse:
+        """Starts a workflow.
+            Args:
+                instance_id (str): the name of the workflow instance,
+                                    e.g. `order_processing_workflow-103784`.
+                workflow_component (str): the name of the workflow component
+                                    that will run the workflow. e.g. `dapr`.
+                workflow_name (str): the name of the workflow that will be executed.
+                input (bytes): the input that the workflow will receive.
+                workflow_options (dict): the key-value options that the workflow will receive.
+            Returns:
+                :class:`StartWorkflowResponse`: Instance ID associated with the started workflow
+        """
+        # Warnings and input validation
+        warn('The Workflow API is an Alpha version and is subject to change.',
+             UserWarning, stacklevel=2)
+        validateNotBlankString(instance_id=instance_id,
+                               workflow_component=workflow_component,
+                               workflow_name=workflow_name)
+        # Actual start workflow invocation
+        req = api_v1.StartWorkflowRequest(
+            instance_id=instance_id,
+            workflow_component=workflow_component,
+            workflow_name=workflow_name,
+            options=workflow_options,
+            input=input)
+        response, call = self._stub.StartWorkflowAlpha1.with_call(req)
+
+        return api_v1.StartWorkflowResponse(instance_id=response.instanceID,
+                                            workflow_component=response.workflowComponent,
+                                            workflow_name=response.workflowName)
+
+    # RRL TODO: Clean up return obkect
+    async def get_workflow(
+            self,
+            instance_id: str,
+            workflow_component: str) -> api_v1.GetWorkflowResponse:
+        """Gets information on a workflow.
+
+            Args:
+                instance_id (str): the name of the workflow instance,
+                                    e.g. `order_processing_workflow-103784`.
+                workflow_component (str): the name of the workflow component
+                                    that will run the workflow. e.g. `dapr`.
+
+            Returns:
+                :class:`GetWorkflowResponse`: Status of the request,
+                    `UnlockResponseStatus.success` if it was successful of some other
+                    status otherwise.
+        """
+        # Warnings and input validation
+        warn('The Workflow API is an Alpha version and is subject to change.',
+             UserWarning, stacklevel=2)
+        validateNotBlankString(instance_id=instance_id,
+                               workflow_component=workflow_component)
+        # Actual get workflow invocation
+        req = api_v1.GetWorkflowRequest(
+            instance_id=instance_id,
+            workflow_component=workflow_component)
+        response, call = self._stub.GetWorkflowAlpha1.with_call(req)
+
+        # RRL TODO: Fix this
+        return api_v1.GetWorkflowResponse(instance_id=instance_id,
+                                          workflow_name=response.workflow_name,
+                                          created_at=response.created_at,
+                                          last_updated_at=response.last_updated_at,
+                                          runtime_status=response.runtime_status)
+
+    async def terminate_workflow(
+            self,
+            instance_id: str,
+            workflow_component: str) -> DaprResponse:
+        """Terminates a workflow.
+            Args:
+                instance_id (str): the name of the workflow instance, e.g.
+                                    `order_processing_workflow-103784`.
+                workflow_component (str): the name of the workflow component
+                                    that will run the workflow. e.g. `dapr`.
+            Returns:
+                :class:`DaprResponse` gRPC metadata returned from callee
+        """
+        # Warnings and input validation
+        warn('The Workflow API is an Alpha version and is subject to change.',
+             UserWarning, stacklevel=2)
+        validateNotBlankString(instance_id=instance_id,
+                               workflow_component=workflow_component)
+        # Actual terminate workflow invocation
+        req = api_v1.TerminateWorkflowRequest(
+            instance_id=instance_id,
+            workflow_component=workflow_component)
+        _, call = self._stub.TerminateWorkflowAlpha1.with_call(req)
+
+        return DaprResponse(
+            headers=call.initial_metadata())
+
+    # RRL TODO: Clean up return obkect comments
+    async def raise_event(
+            self,
+            instance_id: str,
+            workflow_component: str) -> DaprResponse:
+        """Raises an event on a workflow.
+            Args:
+                instance_id (str): the name of the workflow instance,
+                                    e.g. `order_processing_workflow-103784`.
+                workflow_component (str): the name of the workflow component
+                                    that will run the workflow. e.g. `dapr`.
+            Returns:
+                :class:`DaprResponse` gRPC metadata returned from callee
+        """
+        # Warnings and input validation
+        warn('The Workflow API is an Alpha version and is subject to change.',
+             UserWarning, stacklevel=2)
+        validateNotBlankString(instance_id=instance_id,
+                               workflow_component=workflow_component)
+        # Actual terminate workflow invocation
+        req = api_v1.TerminateWorkflowRequest(
+            instance_id=instance_id,
+            workflow_component=workflow_component)
+        _, call = self._stub.TerminateWorkflowAlpha1.with_call(req)
+
+        return DaprResponse(
+            headers=call.initial_metadata())
+
+    async def pause_workflow(
+            self,
+            instance_id: str,
+            workflow_component: str) -> DaprResponse:
+        """Pause a workflow.
+            Args:
+                instance_id (str): the name of the workflow instance,
+                                    e.g. `order_processing_workflow-103784`.
+                workflow_component (str): the name of the workflow component
+                                    that will run the workflow. e.g. `dapr`.
+        Returns:
+            :class:`DaprResponse` gRPC metadata returned from callee
+        """
+        # Warnings and input validation
+        warn('The Workflow API is an Alpha version and is subject to change.',
+             UserWarning, stacklevel=2)
+        validateNotBlankString(instance_id=instance_id,
+                               workflow_component=workflow_component)
+        # Actual pause workflow invocation
+        req = api_v1.PauseWorkflowRequest(
+            instance_id=instance_id,
+            workflow_component=workflow_component)
+        _, call = self._stub.PauseWorkflowAlpha1.with_call(req)
+
+        return DaprResponse(
+            headers=call.initial_metadata())
+
+    async def resume_workflow(
+            self,
+            instance_id: str,
+            workflow_component: str) -> DaprResponse:
+        """Resumes a workflow.
+            Args:
+                instance_id (str): the name of the workflow instance,
+                                    e.g. `order_processing_workflow-103784`.
+                workflow_component (str): the name of the workflow component
+                                    that will run the workflow. e.g. `dapr`.
+            Returns:
+                :class:`DaprResponse` gRPC metadata returned from callee
+        """
+        # Warnings and input validation
+        warn('The Workflow API is an Alpha version and is subject to change.',
+             UserWarning, stacklevel=2)
+        validateNotBlankString(instance_id=instance_id,
+                               workflow_component=workflow_component)
+        # Actual resume workflow invocation
+        req = api_v1.ResumeWorkflowRequest(
+            instance_id=instance_id,
+            workflow_component=workflow_component)
+        _, call = self._stub.ResumeWorkflowAlpha1.with_call(req)
+
+        return DaprResponse(
+            headers=call.initial_metadata())
+
+    async def purge_workflow(
+            self,
+            instance_id: str,
+            workflow_component: str) -> DaprResponse:
+        """Purges a workflow.
+            Args:
+                instance_id (str): the name of the workflow instance,
+                                    e.g. `order_processing_workflow-103784`.
+                workflow_component (str): the name of the workflow component
+                                    that will run the workflow. e.g. `dapr`.
+            Returns:
+                :class:`DaprResponse` gRPC metadata returned from callee
+        """
+        # Warnings and input validation
+        warn('The Workflow API is an Alpha version and is subject to change.',
+             UserWarning, stacklevel=2)
+        validateNotBlankString(instance_id=instance_id,
+                               workflow_component=workflow_component)
+        # Actual purge workflow invocation
+        req = api_v1.PurgeWorkflowRequest(
+            instance_id=instance_id,
+            workflow_component=workflow_component)
+        _, call = self._stub.PurgeWorkflowAlpha1.with_call(req)
+
+        return DaprResponse(
+            headers=call.initial_metadata())
+
     async def wait(self, timeout_s: float):
         """Waits for sidecar to be available within the timeout.
 

dapr/clients/grpc/_response.py

@@ -933,6 +933,64 @@ def extended_metadata(self) -> Dict[str, str]:
         return self._extended_metadata
 
 
+# RRL TODO: Add properties and fix init
+class GetWorkflowResponse():
+    '''The response of get_workflow operation.'''
+
+    def __init__(
+        self,
+        client: DaprGrpcClient,
+        instance_id: str,
+        workflow_name: str,
+        created_at: str,
+        last_updated_at: str,
+        runtime_status: str,
+        properties: Dict[str, str] = {},
+        headers: MetadataTuple = (),
+    ):
+        """Initializes a GetWorkflowResponse.
+
+        Args:
+            client (DaprClient): a reference to the dapr client used for the GetWorkflow request.
+            instance_id (str): the instance ID assocated with this response.
+            workflow_name (str): the name of the workflow that was started.
+            created_at (str): the time at which the workflow started executing.
+            last_updated_at (str): the time at which the workflow was last updated.
+            runtime_status (str): the current runtime status of the workflow.
+            properties (Dict[str, str]): properties sent as a reponse by the workflow.
+            headers (Tuple, optional): the headers from Dapr gRPC response.
+        """
+        super().__init__(headers)
+        self._client = client
+        self._instance_id = instance_id
+        self._workflow_name = workflow_name
+        self._created_at = created_at
+        self._last_updated_at = last_updated_at
+        self._runtime_status = runtime_status
+        self._properties = properties
+
+
+class StartWorkflowResponse():
+    '''The response of start_workflow operation.'''
+
+    def __init__(
+        self,
+        instance_id: str,
+        client: DaprGrpcClient,
+        headers: MetadataTuple = (),
+    ):
+        """Initializes a StartWorkflowResponse.
+
+        Args:
+            instance_id (str): the instance ID assocated with this response.
+            client (DaprClient): a reference to the dapr client used for the GetWorkflow request.
+            headers (Tuple, optional): the headers from Dapr gRPC response.
+        """
+        super().__init__(headers)
+        self.instance_id = instance_id
+        self._client = client
+
+
 class RegisteredComponents(NamedTuple):
     '''Describes a loaded Dapr component.'''