Exceptions module and temporalite support

Author: cretz

.github/workflows/ci.yml

@@ -27,7 +27,12 @@ jobs:
       - uses: actions/setup-python@v1
         with:
           python-version: ${{ matrix.python }}
+      # Needed to tests since they use external server
+      - uses: actions/setup-go@v2
+        with:
+          go-version: "1.17"
       - run: python -m pip install --upgrade wheel poetry poethepoet
       - run: poetry install
       - run: poe lint
       - run: poe build
+      - run: poe test

docs/api.rst

@@ -16,3 +16,9 @@ Converters
 
 .. automodule:: temporalio.converter
    :members:
+
+Exceptions
+----------
+
+.. automodule:: temporalio.exceptions
+   :members:

docs/conf.py

@@ -13,7 +13,7 @@
 import os
 import sys
 
-sys.path.insert(0, os.path.abspath("../"))
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
 
 
 # -- Project information -----------------------------------------------------

poetry.lock

@@ -24,6 +24,7 @@ isort = "^5.10.1"
 maturin = "^0.12.6"
 mypy = "^0.931"
 mypy-protobuf = "^3.2.0"
+psutil = "^5.9.0"
 pydocstyle = "^6.1.1"
 pytest = "^6.2.5"
 pytest-asyncio = "^0.18.0"

pyproject.toml

@@ -83,7 +83,7 @@ def details(self) -> bytes:
 class Client:
     @staticmethod
     async def connect(opts: ClientOptions) -> "Client":
-        return Client(await temporal_sdk_bridge.new_client(opts))
+        return Client(await temporal_sdk_bridge.connect_client(opts))
 
     _ref: temporal_sdk_bridge.ClientRef
 

temporalio/bridge/client.py

@@ -9,7 +9,7 @@ use tonic;
 fn temporal_sdk_bridge(py: Python, m: &PyModule) -> PyResult<()> {
     m.add("RPCError", py.get_type::<RPCError>())?;
     m.add_class::<ClientRef>()?;
-    m.add_function(wrap_pyfunction!(new_client, m)?)?;
+    m.add_function(wrap_pyfunction!(connect_client, m)?)?;
     Ok(())
 }
 
@@ -55,7 +55,7 @@ pub struct ClientRetryConfig {
 }
 
 #[pyfunction]
