@property

@abstractmethod

def instance_id(self) -> str:

"""Get the ID of the current orchestration instance.

pass

@property

@abstractmethod

def version(self) -> Optional[str]:

"""Get the version of the orchestration instance.

pass

@property

@abstractmethod

def current_utc_datetime(self) -> datetime:

"""Get the current date/time as UTC.

pass

@property

@abstractmethod

def is_replaying(self) -> bool:

"""Get the value indicating whether the orchestrator is replaying from history.

pass

@abstractmethod

def set_custom_status(self, custom_status: Any) -> None:

"""Set the orchestration instance's custom status.

pass

@abstractmethod

def create_timer(self, fire_at: Union[datetime, timedelta]) -> Task:

"""Create a Timer Task to fire after at the specified deadline.

pass

@abstractmethod

def call_activity(self, activity: Union[Activity[TInput, TOutput], str], *,

input: Optional[TInput] = None,

retry_policy: Optional[RetryPolicy] = None,

tags: Optional[dict[str, str]] = None) -> CompletableTask[TOutput]:

"""Schedule an activity for execution.

pass

@abstractmethod

def call_entity(self,

entity: EntityInstanceId,

operation: str,

input: Optional[TInput] = None) -> CompletableTask[Any]:

"""Schedule entity function for execution.

pass

@abstractmethod

def signal_entity(

self,

entity_id: EntityInstanceId,

operation_name: str,

input: Optional[TInput] = None

) -> None:

"""Signal an entity function for execution.

pass

@abstractmethod

def lock_entities(self, entities: list[EntityInstanceId]) -> CompletableTask[EntityLock]:

"""Creates a Task object that locks the specified entity instances.

pass

@abstractmethod

def call_sub_orchestrator(self, orchestrator: Union[Orchestrator[TInput, TOutput], str], *,

input: Optional[TInput] = None,

instance_id: Optional[str] = None,

retry_policy: Optional[RetryPolicy] = None,

version: Optional[str] = None) -> CompletableTask[TOutput]:

"""Schedule sub-orchestrator function for execution.

pass

# TOOD: Add a timeout parameter, which allows the task to be canceled if the event is

# not received within the specified timeout. This requires support for task cancellation.

@abstractmethod

def wait_for_external_event(self, name: str) -> CompletableTask:

"""Wait asynchronously for an event to be raised with the name `name`.

pass

@abstractmethod

def continue_as_new(self, new_input: Any, *, save_events: bool = False) -> None:

"""Continue the orchestration execution as a new instance.

pass

@abstractmethod

def new_uuid(self) -> str:

"""Create a new UUID that is safe for replay within an orchestration or operation.

pass

@abstractmethod

def _exit_critical_section(self) -> None:

def __init__(self, message: str, error_type: str, stack_trace: Optional[str]):

self._message = message

self._error_type = error_type

self._stack_trace = stack_trace

@property

def message(self) -> str:

return self._message

@property

def error_type(self) -> str:

return self._error_type

@property

def stack_trace(self) -> Optional[str]:

"""Exception type for all orchestration task failures."""

def __init__(self, message: str, details: Union[pb.TaskFailureDetails, Exception]):

super().__init__(message)

if isinstance(details, Exception):

details = pbh.new_failure_details(details)

self._details = FailureDetails(

details.errorMessage,

details.errorType,

details.stackTrace.value if not pbh.is_empty(details.stackTrace) else None)

@property

def details(self) -> FailureDetails:

"""Abstract base class for asynchronous tasks in a durable orchestration."""

_result: T

_exception: Optional[TaskFailedError]

_parent: Optional[CompositeTask[T]]

def __init__(self) -> None:

super().__init__()

self._is_complete = False

self._exception = None

self._parent = None

@property

def is_complete(self) -> bool:

"""Returns True if the task has completed, False otherwise."""

return self._is_complete

@property

def is_failed(self) -> bool:

"""Returns True if the task has failed, False otherwise."""

return self._exception is not None

def get_result(self) -> T:

"""Returns the result of the task."""

if not self._is_complete:

raise ValueError('The task has not completed.')

elif self._exception is not None:

raise self._exception

return self._result

def get_exception(self) -> TaskFailedError:

"""Returns the exception that caused the task to fail."""

if self._exception is None:

raise ValueError('The task has not failed.')

"""A task that is composed of other tasks."""

_tasks: list[Task]

def __init__(self, tasks: list[Task]):

super().__init__()

self._tasks = tasks

self._completed_tasks = 0

self._failed_tasks = 0

for task in tasks:

task._parent = self

if task.is_complete:

self.on_child_completed(task)

def get_tasks(self) -> list[Task]:

return self._tasks

@abstractmethod

def on_child_completed(self, task: Task[T]):

"""A task that completes when all of its child tasks complete."""

def __init__(self, tasks: list[Task[T]]):

super().__init__(tasks)

self._completed_tasks = 0

self._failed_tasks = 0

@property

def pending_tasks(self) -> int:

"""Returns the number of tasks that have not yet completed."""

return len(self._tasks) - self._completed_tasks

def on_child_completed(self, task: Task[T]):

if self.is_complete:

raise ValueError('The task has already completed.')

