GeezerVisonThe World Wide Web, restored to 800 pixels of glory. |
| Home|About|Gaming cabinet|Archives|How it works|Starter sites |
GeezerVison is showing github.com — sdk-python/temporalio/client/_exceptions.py at main · temporalio/sdk-python Expand file tree/ Copy path_exceptions.pyMore file actions 135 lines (94 loc) · 4.31 KB / Copy path_exceptions.pyFile metadata and controls135 lines (94 loc) · 4.31 KB 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 """Client support for accessing Temporal.""" from __future__ import annotations from typing import ( TYPE_CHECKING, ) import temporalio.exceptions from temporalio.activity import ActivityCancellationDetails if TYPE_CHECKING: from ._workflow import WorkflowExecutionStatus class WorkflowFailureError(temporalio.exceptions.TemporalError): """Error that occurs when a workflow is unsuccessful.""" def __init__(self, *, cause: BaseException) -> None: """Create workflow failure error.""" super().__init__("Workflow execution failed") self.__cause__ = cause @property def cause(self) -> BaseException: """Cause of the workflow failure.""" assert self.__cause__ return self.__cause__ class WorkflowContinuedAsNewError(temporalio.exceptions.TemporalError): """Error that occurs when a workflow was continued as new.""" def __init__(self, new_execution_run_id: str) -> None: """Create workflow continue as new error.""" super().__init__("Workflow continued as new") self._new_execution_run_id = new_execution_run_id @property def new_execution_run_id(self) -> str: """New execution run ID the workflow continued to""" return self._new_execution_run_id class WorkflowQueryRejectedError(temporalio.exceptions.TemporalError): """Error that occurs when a query was rejected.""" def __init__(self, status: WorkflowExecutionStatus | None) -> None: """Create workflow query rejected error.""" super().__init__(f"Query rejected, status: {status}") self._status = status @property def status(self) -> WorkflowExecutionStatus | None: """Get workflow execution status causing rejection.""" return self._status class WorkflowQueryFailedError(temporalio.exceptions.TemporalError): """Error that occurs when a query fails.""" def __init__(self, message: str) -> None: """Create workflow query failed error.""" super().__init__(message) self._message = message @property def message(self) -> str: """Get query failed message.""" return self._message class WorkflowUpdateFailedError(temporalio.exceptions.TemporalError): """Error that occurs when an update fails.""" def __init__(self, cause: BaseException) -> None: """Create workflow update failed error.""" super().__init__("Workflow update failed") self.__cause__ = cause @property def cause(self) -> BaseException: """Cause of the update failure.""" assert self.__cause__ return self.__cause__ class RPCTimeoutOrCancelledError(temporalio.exceptions.TemporalError): """Error that occurs on some client calls that timeout or get cancelled.""" pass class WorkflowUpdateRPCTimeoutOrCancelledError(RPCTimeoutOrCancelledError): """Error that occurs when update RPC call times out or is cancelled. Note, this is not related to any general concept of timing out or cancelling a running update, this is only related to the client call itself. """ def __init__(self) -> None: """Create workflow update timeout or cancelled error.""" super().__init__("Timeout or cancellation waiting for update") class ActivityFailureError(temporalio.exceptions.TemporalError): """Error that occurs when an activity is unsuccessful.""" def __init__(self, *, cause: BaseException) -> None: """Create activity failure error.""" super().__init__("Activity execution failed") self.__cause__ = cause @property def cause(self) -> BaseException: """Cause of the activity failure.""" assert self.__cause__ return self.__cause__ class AsyncActivityCancelledError(temporalio.exceptions.TemporalError): """Error that occurs when async activity attempted heartbeat but was cancelled.""" def __init__(self, details: ActivityCancellationDetails | None = None) -> None: """Create async activity cancelled error.""" super().__init__("Activity cancelled") self.details = details class ScheduleAlreadyRunningError(temporalio.exceptions.TemporalError): """Error when a schedule is already running.""" def __init__(self) -> None: """Create schedule already running error.""" super().__init__("Schedule already running") |
| You are visitor 01402520 — GeezerVison does not store the pages you look up. Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required. |