-fn new_client(py: Python, opts: ClientOptions) -> PyResult<&PyAny> {
+fn connect_client(py: Python, opts: ClientOptions) -> PyResult<&PyAny> {
     // TODO(cretz): Add metrics_meter?
     let opts: temporal_client::ServerGatewayOptions = opts.try_into()?;
     pyo3_asyncio::tokio::future_into_py(py, async move {

temporalio/bridge/src/lib.rs

@@ -17,7 +17,7 @@
 import temporalio.api.workflowservice.v1
 import temporalio.common
 import temporalio.converter
-import temporalio.failure
+import temporalio.exceptions
 import temporalio.workflow_service
 from temporalio.workflow_service import RPCError, RPCStatusCode
 
@@ -396,10 +396,8 @@ async def result(
                     req.next_page_token = b""
                     continue
                 # Ignoring anything after the first response like TypeScript
-                if not complete_attr.result:
-                    return cast(T, None)
-                results = await self._client.data_converter.decode(
-                    complete_attr.result.payloads
+                results = await temporalio.converter.decode_payloads(
+                    complete_attr.result, self._client.data_converter
                 )
                 if not results:
                     return cast(T, None)
@@ -414,38 +412,35 @@ async def result(
                     req.next_page_token = b""
                     continue
                 raise WorkflowFailureError(
-                    cause=await temporalio.failure.FailureError.from_proto(
+                    cause=await temporalio.exceptions.failure_to_error(
                         fail_attr.failure, self._client.data_converter
-                    )
+                    ),
                 )
             elif event.HasField("workflow_execution_canceled_event_attributes"):
                 cancel_attr = event.workflow_execution_canceled_event_attributes
-                details = []
-                if cancel_attr.details and cancel_attr.details.payloads:
-                    details = await self._client.data_converter.decode(
-                        cancel_attr.details.payloads
-                    )
                 raise WorkflowFailureError(
-                    cause=temporalio.failure.FailureError(
+                    cause=temporalio.exceptions.CancelledError(
                         "Workflow cancelled",
-                        temporalio.failure.CancelledFailure(*details),
+                        *(
+                            await temporalio.converter.decode_payloads(
+                                cancel_attr.details.payloads,
+                                self._client.data_converter,
+                            )
+                        ),
                     )
                 )
             elif event.HasField("workflow_execution_terminated_event_attributes"):
                 term_attr = event.workflow_execution_terminated_event_attributes
-                details = []
-                if term_attr.details and term_attr.details.payloads:
-                    details = await self._client.data_converter.decode(
-                        term_attr.details.payloads
-                    )
                 raise WorkflowFailureError(
-                    cause=temporalio.failure.FailureError(
+                    cause=temporalio.exceptions.TerminatedError(
                         term_attr.reason or "Workflow terminated",
-                        temporalio.failure.TerminatedFailure(
-                            *details,
-                            reason=term_attr.reason or None,
+                        *(
+                            await temporalio.converter.decode_payloads(
+                                term_attr.details.payloads,
+                                self._client.data_converter,
+                            )
                         ),
-                    )
+                    ),
                 )
             elif event.HasField("workflow_execution_timed_out_event_attributes"):
                 time_attr = event.workflow_execution_timed_out_event_attributes
@@ -455,12 +450,10 @@ async def result(
                     req.next_page_token = b""
                     continue
                 raise WorkflowFailureError(
-                    cause=temporalio.failure.FailureError(
+                    cause=temporalio.exceptions.TimeoutError(
                         "Workflow timed out",
-                        temporalio.failure.TimeoutFailure(
-                            temporalio.failure.TimeoutType.START_TO_CLOSE
-                        ),
-                    )
+                        type=temporalio.exceptions.TimeoutType.START_TO_CLOSE,
+                    ),
                 )
             elif event.HasField("workflow_execution_continued_as_new_event_attributes"):
                 cont_attr = event.workflow_execution_continued_as_new_event_attributes
@@ -479,7 +472,7 @@ async def cancel(
         self,
         *,
         run_id: Optional[str] = SELF_RUN_ID,
-        first_execution_run_id: Optional[None],
+        first_execution_run_id: Optional[str] = None,
     ) -> None:
         """Cancel the workflow.
 
@@ -902,7 +895,7 @@ async def terminate_workflow(self, input: TerminateWorkflowInput) -> None:
 class WorkflowFailureError(Exception):
     """Error that occurs when a workflow is unsuccessful."""
 
-    def __init__(self, *, cause: temporalio.failure.FailureError) -> None:
+    def __init__(self, *, cause: temporalio.exceptions.FailureError) -> None:
         """Create workflow failure error."""
         super().__init__("Workflow execution failed")
         # TODO(cretz): Confirm setting this __cause__ is acceptable
@@ -917,6 +910,10 @@ def __init__(self, new_execution_run_id: str) -> None:
         super().__init__("Workflow continued as new")
         self._new_execution_run_id = new_execution_run_id
 
+    @property
+    def new_execution_run_id(self) -> str:
+        return self._new_execution_run_id
+
 
 class WorkflowQueryRejectedError(Exception):
     """Error that occurs when a query was rejected."""

temporalio/client.py

@@ -419,3 +419,11 @@ def default() -> CompositeDataConverter:
         BinaryProtoPayloadConverter(),
         JSONPlainPayloadConverter(),
     )
+
+
+async def decode_payloads(
+    payloads: Optional[temporalio.api.common.v1.Payloads], converter: DataConverter
+) -> List[Any]:
+    if not payloads or not payloads.payloads:
+        return []
+    return await converter.decode(payloads.payloads)