self._completed_tasks += 1

if task.is_failed and self._exception is None:

self._exception = task.get_exception()

self._is_complete = True

if self._completed_tasks == len(self._tasks):

# The order of the result MUST match the order of the tasks provided to the constructor.

self._result = [task.get_result() for task in self._tasks]

self._is_complete = True

def get_completed_tasks(self) -> int:

def __init__(self):

super().__init__()

self._retryable_parent = None

def complete(self, result: T):

if self._is_complete:

raise ValueError('The task has already completed.')

self._result = result

self._is_complete = True

if self._parent is not None:

self._parent.on_child_completed(self)

def fail(self, message: str, details: Union[Exception, pb.TaskFailureDetails]):

if self._is_complete:

raise ValueError('The task has already completed.')

self._exception = TaskFailedError(message, details)

self._is_complete = True

if self._parent is not None:

"""A task that can be retried according to a retry policy."""

def __init__(self, retry_policy: RetryPolicy, action: pb.OrchestratorAction,

start_time: datetime, is_sub_orch: bool) -> None:

super().__init__()

self._action = action

self._retry_policy = retry_policy

self._attempt_count = 1

self._start_time = start_time

self._is_sub_orch = is_sub_orch

def increment_attempt_count(self) -> None:

self._attempt_count += 1

def compute_next_delay(self) -> Optional[timedelta]:

if self._attempt_count >= self._retry_policy.max_number_of_attempts:

return None

retry_expiration: datetime = datetime.max

if self._retry_policy.retry_timeout is not None and self._retry_policy.retry_timeout != datetime.max:

retry_expiration = self._start_time + self._retry_policy.retry_timeout

if self._retry_policy.backoff_coefficient is None:

backoff_coefficient = 1.0

else:

backoff_coefficient = self._retry_policy.backoff_coefficient

if datetime.utcnow() < retry_expiration:

next_delay_f = math.pow(backoff_coefficient, self._attempt_count - 1) * self._retry_policy.first_retry_interval.total_seconds()

if self._retry_policy.max_retry_interval is not None:

next_delay_f = min(next_delay_f, self._retry_policy.max_retry_interval.total_seconds())

return timedelta(seconds=next_delay_f)

def __init__(self) -> None:

super().__init__()

def set_retryable_parent(self, retryable_task: RetryableTask):

"""A task that completes when any of its child tasks complete."""

def __init__(self, tasks: list[Task]):

super().__init__(tasks)

def on_child_completed(self, task: Task):

# The first task to complete is the result of the WhenAnyTask.

if not self.is_complete:

self._is_complete = True

def __init__(self, orchestration_id: str, task_id: int):

self._orchestration_id = orchestration_id

self._task_id = task_id

@property

def orchestration_id(self) -> str:

"""Get the ID of the orchestration instance that scheduled this activity.

return self._orchestration_id

@property

def task_id(self) -> int:

"""Get the task ID associated with this activity invocation.

"""Represents the retry policy for an orchestration or activity function."""

def __init__(self, *,

first_retry_interval: timedelta,

max_number_of_attempts: int,

backoff_coefficient: Optional[float] = 1.0,

max_retry_interval: Optional[timedelta] = None,

retry_timeout: Optional[timedelta] = None):

"""Creates a new RetryPolicy instance.

# validate inputs

if first_retry_interval < timedelta(seconds=0):

raise ValueError('first_retry_interval must be >= 0')

if max_number_of_attempts < 1:

raise ValueError('max_number_of_attempts must be >= 1')

if backoff_coefficient is not None and backoff_coefficient < 1:

raise ValueError('backoff_coefficient must be >= 1')

if max_retry_interval is not None and max_retry_interval < timedelta(seconds=0):

raise ValueError('max_retry_interval must be >= 0')

if retry_timeout is not None and retry_timeout < timedelta(seconds=0):

raise ValueError('retry_timeout must be >= 0')

self._first_retry_interval = first_retry_interval

self._max_number_of_attempts = max_number_of_attempts

self._backoff_coefficient = backoff_coefficient

self._max_retry_interval = max_retry_interval

self._retry_timeout = retry_timeout

@property

def first_retry_interval(self) -> timedelta:

"""The retry interval to use for the first retry attempt."""

return self._first_retry_interval

@property

def max_number_of_attempts(self) -> int:

"""The maximum number of retry attempts."""

return self._max_number_of_attempts

@property

def backoff_coefficient(self) -> Optional[float]:

"""The backoff coefficient to use for calculating the next retry interval."""

return self._backoff_coefficient

@property

def max_retry_interval(self) -> Optional[timedelta]:

"""The maximum retry interval to use for any retry attempt."""

return self._max_retry_interval

@property

def retry_timeout(self) -> Optional[timedelta]:

"""The maximum amount of time to spend retrying the operation."""

if hasattr(fn, "__durable_entity_name__"):

return getattr(fn, "__durable_entity_name__")

if isinstance(fn, type) and issubclass(fn, DurableEntity):

return fn.__name__

"""Returns the name of the provided function"""

name = fn.__name__

if name == '<lambda>':

raise ValueError('Cannot infer a name from a lambda function. Please provide a name explicitly.')