Temporal

The temporal adapter provides integration with Temporal, a durable workflow orchestration platform, including workflow and activity definitions, worker management, and runtime configuration.

Ports

Abstract port interface defining the Temporal adapter contract.

Port interfaces for Temporal workflow orchestration.

This module defines the abstract interfaces for Temporal workflow and activity operations, providing a standardized contract for workflow orchestration within the ArchiPy architecture.

archipy.adapters.temporal.ports.T module-attribute

T = TypeVar('T')

archipy.adapters.temporal.ports.TemporalPort

Interface for Temporal workflow operations providing a standardized access pattern.

This interface defines the contract for Temporal adapters, ensuring consistent implementation of workflow operations across different adapters. It covers workflow lifecycle management, execution control, and query operations.

Implementing classes should provide concrete implementations for all methods, typically by wrapping a Temporal client library.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

class TemporalPort:
    """Interface for Temporal workflow operations providing a standardized access pattern.

    This interface defines the contract for Temporal adapters, ensuring consistent
    implementation of workflow operations across different adapters. It covers
    workflow lifecycle management, execution control, and query operations.

    Implementing classes should provide concrete implementations for all
    methods, typically by wrapping a Temporal client library.
    """

    @abstractmethod
    async def start_workflow(
        self,
        workflow: str | Callable[..., Any],
        arg: Any = None,
        workflow_id: str | None = None,
        task_queue: str | None = None,
        execution_timeout: int | None = None,
        run_timeout: int | None = None,
        task_timeout: int | None = None,
        memo: dict[str, Any] | None = None,
        search_attributes: dict[str, Any] | None = None,
        id_reuse_policy: WorkflowIDReusePolicy | None = None,
        id_conflict_policy: WorkflowIDConflictPolicy | None = None,
    ) -> TemporalWorkflowHandle[T, Any]:
        """Start a workflow execution asynchronously.

        Args:
            workflow (str | Callable): The workflow function or workflow type name.
            arg (Any, optional): Input argument for the workflow. Defaults to None.
            workflow_id (str, optional): Unique identifier for the workflow execution.
                If None, a UUID will be generated. Defaults to None.
            task_queue (str, optional): Task queue name for workflow execution.
                If None, uses the default task queue. Defaults to None.
            execution_timeout (int, optional): Maximum workflow execution time in seconds.
                Overrides config default. Defaults to None.
            run_timeout (int, optional): Maximum single workflow run time in seconds.
                Overrides config default. Defaults to None.
            task_timeout (int, optional): Maximum workflow task processing time in seconds.
                Overrides config default. Defaults to None.
            memo (dict[str, Any], optional): Non-indexed metadata for the workflow.
                Defaults to None.
            search_attributes (dict[str, Any], optional): Indexed metadata for workflow search.
                Defaults to None.
            id_reuse_policy (WorkflowIDReusePolicy, optional): Whether a closed workflow's ID may be reused.
                Defaults to None (server default: allow duplicate).
            id_conflict_policy (WorkflowIDConflictPolicy, optional): What to do when a workflow with the same ID
                is still running, e.g. ``USE_EXISTING`` for idempotent starts. Defaults to None (server default: fail).

        Returns:
            WorkflowHandle[T, Any]: Handle to the started workflow execution.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

    @abstractmethod
    async def execute_workflow(
        self,
        workflow: str | Callable[..., Any],
        arg: Any = None,
        workflow_id: str | None = None,
        task_queue: str | None = None,
        execution_timeout: int | None = None,
        run_timeout: int | None = None,
        task_timeout: int | None = None,
        id_reuse_policy: WorkflowIDReusePolicy | None = None,
        id_conflict_policy: WorkflowIDConflictPolicy | None = None,
    ) -> T:
        """Execute a workflow and wait for its completion.

        Args:
            workflow (str | Callable): The workflow function or workflow type name.
            arg (Any, optional): Input argument for the workflow. Defaults to None.
            workflow_id (str, optional): Unique identifier for the workflow execution.
                If None, a UUID will be generated. Defaults to None.
            task_queue (str, optional): Task queue name for workflow execution.
                If None, uses the default task queue. Defaults to None.
            execution_timeout (int, optional): Maximum workflow execution time in seconds.
                Overrides config default. Defaults to None.
            run_timeout (int, optional): Maximum single workflow run time in seconds.
                Overrides config default. Defaults to None.
            task_timeout (int, optional): Maximum workflow task processing time in seconds.
                Overrides config default. Defaults to None.
            id_reuse_policy (WorkflowIDReusePolicy, optional): Whether a closed workflow's ID may be reused.
                Defaults to None (server default: allow duplicate).
            id_conflict_policy (WorkflowIDConflictPolicy, optional): What to do when a workflow with the same ID
                is still running, e.g. ``USE_EXISTING`` for idempotent starts. Defaults to None (server default: fail).

        Returns:
            T: The workflow execution result.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

    @abstractmethod
    async def get_workflow_handle(self, workflow_id: str, run_id: str | None = None) -> TemporalWorkflowHandle[T, Any]:
        """Get a handle to an existing workflow execution.

        Args:
            workflow_id (str): The unique identifier of the workflow execution.
            run_id (str, optional): The specific run identifier within the workflow.
                If None, gets the latest run. Defaults to None.

        Returns:
            WorkflowHandle[T, Any]: Handle to the workflow execution.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

    @abstractmethod
    async def cancel_workflow(self, workflow_id: str, run_id: str | None = None, reason: str | None = None) -> None:
        """Cancel a running workflow execution.

        Args:
            workflow_id (str): The unique identifier of the workflow execution.
            run_id (str, optional): The specific run identifier within the workflow.
                If None, cancels the latest run. Defaults to None.
            reason (str, optional): Reason for cancellation. Defaults to None.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

    @abstractmethod
    async def terminate_workflow(self, workflow_id: str, run_id: str | None = None, reason: str | None = None) -> None:
        """Terminate a running workflow execution immediately.

        Args:
            workflow_id (str): The unique identifier of the workflow execution.
            run_id (str, optional): The specific run identifier within the workflow.
                If None, terminates the latest run. Defaults to None.
            reason (str, optional): Reason for termination. Defaults to None.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

    @abstractmethod
    async def signal_workflow(
        self,
        workflow_id: str,
        signal_name: str,
        arg: Any = None,
        run_id: str | None = None,
    ) -> None:
        """Send a signal to a running workflow execution.

        Args:
            workflow_id (str): The unique identifier of the workflow execution.
            signal_name (str): The name of the signal to send.
            arg (Any, optional): Argument to pass with the signal. Defaults to None.
            run_id (str, optional): The specific run identifier within the workflow.
                If None, signals the latest run. Defaults to None.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

    @abstractmethod
    async def query_workflow(
        self,
        workflow_id: str,
        query_name: str,
        arg: Any = None,
        run_id: str | None = None,
    ) -> Any:
        """Query a running workflow execution for information.

        Args:
            workflow_id (str): The unique identifier of the workflow execution.
            query_name (str): The name of the query to execute.
            arg (Any, optional): Argument to pass with the query. Defaults to None.
            run_id (str, optional): The specific run identifier within the workflow.
                If None, queries the latest run. Defaults to None.

        Returns:
            Any: The query result from the workflow.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

    @abstractmethod
    async def list_workflows(
        self,
        query: str | None = None,
        page_size: int | None = None,
        next_page_token: bytes | None = None,
        limit: int | None = None,
    ) -> list[WorkflowExecution]:
        """List workflow executions matching the given criteria.

        Args:
            query (str, optional): List filter query in Temporal SQL syntax.
                Defaults to None (no filter).
            page_size (int, optional): Maximum number of results per page.
                Defaults to None (server default).
            next_page_token (bytes, optional): Token for pagination.
                Defaults to None (first page).
            limit (int, optional): Maximum number of executions to return. Defaults to None, which loads every
                matching execution into memory; set it, or use ``iter_workflows``, for large namespaces.

        Returns:
            list[WorkflowExecution]: The workflow executions matching the query.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

    @abstractmethod
    async def describe_workflow(self, workflow_id: str, run_id: str | None = None) -> WorkflowExecutionDescription:
        """Get detailed information about a workflow execution.

        Args:
            workflow_id (str): The unique identifier of the workflow execution.
            run_id (str, optional): The specific run identifier within the workflow.
                If None, describes the latest run. Defaults to None.

        Returns:
            WorkflowExecutionDescription: Detailed workflow execution information.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

    @abstractmethod
    async def close(self) -> None:
        """Close the Temporal client connection.

        Performs cleanup of resources and closes the connection to the Temporal server.
        Should be called when the adapter is no longer needed.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

    @abstractmethod
    async def create_schedule(
        self,
        schedule_id: str,
        workflow_class: Any,
        spec: ScheduleSpec,
        task_queue: str,
    ) -> None:
        """Create a new schedule."""
        raise NotImplementedError

    @abstractmethod
    async def stop_schedule(self, schedule_id: str) -> None:
        """Delete a schedule (deprecated name, use ``delete_schedule``)."""
        raise NotImplementedError

    async def delete_schedule(self, schedule_id: str) -> None:
        """Delete a schedule permanently; ``pause_schedule`` stops it while keeping it.

        Defaults to ``stop_schedule`` so existing implementations keep working.

        Args:
            schedule_id (str): The schedule ID.
        """
        await self.stop_schedule(schedule_id)

archipy.adapters.temporal.ports.TemporalPort.start_workflow abstractmethod async

start_workflow(
    workflow: str | Callable[..., Any],
    arg: Any = None,
    workflow_id: str | None = None,
    task_queue: str | None = None,
    execution_timeout: int | None = None,
    run_timeout: int | None = None,
    task_timeout: int | None = None,
    memo: dict[str, Any] | None = None,
    search_attributes: dict[str, Any] | None = None,
    id_reuse_policy: WorkflowIDReusePolicy | None = None,
    id_conflict_policy: WorkflowIDConflictPolicy
    | None = None,
) -> TemporalWorkflowHandle[T, Any]

Start a workflow execution asynchronously.

Parameters:

Name	Type	Description	Default
workflow	str | Callable	The workflow function or workflow type name.	required
arg	Any	Input argument for the workflow. Defaults to None.	None
workflow_id	str	Unique identifier for the workflow execution. If None, a UUID will be generated. Defaults to None.	None
task_queue	str	Task queue name for workflow execution. If None, uses the default task queue. Defaults to None.	None
execution_timeout	int	Maximum workflow execution time in seconds. Overrides config default. Defaults to None.	None
run_timeout	int	Maximum single workflow run time in seconds. Overrides config default. Defaults to None.	None
task_timeout	int	Maximum workflow task processing time in seconds. Overrides config default. Defaults to None.	None
memo	dict[str, Any]	Non-indexed metadata for the workflow. Defaults to None.	None
search_attributes	dict[str, Any]	Indexed metadata for workflow search. Defaults to None.	None
id_reuse_policy	WorkflowIDReusePolicy	Whether a closed workflow's ID may be reused. Defaults to None (server default: allow duplicate).	None
id_conflict_policy	WorkflowIDConflictPolicy	What to do when a workflow with the same ID is still running, e.g. USE_EXISTING for idempotent starts. Defaults to None (server default: fail).	None

Returns:

Type	Description
WorkflowHandle[T, Any]	WorkflowHandle[T, Any]: Handle to the started workflow execution.

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def start_workflow(
    self,
    workflow: str | Callable[..., Any],
    arg: Any = None,
    workflow_id: str | None = None,
    task_queue: str | None = None,
    execution_timeout: int | None = None,
    run_timeout: int | None = None,
    task_timeout: int | None = None,
    memo: dict[str, Any] | None = None,
    search_attributes: dict[str, Any] | None = None,
    id_reuse_policy: WorkflowIDReusePolicy | None = None,
    id_conflict_policy: WorkflowIDConflictPolicy | None = None,
) -> TemporalWorkflowHandle[T, Any]:
    """Start a workflow execution asynchronously.

    Args:
        workflow (str | Callable): The workflow function or workflow type name.
        arg (Any, optional): Input argument for the workflow. Defaults to None.
        workflow_id (str, optional): Unique identifier for the workflow execution.
            If None, a UUID will be generated. Defaults to None.
        task_queue (str, optional): Task queue name for workflow execution.
            If None, uses the default task queue. Defaults to None.
        execution_timeout (int, optional): Maximum workflow execution time in seconds.
            Overrides config default. Defaults to None.
        run_timeout (int, optional): Maximum single workflow run time in seconds.
            Overrides config default. Defaults to None.
        task_timeout (int, optional): Maximum workflow task processing time in seconds.
            Overrides config default. Defaults to None.
        memo (dict[str, Any], optional): Non-indexed metadata for the workflow.
            Defaults to None.
        search_attributes (dict[str, Any], optional): Indexed metadata for workflow search.
            Defaults to None.
        id_reuse_policy (WorkflowIDReusePolicy, optional): Whether a closed workflow's ID may be reused.
            Defaults to None (server default: allow duplicate).
        id_conflict_policy (WorkflowIDConflictPolicy, optional): What to do when a workflow with the same ID
            is still running, e.g. ``USE_EXISTING`` for idempotent starts. Defaults to None (server default: fail).

    Returns:
        WorkflowHandle[T, Any]: Handle to the started workflow execution.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.TemporalPort.execute_workflow abstractmethod async

execute_workflow(
    workflow: str | Callable[..., Any],
    arg: Any = None,
    workflow_id: str | None = None,
    task_queue: str | None = None,
    execution_timeout: int | None = None,
    run_timeout: int | None = None,
    task_timeout: int | None = None,
    id_reuse_policy: WorkflowIDReusePolicy | None = None,
    id_conflict_policy: WorkflowIDConflictPolicy
    | None = None,
) -> T

Execute a workflow and wait for its completion.

Parameters:

Name	Type	Description	Default
workflow	str | Callable	The workflow function or workflow type name.	required
arg	Any	Input argument for the workflow. Defaults to None.	None
workflow_id	str	Unique identifier for the workflow execution. If None, a UUID will be generated. Defaults to None.	None
task_queue	str	Task queue name for workflow execution. If None, uses the default task queue. Defaults to None.	None
execution_timeout	int	Maximum workflow execution time in seconds. Overrides config default. Defaults to None.	None
run_timeout	int	Maximum single workflow run time in seconds. Overrides config default. Defaults to None.	None
task_timeout	int	Maximum workflow task processing time in seconds. Overrides config default. Defaults to None.	None
id_reuse_policy	WorkflowIDReusePolicy	Whether a closed workflow's ID may be reused. Defaults to None (server default: allow duplicate).	None
id_conflict_policy	WorkflowIDConflictPolicy	What to do when a workflow with the same ID is still running, e.g. USE_EXISTING for idempotent starts. Defaults to None (server default: fail).	None

Returns:

Name	Type	Description
T	T	The workflow execution result.

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def execute_workflow(
    self,
    workflow: str | Callable[..., Any],
    arg: Any = None,
    workflow_id: str | None = None,
    task_queue: str | None = None,
    execution_timeout: int | None = None,
    run_timeout: int | None = None,
    task_timeout: int | None = None,
    id_reuse_policy: WorkflowIDReusePolicy | None = None,
    id_conflict_policy: WorkflowIDConflictPolicy | None = None,
) -> T:
    """Execute a workflow and wait for its completion.

    Args:
        workflow (str | Callable): The workflow function or workflow type name.
        arg (Any, optional): Input argument for the workflow. Defaults to None.
        workflow_id (str, optional): Unique identifier for the workflow execution.
            If None, a UUID will be generated. Defaults to None.
        task_queue (str, optional): Task queue name for workflow execution.
            If None, uses the default task queue. Defaults to None.
        execution_timeout (int, optional): Maximum workflow execution time in seconds.
            Overrides config default. Defaults to None.
        run_timeout (int, optional): Maximum single workflow run time in seconds.
            Overrides config default. Defaults to None.
        task_timeout (int, optional): Maximum workflow task processing time in seconds.
            Overrides config default. Defaults to None.
        id_reuse_policy (WorkflowIDReusePolicy, optional): Whether a closed workflow's ID may be reused.
            Defaults to None (server default: allow duplicate).
        id_conflict_policy (WorkflowIDConflictPolicy, optional): What to do when a workflow with the same ID
            is still running, e.g. ``USE_EXISTING`` for idempotent starts. Defaults to None (server default: fail).

    Returns:
        T: The workflow execution result.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.TemporalPort.get_workflow_handle abstractmethod async

get_workflow_handle(
    workflow_id: str, run_id: str | None = None
) -> TemporalWorkflowHandle[T, Any]

Get a handle to an existing workflow execution.

Parameters:

Name	Type	Description	Default
workflow_id	str	The unique identifier of the workflow execution.	required
run_id	str	The specific run identifier within the workflow. If None, gets the latest run. Defaults to None.	None

Returns:

Type	Description
WorkflowHandle[T, Any]	WorkflowHandle[T, Any]: Handle to the workflow execution.

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def get_workflow_handle(self, workflow_id: str, run_id: str | None = None) -> TemporalWorkflowHandle[T, Any]:
    """Get a handle to an existing workflow execution.

    Args:
        workflow_id (str): The unique identifier of the workflow execution.
        run_id (str, optional): The specific run identifier within the workflow.
            If None, gets the latest run. Defaults to None.

    Returns:
        WorkflowHandle[T, Any]: Handle to the workflow execution.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.TemporalPort.cancel_workflow abstractmethod async

cancel_workflow(
    workflow_id: str,
    run_id: str | None = None,
    reason: str | None = None,
) -> None

Cancel a running workflow execution.

Parameters:

Name	Type	Description	Default
workflow_id	str	The unique identifier of the workflow execution.	required
run_id	str	The specific run identifier within the workflow. If None, cancels the latest run. Defaults to None.	None
reason	str	Reason for cancellation. Defaults to None.	None

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def cancel_workflow(self, workflow_id: str, run_id: str | None = None, reason: str | None = None) -> None:
    """Cancel a running workflow execution.

    Args:
        workflow_id (str): The unique identifier of the workflow execution.
        run_id (str, optional): The specific run identifier within the workflow.
            If None, cancels the latest run. Defaults to None.
        reason (str, optional): Reason for cancellation. Defaults to None.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.TemporalPort.terminate_workflow abstractmethod async

terminate_workflow(
    workflow_id: str,
    run_id: str | None = None,
    reason: str | None = None,
) -> None

Terminate a running workflow execution immediately.

Parameters:

Name	Type	Description	Default
workflow_id	str	The unique identifier of the workflow execution.	required
run_id	str	The specific run identifier within the workflow. If None, terminates the latest run. Defaults to None.	None
reason	str	Reason for termination. Defaults to None.	None

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def terminate_workflow(self, workflow_id: str, run_id: str | None = None, reason: str | None = None) -> None:
    """Terminate a running workflow execution immediately.

    Args:
        workflow_id (str): The unique identifier of the workflow execution.
        run_id (str, optional): The specific run identifier within the workflow.
            If None, terminates the latest run. Defaults to None.
        reason (str, optional): Reason for termination. Defaults to None.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.TemporalPort.signal_workflow abstractmethod async

signal_workflow(
    workflow_id: str,
    signal_name: str,
    arg: Any = None,
    run_id: str | None = None,
) -> None

Send a signal to a running workflow execution.

Parameters:

Name	Type	Description	Default
workflow_id	str	The unique identifier of the workflow execution.	required
signal_name	str	The name of the signal to send.	required
arg	Any	Argument to pass with the signal. Defaults to None.	None
run_id	str	The specific run identifier within the workflow. If None, signals the latest run. Defaults to None.	None

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def signal_workflow(
    self,
    workflow_id: str,
    signal_name: str,
    arg: Any = None,
    run_id: str | None = None,
) -> None:
    """Send a signal to a running workflow execution.

    Args:
        workflow_id (str): The unique identifier of the workflow execution.
        signal_name (str): The name of the signal to send.
        arg (Any, optional): Argument to pass with the signal. Defaults to None.
        run_id (str, optional): The specific run identifier within the workflow.
            If None, signals the latest run. Defaults to None.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.TemporalPort.query_workflow abstractmethod async

query_workflow(
    workflow_id: str,
    query_name: str,
    arg: Any = None,
    run_id: str | None = None,
) -> Any

Query a running workflow execution for information.

Parameters:

Name	Type	Description	Default
workflow_id	str	The unique identifier of the workflow execution.	required
query_name	str	The name of the query to execute.	required
arg	Any	Argument to pass with the query. Defaults to None.	None
run_id	str	The specific run identifier within the workflow. If None, queries the latest run. Defaults to None.	None

Returns:

Name	Type	Description
Any	Any	The query result from the workflow.

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def query_workflow(
    self,
    workflow_id: str,
    query_name: str,
    arg: Any = None,
    run_id: str | None = None,
) -> Any:
    """Query a running workflow execution for information.

    Args:
        workflow_id (str): The unique identifier of the workflow execution.
        query_name (str): The name of the query to execute.
        arg (Any, optional): Argument to pass with the query. Defaults to None.
        run_id (str, optional): The specific run identifier within the workflow.
            If None, queries the latest run. Defaults to None.

    Returns:
        Any: The query result from the workflow.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.TemporalPort.list_workflows abstractmethod async

list_workflows(
    query: str | None = None,
    page_size: int | None = None,
    next_page_token: bytes | None = None,
    limit: int | None = None,
) -> list[WorkflowExecution]

List workflow executions matching the given criteria.

Parameters:

Name	Type	Description	Default
query	str	List filter query in Temporal SQL syntax. Defaults to None (no filter).	None
page_size	int	Maximum number of results per page. Defaults to None (server default).	None
next_page_token	bytes	Token for pagination. Defaults to None (first page).	None
limit	int	Maximum number of executions to return. Defaults to None, which loads every matching execution into memory; set it, or use iter_workflows, for large namespaces.	None

Returns:

Type	Description
list[WorkflowExecution]	list[WorkflowExecution]: The workflow executions matching the query.

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def list_workflows(
    self,
    query: str | None = None,
    page_size: int | None = None,
    next_page_token: bytes | None = None,
    limit: int | None = None,
) -> list[WorkflowExecution]:
    """List workflow executions matching the given criteria.

    Args:
        query (str, optional): List filter query in Temporal SQL syntax.
            Defaults to None (no filter).
        page_size (int, optional): Maximum number of results per page.
            Defaults to None (server default).
        next_page_token (bytes, optional): Token for pagination.
            Defaults to None (first page).
        limit (int, optional): Maximum number of executions to return. Defaults to None, which loads every
            matching execution into memory; set it, or use ``iter_workflows``, for large namespaces.

    Returns:
        list[WorkflowExecution]: The workflow executions matching the query.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.TemporalPort.describe_workflow abstractmethod async

describe_workflow(
    workflow_id: str, run_id: str | None = None
) -> WorkflowExecutionDescription

Get detailed information about a workflow execution.

Parameters:

Name	Type	Description	Default
workflow_id	str	The unique identifier of the workflow execution.	required
run_id	str	The specific run identifier within the workflow. If None, describes the latest run. Defaults to None.	None

Returns:

Name	Type	Description
WorkflowExecutionDescription	WorkflowExecutionDescription	Detailed workflow execution information.

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def describe_workflow(self, workflow_id: str, run_id: str | None = None) -> WorkflowExecutionDescription:
    """Get detailed information about a workflow execution.

    Args:
        workflow_id (str): The unique identifier of the workflow execution.
        run_id (str, optional): The specific run identifier within the workflow.
            If None, describes the latest run. Defaults to None.

    Returns:
        WorkflowExecutionDescription: Detailed workflow execution information.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.TemporalPort.close abstractmethod async

close() -> None

Close the Temporal client connection.

Performs cleanup of resources and closes the connection to the Temporal server. Should be called when the adapter is no longer needed.

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def close(self) -> None:
    """Close the Temporal client connection.

    Performs cleanup of resources and closes the connection to the Temporal server.
    Should be called when the adapter is no longer needed.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.TemporalPort.create_schedule abstractmethod async

create_schedule(
    schedule_id: str,
    workflow_class: Any,
    spec: ScheduleSpec,
    task_queue: str,
) -> None

Create a new schedule.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def create_schedule(
    self,
    schedule_id: str,
    workflow_class: Any,
    spec: ScheduleSpec,
    task_queue: str,
) -> None:
    """Create a new schedule."""
    raise NotImplementedError

archipy.adapters.temporal.ports.TemporalPort.stop_schedule abstractmethod async

stop_schedule(schedule_id: str) -> None

Delete a schedule (deprecated name, use delete_schedule).

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def stop_schedule(self, schedule_id: str) -> None:
    """Delete a schedule (deprecated name, use ``delete_schedule``)."""
    raise NotImplementedError

archipy.adapters.temporal.ports.TemporalPort.delete_schedule async

delete_schedule(schedule_id: str) -> None

Delete a schedule permanently; pause_schedule stops it while keeping it.

Defaults to stop_schedule so existing implementations keep working.

Parameters:

Name	Type	Description	Default
schedule_id	str	The schedule ID.	required

Source code in packages/temporal/archipy/adapters/temporal/ports.py

async def delete_schedule(self, schedule_id: str) -> None:
    """Delete a schedule permanently; ``pause_schedule`` stops it while keeping it.

    Defaults to ``stop_schedule`` so existing implementations keep working.

    Args:
        schedule_id (str): The schedule ID.
    """
    await self.stop_schedule(schedule_id)

archipy.adapters.temporal.ports.WorkerPort

Interface for Temporal worker operations providing a standardized access pattern.

This interface defines the contract for Temporal worker management, ensuring consistent implementation of worker lifecycle operations. Workers are responsible for executing workflows and activities.

Implementing classes should provide concrete implementations for all methods, typically by wrapping a Temporal worker.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

class WorkerPort:
    """Interface for Temporal worker operations providing a standardized access pattern.

    This interface defines the contract for Temporal worker management, ensuring consistent
    implementation of worker lifecycle operations. Workers are responsible for executing
    workflows and activities.

    Implementing classes should provide concrete implementations for all
    methods, typically by wrapping a Temporal worker.
    """

    @abstractmethod
    async def start_worker(
        self,
        task_queue: str,
        workflows: list[type] | None = None,
        activities: list[Callable[..., Any]] | None = None,
        build_id: str | None = None,
        identity: str | None = None,
        max_concurrent_workflow_tasks: int | None = None,
        max_concurrent_activities: int | None = None,
    ) -> WorkerHandle:
        """Start a Temporal worker for the specified task queue.

        Args:
            task_queue (str): The task queue this worker will poll from.
            workflows (list[type], optional): List of workflow classes to register.
                Defaults to None.
            activities (list[Callable], optional): List of activity callables to register.
                Defaults to None.
            build_id (str, optional): Build identifier for worker versioning.
                Defaults to None.
            identity (str, optional): Unique worker identity. If None, auto-generated.
                Defaults to None.
            max_concurrent_workflow_tasks (int, optional): Maximum concurrent workflow tasks.
                Defaults to None (server default).
            max_concurrent_activities (int, optional): Maximum concurrent activity tasks.
                Defaults to None (server default).

        Returns:
            WorkerHandle: Handle to the started worker.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

    @abstractmethod
    async def stop_worker(self, worker_handle: WorkerHandle) -> None:
        """Stop a running Temporal worker.

        Args:
            worker_handle (WorkerHandle): Handle to the worker to stop.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

    @abstractmethod
    async def shutdown_all_workers(self) -> None:
        """Shutdown all workers managed by this port.

        Performs graceful shutdown of all active workers, waiting for current
        tasks to complete before terminating.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError

archipy.adapters.temporal.ports.WorkerPort.start_worker abstractmethod async

start_worker(
    task_queue: str,
    workflows: list[type] | None = None,
    activities: list[Callable[..., Any]] | None = None,
    build_id: str | None = None,
    identity: str | None = None,
    max_concurrent_workflow_tasks: int | None = None,
    max_concurrent_activities: int | None = None,
) -> WorkerHandle

Start a Temporal worker for the specified task queue.

Parameters:

Name	Type	Description	Default
task_queue	str	The task queue this worker will poll from.	required
workflows	list[type]	List of workflow classes to register. Defaults to None.	None
activities	list[Callable]	List of activity callables to register. Defaults to None.	None
build_id	str	Build identifier for worker versioning. Defaults to None.	None
identity	str	Unique worker identity. If None, auto-generated. Defaults to None.	None
max_concurrent_workflow_tasks	int	Maximum concurrent workflow tasks. Defaults to None (server default).	None
max_concurrent_activities	int	Maximum concurrent activity tasks. Defaults to None (server default).	None

Returns:

Name	Type	Description
WorkerHandle	WorkerHandle	Handle to the started worker.

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def start_worker(
    self,
    task_queue: str,
    workflows: list[type] | None = None,
    activities: list[Callable[..., Any]] | None = None,
    build_id: str | None = None,
    identity: str | None = None,
    max_concurrent_workflow_tasks: int | None = None,
    max_concurrent_activities: int | None = None,
) -> WorkerHandle:
    """Start a Temporal worker for the specified task queue.

    Args:
        task_queue (str): The task queue this worker will poll from.
        workflows (list[type], optional): List of workflow classes to register.
            Defaults to None.
        activities (list[Callable], optional): List of activity callables to register.
            Defaults to None.
        build_id (str, optional): Build identifier for worker versioning.
            Defaults to None.
        identity (str, optional): Unique worker identity. If None, auto-generated.
            Defaults to None.
        max_concurrent_workflow_tasks (int, optional): Maximum concurrent workflow tasks.
            Defaults to None (server default).
        max_concurrent_activities (int, optional): Maximum concurrent activity tasks.
            Defaults to None (server default).

    Returns:
        WorkerHandle: Handle to the started worker.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.WorkerPort.stop_worker abstractmethod async

stop_worker(worker_handle: WorkerHandle) -> None

Stop a running Temporal worker.

Parameters:

Name	Type	Description	Default
worker_handle	WorkerHandle	Handle to the worker to stop.	required

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def stop_worker(self, worker_handle: WorkerHandle) -> None:
    """Stop a running Temporal worker.

    Args:
        worker_handle (WorkerHandle): Handle to the worker to stop.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.WorkerPort.shutdown_all_workers abstractmethod async

shutdown_all_workers() -> None

Shutdown all workers managed by this port.

Performs graceful shutdown of all active workers, waiting for current tasks to complete before terminating.

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def shutdown_all_workers(self) -> None:
    """Shutdown all workers managed by this port.

    Performs graceful shutdown of all active workers, waiting for current
    tasks to complete before terminating.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError

archipy.adapters.temporal.ports.WorkflowHandle

Type stub for workflow handle.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

class WorkflowHandle:
    """Type stub for workflow handle."""

archipy.adapters.temporal.ports.WorkflowListResponse

Type stub for workflow list response.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

class WorkflowListResponse:
    """Type stub for workflow list response."""

archipy.adapters.temporal.ports.WorkflowDescription

Type stub for workflow description.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

class WorkflowDescription:
    """Type stub for workflow description."""

archipy.adapters.temporal.ports.WorkerHandle

Bases: ABC

Base type for worker handle.

This is an abstract base class that concrete implementations should extend. It provides a common interface for worker handle operations.

Source code in packages/temporal/archipy/adapters/temporal/ports.py

class WorkerHandle(ABC):
    """Base type for worker handle.

    This is an abstract base class that concrete implementations should extend.
    It provides a common interface for worker handle operations.
    """

    worker_id: str
    task_queue: str

    @abstractmethod
    async def stop(self, grace_period: int = 30) -> None:
        """Stop the worker gracefully.

        Args:
            grace_period: Maximum time in seconds to wait for graceful shutdown.
        """
        ...

archipy.adapters.temporal.ports.WorkerHandle.worker_id instance-attribute

worker_id: str

archipy.adapters.temporal.ports.WorkerHandle.task_queue instance-attribute

task_queue: str

archipy.adapters.temporal.ports.WorkerHandle.stop abstractmethod async

stop(grace_period: int = 30) -> None

Stop the worker gracefully.

Parameters:

Name	Type	Description	Default
grace_period	int	Maximum time in seconds to wait for graceful shutdown.	30

Source code in packages/temporal/archipy/adapters/temporal/ports.py

@abstractmethod
async def stop(self, grace_period: int = 30) -> None:
    """Stop the worker gracefully.

    Args:
        grace_period: Maximum time in seconds to wait for graceful shutdown.
    """
    ...

options: show_root_toc_entry: false heading_level: 3

Base

Base classes for Temporal workflow and activity implementations.

Base classes for Temporal workflows and activities.

This module provides base classes and utilities for implementing Temporal workflows and activities within the ArchiPy architecture, including integration with existing adapters and standardized patterns.

archipy.adapters.temporal.base.HTTP_CLIENT_ERROR_MIN module-attribute

HTTP_CLIENT_ERROR_MIN = 400

archipy.adapters.temporal.base.HTTP_SERVER_ERROR_MIN module-attribute

HTTP_SERVER_ERROR_MIN = 500

archipy.adapters.temporal.base.T module-attribute

T = TypeVar('T')

archipy.adapters.temporal.base.R module-attribute

R = TypeVar('R')

archipy.adapters.temporal.base.BaseWorkflow

Base class for all Temporal workflows in ArchiPy.

Provides common functionality and patterns for workflow implementations, including standardized logging, error handling, and integration with ArchiPy services through activities.

Class Type Parameters:

Name	Bound or Constraints	Description	Default
T		Type of the workflow input parameter.	required
R		Type of the workflow return value.	required

Source code in packages/temporal/archipy/adapters/temporal/base.py

class BaseWorkflow[T, R]:
    """Base class for all Temporal workflows in ArchiPy.

    Provides common functionality and patterns for workflow implementations,
    including standardized logging, error handling, and integration with
    ArchiPy services through activities.

    Type Parameters:
        T: Type of the workflow input parameter.
        R: Type of the workflow return value.
    """

    @workflow.run
    async def run(self, workflow_input: T) -> R:
        """Main workflow execution method.

        This method must be implemented by concrete workflow classes to define
        the workflow logic. It should orchestrate activities and child workflows
        to accomplish the business process.

        Args:
            workflow_input (T): The input data for the workflow.

        Returns:
            R: The result of the workflow execution.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError("Workflow must implement run method")

    async def _execute_activity_with_retry(
        self,
        activity_func: Any,
        arg: Any,
        start_to_close_timeout: timedelta | None = None,
        heartbeat_timeout: timedelta | None = None,
        retry_policy: RetryPolicy | None = None,
        task_queue: str | None = None,
    ) -> Any:
        """Execute an activity with standardized retry configuration.

        Args:
            activity_func (Any): The activity function to execute.
            arg (Any): Argument to pass to the activity.
            start_to_close_timeout (timedelta): Maximum execution time for the activity.
                Defaults to 30 seconds.
            heartbeat_timeout (timedelta, optional): Heartbeat timeout for long-running activities.
                Defaults to None.
            retry_policy (RetryPolicy, optional): Custom retry policy.
                If None, uses default retry policy. Defaults to None.
            task_queue (str, optional): Task queue for activity execution.
                If None, uses workflow's task queue. Defaults to None.

        Returns:
            Any: The result of the activity execution.
        """
        # Read the host process's config: the import below is passed through the sandbox (see module imports).
        temporal_config = BaseConfig.global_config().TEMPORAL

        # Use config defaults if not provided
        if start_to_close_timeout is None:
            start_to_close_timeout = timedelta(seconds=temporal_config.ACTIVITY_START_TO_CLOSE_TIMEOUT)
        if heartbeat_timeout is None and temporal_config.ACTIVITY_HEARTBEAT_TIMEOUT is not None:
            heartbeat_timeout = timedelta(seconds=temporal_config.ACTIVITY_HEARTBEAT_TIMEOUT)
        if retry_policy is None:
            retry_policy = RetryPolicy(
                initial_interval=timedelta(seconds=temporal_config.RETRY_INITIAL_INTERVAL),
                maximum_attempts=temporal_config.RETRY_MAXIMUM_ATTEMPTS,
                backoff_coefficient=temporal_config.RETRY_BACKOFF_COEFFICIENT,
                maximum_interval=timedelta(seconds=temporal_config.RETRY_MAXIMUM_INTERVAL),
                non_retryable_error_types=temporal_config.RETRY_NON_RETRYABLE_ERROR_TYPES,
            )

        return await workflow.execute_activity(
            activity_func,
            arg,
            start_to_close_timeout=start_to_close_timeout,
            heartbeat_timeout=heartbeat_timeout,
            retry_policy=retry_policy,
            task_queue=task_queue,
        )

    async def _execute_child_workflow(
        self,
        child_workflow: Any,
        arg: Any,
        workflow_id: str | None = None,
        task_queue: str | None = None,
        execution_timeout: timedelta | None = None,
    ) -> Any:
        """Execute a child workflow with standardized configuration.

        Args:
            child_workflow (Any): The child workflow function to execute.
            arg (Any): Argument to pass to the child workflow.
            workflow_id (str, optional): Unique ID for the child workflow.
                If None, auto-generated. Defaults to None.
            task_queue (str, optional): Task queue for child workflow execution.
                If None, uses parent workflow's task queue. Defaults to None.
            execution_timeout (timedelta, optional): Maximum execution time for the child workflow.
                If None, uses default timeout. Defaults to None.

        Returns:
            Any: The result of the child workflow execution.
        """
        # Read the host process's config: the import below is passed through the sandbox (see module imports).
        temporal_config = BaseConfig.global_config().TEMPORAL

        # Use config defaults if not provided
        if execution_timeout is None and temporal_config.WORKFLOW_EXECUTION_TIMEOUT is not None:
            execution_timeout = timedelta(seconds=temporal_config.WORKFLOW_EXECUTION_TIMEOUT)

        return await workflow.execute_child_workflow(
            child_workflow,
            arg,
            id=workflow_id,
            task_queue=task_queue,
            execution_timeout=execution_timeout,
        )

    def _log_workflow_event(self, event: str, details: dict[str, Any] | None = None) -> None:
        """Log workflow events with consistent formatting.

        Args:
            event (str): The event description.
            details (dict[str, Any], optional): Additional event details.
                Defaults to None.
        """
        log_data = {
            "workflow_id": workflow.info().workflow_id,
            "workflow_type": workflow.info().workflow_type,
            "event": event,
        }

        if details:
            log_data.update(details)

        workflow.logger.info("Workflow event", extra=log_data)

archipy.adapters.temporal.base.BaseWorkflow.run async

run(workflow_input: T) -> R

Main workflow execution method.

This method must be implemented by concrete workflow classes to define the workflow logic. It should orchestrate activities and child workflows to accomplish the business process.

Parameters:

Name	Type	Description	Default
workflow_input	T	The input data for the workflow.	required

Returns:

Name	Type	Description
R	R	The result of the workflow execution.

Raises:

Type	Description
NotImplementedError	If not implemented by the subclass.

Source code in packages/temporal/archipy/adapters/temporal/base.py

@workflow.run
async def run(self, workflow_input: T) -> R:
    """Main workflow execution method.

    This method must be implemented by concrete workflow classes to define
    the workflow logic. It should orchestrate activities and child workflows
    to accomplish the business process.

    Args:
        workflow_input (T): The input data for the workflow.

    Returns:
        R: The result of the workflow execution.

    Raises:
        NotImplementedError: If not implemented by the subclass.
    """
    raise NotImplementedError("Workflow must implement run method")

archipy.adapters.temporal.base.BaseActivity

Base class for all Temporal activities in ArchiPy.

Provides common functionality for activity implementations, including integration with your logic layer, standardized error handling, and execution hooks for cross-cutting concerns.

Class Type Parameters:

Name	Bound or Constraints	Description	Default
T		Type of the activity input parameter.	required
R		Type of the activity return value.	required

Source code in packages/temporal/archipy/adapters/temporal/base.py

class BaseActivity[T, R]:
    """Base class for all Temporal activities in ArchiPy.

    Provides common functionality for activity implementations, including
    integration with your logic layer, standardized error handling, and
    execution hooks for cross-cutting concerns.

    Type Parameters:
        T: Type of the activity input parameter.
        R: Type of the activity return value.
    """

    def __init__(self, logic: Any | None = None) -> None:
        """Initialize the activity with a logic instance.

        Args:
            logic (Any, optional): Your business logic instance (object) that contains
                a repository with access to adapters. If None, subclass should override _get_logic().
                Defaults to None.
        """
        self._logic = logic

    def _get_logic(self) -> Any:
        """Get the logic instance for this activity.

        Override this method in subclasses to provide your specific logic instance,
        or pass it via constructor using dependency injection. Your logic instance
        should have a repository that manages adapter access.

        Returns:
            Any: Your business logic instance.

        Raises:
            NotImplementedError: If not implemented by the subclass and no logic provided.
        """
        if self._logic is not None:
            return self._logic
        raise NotImplementedError("Activity must provide a logic instance via constructor or override _get_logic()")

    @activity.defn
    async def execute(self, activity_input: T) -> R:
        """Main activity execution method with hooks.

        This method provides a template for activity execution with pre/post hooks
        for common concerns like caching, validation, and monitoring.

        Args:
            activity_input (T): The input data for the activity.

        Returns:
            R: The result of the activity execution.

        Raises:
            Exception: Any exception that occurs during activity execution.
        """
        try:
            # Pre-execution hook
            await self._before_execute(activity_input)

            # Check cache if enabled
            if self._is_cacheable():
                cache_key = self._get_cache_key(activity_input)
                cached_result = await self._get_from_cache(cache_key)
                if cached_result is not None:
                    activity.logger.info("Using cached result", extra={"cache_key": cache_key})
                    return cached_result

            # Main business logic execution
            result = await self._do_execute(activity_input)

            # Cache result if enabled
            if self._is_cacheable() and result is not None:
                cache_key = self._get_cache_key(activity_input)
                await self._store_in_cache(cache_key, result)

            # Post-execution hook
            await self._after_execute(activity_input, result)

        except Exception as error:
            await self._handle_error(activity_input, error)
            if self._is_non_retryable(error):
                # Retrying cannot fix a client-side error, so fail the activity at once; the code names the cause
                # without carrying the (possibly sensitive) error message.
                raise ApplicationError(
                    type(error).__name__,
                    type=getattr(error, "code", type(error).__name__),
                    non_retryable=True,
                ) from error
            raise
        else:
            return result

    @abstractmethod
    async def _do_execute(self, activity_input: T) -> R:
        """Execute the main activity business logic.

        This method must be implemented by concrete activity classes to define
        the specific business logic for the activity.

        Args:
            activity_input (T): The input data for the activity.

        Returns:
            R: The result of the activity execution.

        Raises:
            NotImplementedError: If not implemented by the subclass.
        """
        raise NotImplementedError("Activity must implement _do_execute method")

    async def _before_execute(self, activity_input: T) -> None:
        """Pre-execution hook for common setup tasks.

        Override this method to perform tasks before the main activity logic,
        such as validation, setup, or preparation.

        Args:
            activity_input (T): The input data for the activity.
        """
        self._log_activity_event("execution_started", {"input_type": type(activity_input).__name__})

    async def _after_execute(self, activity_input: T, result: R) -> None:
        """Post-execution hook for cleanup and monitoring.

        Override this method to perform tasks after successful activity execution,
        such as cleanup, metrics emission, or notifications.

        Args:
            activity_input (T): The input data that was processed.
            result (R): The result of the activity execution.
        """
        self._log_activity_event(
            "execution_completed",
            {
                "input_type": type(activity_input).__name__,
                "result_type": type(result).__name__,
            },
        )

    async def _handle_error(self, activity_input: T, error: Exception) -> None:
        """Handle activity execution errors.

        Override this method to implement custom error handling logic,
        such as error reporting, cleanup, or compensation actions.

        Args:
            activity_input (T): The input data that was being processed.
            error (Exception): The exception that occurred.
        """
        # The error type only: messages can carry user data, and the activity failure keeps the full error.
        self._log_activity_event(
            "execution_failed",
            {
                "input_type": type(activity_input).__name__,
                "error_type": type(error).__name__,
            },
        )

    def _is_non_retryable(self, error: Exception) -> bool:
        """Decide whether an activity error should fail the activity without retries.

        By default ArchiPy errors with a 4xx HTTP status (validation, not found, permission, ...) are non-retryable,
        since retrying the same input cannot succeed. Override to change the rule.

        Args:
            error (Exception): The error raised by the activity.

        Returns:
            bool: True to fail the activity at once, False to let the retry policy apply.
        """
        return isinstance(error, BaseError) and HTTP_CLIENT_ERROR_MIN <= error.http_status < HTTP_SERVER_ERROR_MIN

    def _is_cacheable(self) -> bool:
        """Determine if this activity's results should be cached.

        Override this method to enable caching for specific activities.

        Returns:
            bool: True if results should be cached, False otherwise.
        """
        return False

    def _get_cache_key(self, activity_input: T) -> str:
        """Generate a cache key for the given input.

        Override this method to customize cache key generation for activities
        that support caching.

        Args:
            activity_input (T): The activity input data.

        Returns:
            str: The cache key for storing/retrieving results.
        """
        # ``hash()`` is salted per process, so workers would never share a key; a content digest is stable.
        if isinstance(activity_input, BaseModel):
            payload = activity_input.model_dump_json()
        else:
            payload = json.dumps(activity_input, sort_keys=True, default=str)
        digest = hashlib.sha256(payload.encode()).hexdigest()
        return f"{self.__class__.__name__}:{digest}"

    async def _get_from_cache(self, cache_key: str) -> R | None:
        """Retrieve a result from cache using your logic instance.

        Override this method to implement caching using your repository pattern.
        By default, this returns None (no caching).

        Args:
            cache_key (str): The cache key to look up.

        Returns:
            R | None: The cached result if found, None otherwise.
        """
        # Override this method to use your logic instance's repository
        return None

    async def _store_in_cache(self, cache_key: str, result: R, ttl: int = 3600) -> None:
        """Store a result in cache using your logic instance.

        Override this method to implement caching using your repository pattern.
        By default, this does nothing.

        Args:
            cache_key (str): The cache key for storage.
            result (R): The result to cache.
            ttl (int): Time-to-live in seconds. Defaults to 3600 (1 hour).
        """
        # Override this method to use your logic instance's repository

    def _log_activity_event(self, event: str, details: dict[str, Any] | None = None) -> None:
        """Log activity events with consistent formatting.

        Args:
            event (str): The event description.
            details (dict[str, Any], optional): Additional event details.
                Defaults to None.
        """
        log_data = {
            "activity_type": self.__class__.__name__,
            "event": event,
        }

        if details:
            log_data.update(details)

        activity.logger.info("Activity event", extra=log_data)

archipy.adapters.temporal.base.BaseActivity.execute async

execute(activity_input: T) -> R

Main activity execution method with hooks.

This method provides a template for activity execution with pre/post hooks for common concerns like caching, validation, and monitoring.

Parameters:

Name	Type	Description	Default
activity_input	T	The input data for the activity.	required

Returns:

Name	Type	Description
R	R	The result of the activity execution.

Raises:

Type	Description
Exception	Any exception that occurs during activity execution.

Source code in packages/temporal/archipy/adapters/temporal/base.py

@activity.defn
async def execute(self, activity_input: T) -> R:
    """Main activity execution method with hooks.

    This method provides a template for activity execution with pre/post hooks
    for common concerns like caching, validation, and monitoring.

    Args:
        activity_input (T): The input data for the activity.

    Returns:
        R: The result of the activity execution.

    Raises:
        Exception: Any exception that occurs during activity execution.
    """
    try:
        # Pre-execution hook
        await self._before_execute(activity_input)

        # Check cache if enabled
        if self._is_cacheable():
            cache_key = self._get_cache_key(activity_input)
            cached_result = await self._get_from_cache(cache_key)
            if cached_result is not None:
                activity.logger.info("Using cached result", extra={"cache_key": cache_key})
                return cached_result

        # Main business logic execution
        result = await self._do_execute(activity_input)

        # Cache result if enabled
        if self._is_cacheable() and result is not None:
            cache_key = self._get_cache_key(activity_input)
            await self._store_in_cache(cache_key, result)

        # Post-execution hook
        await self._after_execute(activity_input, result)

    except Exception as error:
        await self._handle_error(activity_input, error)
        if self._is_non_retryable(error):
            # Retrying cannot fix a client-side error, so fail the activity at once; the code names the cause
            # without carrying the (possibly sensitive) error message.
            raise ApplicationError(
                type(error).__name__,
                type=getattr(error, "code", type(error).__name__),
                non_retryable=True,
            ) from error
        raise
    else:
        return result

archipy.adapters.temporal.base.LogicIntegratedActivity

Bases: BaseActivity[T, R]

Activity base class that enforces the logic layer pattern.

This class provides helper methods that delegate to your logic instance, ensuring all business operations go through your established architecture with a single repository managing adapter access.

Source code in packages/temporal/archipy/adapters/temporal/base.py

class LogicIntegratedActivity(BaseActivity[T, R]):
    """Activity base class that enforces the logic layer pattern.

    This class provides helper methods that delegate to your logic instance,
    ensuring all business operations go through your established architecture
    with a single repository managing adapter access.
    """

    async def _execute_with_logic(self, operation_func: str, *args: Any, **kwargs: Any) -> Any:
        """Execute an operation using your logic instance.

        This is a convenience method to call methods on your logic instance.

        Args:
            operation_func (str): The name of the method to call on your logic instance.
            *args (Any): Arguments to pass to the logic method.
            **kwargs (Any): Keyword arguments to pass to the logic method.

        Returns:
            Any: Result of the logic operation.

        Example:
            # Call logic.get_user_by_id(user_id)
            user = await self._execute_with_logic("get_user_by_id", user_id)
        """
        logic = self._get_logic()
        method = getattr(logic, operation_func)
        return await method(*args, **kwargs)

    async def _execute_with_atomic(self, operation_func: str, *args: Any, **kwargs: Any) -> Any:
        """Execute an operation using your logic instance method decorated with @atomic.

        This method assumes your logic methods are decorated with atomic decorators
        for transaction management.

        Args:
            operation_func (str): The name of the method to call on your logic instance.
                This method should be decorated with @atomic for transaction support.
            *args (Any): Arguments to pass to the logic method.
            **kwargs (Any): Keyword arguments to pass to the logic method.

        Returns:
            Any: Result of the atomic operation.

        Example:
            # Call logic.process_order(order_data) - method decorated with @atomic
            result = await self._execute_with_atomic("process_order", order_data)
        """
        logic = self._get_logic()
        method = getattr(logic, operation_func)
        # Method should be decorated with @atomic, so transaction is handled automatically
        return await method(*args, **kwargs)

    async def _call_atomic_method(self, method_name: str, *args: Any, **kwargs: Any) -> Any:
        """Call an atomic method directly on your logic instance.

        This is an alias for _execute_with_atomic for clearer semantic meaning.

        Args:
            method_name (str): The name of the atomic method to call.
            *args (Any): Arguments to pass to the method.
            **kwargs (Any): Keyword arguments to pass to the method.

        Returns:
            Any: Result of the atomic method call.

        Example:
            # Direct call to @atomic decorated method
            result = await self._call_atomic_method("create_order_with_payment", order_data)
        """
        return await self._execute_with_atomic(method_name, *args, **kwargs)

archipy.adapters.temporal.base.LogicIntegratedActivity.execute async

execute(activity_input: T) -> R

Main activity execution method with hooks.

This method provides a template for activity execution with pre/post hooks for common concerns like caching, validation, and monitoring.

Parameters:

Name	Type	Description	Default
activity_input	T	The input data for the activity.	required

Returns:

Name	Type	Description
R	R	The result of the activity execution.

Raises:

Type	Description
Exception	Any exception that occurs during activity execution.

Source code in packages/temporal/archipy/adapters/temporal/base.py

@activity.defn
async def execute(self, activity_input: T) -> R:
    """Main activity execution method with hooks.

    This method provides a template for activity execution with pre/post hooks
    for common concerns like caching, validation, and monitoring.

    Args:
        activity_input (T): The input data for the activity.

    Returns:
        R: The result of the activity execution.

    Raises:
        Exception: Any exception that occurs during activity execution.
    """
    try:
        # Pre-execution hook
        await self._before_execute(activity_input)

        # Check cache if enabled
        if self._is_cacheable():
            cache_key = self._get_cache_key(activity_input)
            cached_result = await self._get_from_cache(cache_key)
            if cached_result is not None:
                activity.logger.info("Using cached result", extra={"cache_key": cache_key})
                return cached_result

        # Main business logic execution
        result = await self._do_execute(activity_input)

        # Cache result if enabled
        if self._is_cacheable() and result is not None:
            cache_key = self._get_cache_key(activity_input)
            await self._store_in_cache(cache_key, result)

        # Post-execution hook
        await self._after_execute(activity_input, result)

    except Exception as error:
        await self._handle_error(activity_input, error)
        if self._is_non_retryable(error):
            # Retrying cannot fix a client-side error, so fail the activity at once; the code names the cause
            # without carrying the (possibly sensitive) error message.
            raise ApplicationError(
                type(error).__name__,
                type=getattr(error, "code", type(error).__name__),
                non_retryable=True,
            ) from error
        raise
    else:
        return result

archipy.adapters.temporal.base.AtomicActivity

Bases: BaseActivity[T, R]

Activity base class with built-in atomic transaction support.

This class extends BaseActivity to provide direct atomic transaction support within activity execution, ensuring database consistency during activity operations.

Parameters:

Name	Type	Description	Default
logic	Any	Business logic instance with repository access. Defaults to None.	None
db_type	str	Database type for atomic operations. Defaults to "postgres".	'postgres'

Source code in packages/temporal/archipy/adapters/temporal/base.py

class AtomicActivity(BaseActivity[T, R]):
    """Activity base class with built-in atomic transaction support.

    This class extends BaseActivity to provide direct atomic transaction support
    within activity execution, ensuring database consistency during activity operations.

    Args:
        logic (Any, optional): Business logic instance with repository access. Defaults to None.
        db_type (str, optional): Database type for atomic operations. Defaults to "postgres".
    """

    def __init__(self, logic: Any | None = None, db_type: str = "postgres") -> None:
        """Initialize the atomic activity.

        Args:
            logic (Any, optional): Business logic instance. Defaults to None.
            db_type (str): Database type ("postgres", "sqlite", "starrocks"). Defaults to "postgres".

        Raises:
            ValueError: If an invalid db_type is provided.
        """
        super().__init__(logic)
        self._validate_db_type(db_type)
        self.db_type = db_type

    @staticmethod
    def _validate_db_type(db_type: str) -> None:
        """Validate the database type used for atomic operations.

        Args:
            db_type (str): The database type.

        Raises:
            InvalidArgumentError: If the database type is not supported.
        """
        if db_type not in ("postgres", "sqlite", "starrocks"):
            raise InvalidArgumentError(
                argument_name="db_type",
                additional_data={
                    "got": db_type,
                    "allowed": ["postgres", "sqlite", "starrocks"],
                },
            )

    def _get_atomic_decorator(self) -> Callable[..., Any]:
        """Get the appropriate async atomic decorator for the configured database type.

        Returns:
            Callable: The async atomic decorator function for the configured database.

        Raises:
            ImportError: If SQLAlchemy is not installed and atomic decorators are needed.
        """
        # Lazy import to avoid requiring SQLAlchemy when using temporalio without sqlalchemy extra
        from archipy.helpers.decorators import (
            postgres_sqlalchemy_atomic,
            sqlite_sqlalchemy_atomic,
            starrocks_sqlalchemy_atomic,
        )

        decorators_map = {
            "postgres": postgres_sqlalchemy_atomic,
            "sqlite": sqlite_sqlalchemy_atomic,
            "starrocks": starrocks_sqlalchemy_atomic,
        }
        return decorators_map[self.db_type]

    @activity.defn
    async def execute_atomic(self, activity_input: T) -> R:
        """Execute the activity within a database transaction.

        This method wraps the entire activity execution (including pre/post hooks)
        within a database transaction, ensuring atomicity of all database operations.

        Args:
            activity_input (T): The input data for the activity.

        Returns:
            R: The result of the activity execution.

        Raises:
            Exception: Any exception that occurs during activity execution.
        """
        atomic_decorator = self._get_atomic_decorator()

        @atomic_decorator
        async def _atomic_execute() -> R:
            return await super(AtomicActivity, self).execute(activity_input)

        return await _atomic_execute()

    async def _do_execute_atomic(self, activity_input: T) -> R:
        """Execute main business logic within a database transaction.

        This method provides atomic transaction support for the core business logic only,
        excluding pre/post hooks from the transaction scope.

        Args:
            activity_input (T): The input data for the activity.

        Returns:
            R: The result of the business logic execution.
        """
        atomic_decorator = self._get_atomic_decorator()

        @atomic_decorator
        async def _atomic_do_execute() -> R:
            return await self._do_execute(activity_input)

        return await _atomic_do_execute()

    async def execute_custom_atomic_operation(self, operation: Callable[[], Any]) -> Any:
        """Execute a custom operation within a database transaction.

        This utility method allows executing custom logic within the activity's
        configured atomic transaction context.

        Args:
            operation (Callable[[], T]): The operation to execute atomically.

        Returns:
            Any: The result of the operation.

        Example:
            ```python
            async def custom_logic():
                # Custom database operations
                return await some_database_work()


            result = await self.execute_custom_atomic_operation(custom_logic)
            ```
        """
        atomic_decorator = self._get_atomic_decorator()

        @atomic_decorator
        async def _execute_operation() -> Any:
            return await operation()

        return await _execute_operation()

    def with_db_type(self, db_type: str) -> AtomicActivity[T, R]:
        """Create a copy of this activity that runs its transactions on another database type.

        Args:
            db_type (str): The new database type ("postgres", "sqlite", or "starrocks").

        Returns:
            AtomicActivity[T, R]: A shallow copy with the new database type; this instance is unchanged.

        Raises:
            InvalidArgumentError: If an invalid db_type is provided.
        """
        self._validate_db_type(db_type)
        clone = copy.copy(self)
        clone.db_type = db_type
        return clone

archipy.adapters.temporal.base.AtomicActivity.db_type instance-attribute

db_type = db_type

archipy.adapters.temporal.base.AtomicActivity.execute_atomic async

execute_atomic(activity_input: T) -> R

Execute the activity within a database transaction.

This method wraps the entire activity execution (including pre/post hooks) within a database transaction, ensuring atomicity of all database operations.

Parameters:

Name	Type	Description	Default
activity_input	T	The input data for the activity.	required

Returns:

Name	Type	Description
R	R	The result of the activity execution.

Raises:

Type	Description
Exception	Any exception that occurs during activity execution.

Source code in packages/temporal/archipy/adapters/temporal/base.py

@activity.defn
async def execute_atomic(self, activity_input: T) -> R:
    """Execute the activity within a database transaction.

    This method wraps the entire activity execution (including pre/post hooks)
    within a database transaction, ensuring atomicity of all database operations.

    Args:
        activity_input (T): The input data for the activity.

    Returns:
        R: The result of the activity execution.

    Raises:
        Exception: Any exception that occurs during activity execution.
    """
    atomic_decorator = self._get_atomic_decorator()

    @atomic_decorator
    async def _atomic_execute() -> R:
        return await super(AtomicActivity, self).execute(activity_input)

    return await _atomic_execute()

archipy.adapters.temporal.base.AtomicActivity.execute_custom_atomic_operation async

execute_custom_atomic_operation(
    operation: Callable[[], Any],
) -> Any

Execute a custom operation within a database transaction.

This utility method allows executing custom logic within the activity's configured atomic transaction context.

Parameters:

Name	Type	Description	Default
operation	Callable[[], T]	The operation to execute atomically.	required

Returns:

Name	Type	Description
Any	Any	The result of the operation.

Example

async def custom_logic():
    # Custom database operations
    return await some_database_work()


result = await self.execute_custom_atomic_operation(custom_logic)

Source code in packages/temporal/archipy/adapters/temporal/base.py

async def execute_custom_atomic_operation(self, operation: Callable[[], Any]) -> Any:
    """Execute a custom operation within a database transaction.

    This utility method allows executing custom logic within the activity's
    configured atomic transaction context.

    Args:
        operation (Callable[[], T]): The operation to execute atomically.

    Returns:
        Any: The result of the operation.

    Example:
        ```python
        async def custom_logic():
            # Custom database operations
            return await some_database_work()


        result = await self.execute_custom_atomic_operation(custom_logic)
        ```
    """
    atomic_decorator = self._get_atomic_decorator()

    @atomic_decorator
    async def _execute_operation() -> Any:
        return await operation()

    return await _execute_operation()

archipy.adapters.temporal.base.AtomicActivity.with_db_type

with_db_type(db_type: str) -> AtomicActivity[T, R]

Create a copy of this activity that runs its transactions on another database type.

Parameters:

Name	Type	Description	Default
db_type	str	The new database type ("postgres", "sqlite", or "starrocks").	required

Returns:

Type	Description
AtomicActivity[T, R]	AtomicActivity[T, R]: A shallow copy with the new database type; this instance is unchanged.

Raises:

Type	Description
InvalidArgumentError	If an invalid db_type is provided.

Source code in packages/temporal/archipy/adapters/temporal/base.py

def with_db_type(self, db_type: str) -> AtomicActivity[T, R]:
    """Create a copy of this activity that runs its transactions on another database type.

    Args:
        db_type (str): The new database type ("postgres", "sqlite", or "starrocks").

    Returns:
        AtomicActivity[T, R]: A shallow copy with the new database type; this instance is unchanged.

    Raises:
        InvalidArgumentError: If an invalid db_type is provided.
    """
    self._validate_db_type(db_type)
    clone = copy.copy(self)
    clone.db_type = db_type
    return clone

archipy.adapters.temporal.base.AtomicActivity.execute async

execute(activity_input: T) -> R

Main activity execution method with hooks.

This method provides a template for activity execution with pre/post hooks for common concerns like caching, validation, and monitoring.

Parameters:

Name	Type	Description	Default
activity_input	T	The input data for the activity.	required

Returns:

Name	Type	Description
R	R	The result of the activity execution.

Raises:

Type	Description
Exception	Any exception that occurs during activity execution.

Source code in packages/temporal/archipy/adapters/temporal/base.py

@activity.defn
async def execute(self, activity_input: T) -> R:
    """Main activity execution method with hooks.

    This method provides a template for activity execution with pre/post hooks
    for common concerns like caching, validation, and monitoring.

    Args:
        activity_input (T): The input data for the activity.

    Returns:
        R: The result of the activity execution.

    Raises:
        Exception: Any exception that occurs during activity execution.
    """
    try:
        # Pre-execution hook
        await self._before_execute(activity_input)

        # Check cache if enabled
        if self._is_cacheable():
            cache_key = self._get_cache_key(activity_input)
            cached_result = await self._get_from_cache(cache_key)
            if cached_result is not None:
                activity.logger.info("Using cached result", extra={"cache_key": cache_key})
                return cached_result

        # Main business logic execution
        result = await self._do_execute(activity_input)

        # Cache result if enabled
        if self._is_cacheable() and result is not None:
            cache_key = self._get_cache_key(activity_input)
            await self._store_in_cache(cache_key, result)

        # Post-execution hook
        await self._after_execute(activity_input, result)

    except Exception as error:
        await self._handle_error(activity_input, error)
        if self._is_non_retryable(error):
            # Retrying cannot fix a client-side error, so fail the activity at once; the code names the cause
            # without carrying the (possibly sensitive) error message.
            raise ApplicationError(
                type(error).__name__,
                type=getattr(error, "code", type(error).__name__),
                non_retryable=True,
            ) from error
        raise
    else:
        return result

options: show_root_toc_entry: false heading_level: 3

Adapters

Concrete Temporal adapter implementing workflow and activity client operations.

Temporal adapter implementation for workflow orchestration.

This module provides concrete implementations of the Temporal port interfaces, integrating with the Temporal workflow engine while following ArchiPy patterns and conventions.

archipy.adapters.temporal.adapters.logger module-attribute

logger = logging.getLogger(__name__)

archipy.adapters.temporal.adapters.T module-attribute

T = TypeVar('T')

archipy.adapters.temporal.adapters.TemporalAdapter

Bases: TemporalPort

Temporal workflow adapter implementing the TemporalPort interface.

This adapter provides a standardized interface for interacting with Temporal workflow orchestration services, following ArchiPy architecture patterns. It handles client connections, TLS configuration, and workflow lifecycle management.

Parameters:

Name	Type	Description	Default
temporal_config	TemporalConfig	Configuration settings for Temporal. If None, retrieves from global config. Defaults to None.	None

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

class TemporalAdapter(TemporalPort):
    """Temporal workflow adapter implementing the TemporalPort interface.

    This adapter provides a standardized interface for interacting with Temporal
    workflow orchestration services, following ArchiPy architecture patterns.
    It handles client connections, TLS configuration, and workflow lifecycle
    management.

    Args:
        temporal_config (TemporalConfig, optional): Configuration settings for Temporal.
            If None, retrieves from global config. Defaults to None.
    """

    def __init__(
        self,
        temporal_config: TemporalConfig | None = None,
        data_converter: DataConverter | None = None,
        interceptors: Sequence[Interceptor] | None = None,
    ) -> None:
        """Initialize the TemporalAdapter with configuration settings.

        Args:
            temporal_config (TemporalConfig, optional): Configuration settings for Temporal.
                If None, retrieves from global config. Defaults to None.
            data_converter (DataConverter, optional): Data converter for payloads, e.g.
                ``dataclasses.replace(pydantic_data_converter, payload_codec=MyCodec())`` to encrypt or compress them.
                If None, the Pydantic converter is used when ``PYDANTIC_DATA_CONVERTER`` is enabled. Defaults to None.
            interceptors (Sequence[Interceptor], optional): Client interceptors. Those that also implement
                ``temporalio.worker.Interceptor`` apply to workers created from this client. Defaults to None.
        """
        # Get temporal config from the global config or use provided one
        if temporal_config is None:
            global_config = BaseConfig.global_config()
            if hasattr(global_config, "TEMPORAL"):
                self.config = global_config.TEMPORAL
            else:
                # Create a default config if none exists
                from archipy.configs.config_template import TemporalConfig

                self.config = TemporalConfig()
        else:
            self.config = temporal_config
        self._data_converter = data_converter
        self._interceptors = list(interceptors or [])
        self._client: Client | None = None
        self._client_lock = asyncio.Lock()

    async def get_client(self) -> Client:
        """Get or create the Temporal client connection.

        Returns:
            Client: The Temporal client instance.

        Raises:
            ConnectionError: If unable to connect to Temporal server.
        """
        if self._client is not None:
            return self._client
        # Concurrent first calls must share one connection.
        async with self._client_lock:
            if self._client is None:
                self._client = await self._connect()
        return self._client

    async def _connect(self) -> Client:
        """Connect a new Temporal client from the configuration.

        Returns:
            Client: The connected client.

        Raises:
            BaseError: If unable to connect to the Temporal server.
        """
        try:
            return await Client.connect(f"{self.config.HOST}:{self.config.PORT}", **self._build_connect_kwargs())
        except Exception as error:
            raise WorkerConnectionError(
                additional_data={
                    "server": f"{self.config.HOST}:{self.config.PORT}",
                    "namespace": self.config.NAMESPACE,
                    "original_error": str(error),
                },
            ) from error

    def _optional_connect_kwargs(self) -> dict[str, Any]:
        """Build the ``Client.connect`` keyword arguments that are only passed when configured.

        Returns:
            dict[str, Any]: The configured optional connection keyword arguments; unset ones are left out so the SDK
                defaults apply.
        """
        data_converter = self._data_converter
        if data_converter is None and self.config.PYDANTIC_DATA_CONVERTER:
            data_converter = pydantic_data_converter
        optional: dict[str, Any] = {
            "data_converter": data_converter,
            "interceptors": self._interceptors or None,
            "identity": self.config.CLIENT_IDENTITY,
            "api_key": self.config.API_KEY,
            "rpc_metadata": self.config.RPC_METADATA or None,
            "tls": self._tls_option(),
            "connect_timeout": (
                timedelta(seconds=self.config.CONNECT_TIMEOUT) if self.config.CONNECT_TIMEOUT is not None else None
            ),
            "grpc_compression": self._grpc_compression(),
            "payload_limits": self._payload_limits(),
        }
        return {key: value for key, value in optional.items() if value is not None}

    def _grpc_compression(self) -> GrpcCompression | None:
        """Resolve the ``grpc_compression`` argument of ``Client.connect`` from ``GRPC_COMPRESSION``.

        Returns:
            GrpcCompression | None: The configured compression, or None to keep the SDK default.
        """
        match self.config.GRPC_COMPRESSION:
            case "gzip":
                return GrpcCompression.GZIP
            case "none":
                return GrpcCompression.NONE
            case _:
                return None

    def _payload_limits(self) -> PayloadLimitsConfig | None:
        """Resolve the ``payload_limits`` argument of ``Client.connect`` from the warning size settings.

        Returns:
            PayloadLimitsConfig | None: The limits when either size is set, the other keeping the SDK default;
                None when neither is set.
        """
        if self.config.PAYLOADS_WARN_SIZE is None and self.config.MEMO_WARN_SIZE is None:
            return None
        defaults = PayloadLimitsConfig()
        return PayloadLimitsConfig(
            payloads_warn_size=self.config.PAYLOADS_WARN_SIZE or defaults.payloads_warn_size,
            memo_warn_size=self.config.MEMO_WARN_SIZE or defaults.memo_warn_size,
        )

    def _build_connect_kwargs(self) -> dict[str, Any]:
        """Build the ``Client.connect`` keyword arguments, only including optional settings when configured.

        Returns:
            dict[str, Any]: The connection keyword arguments.
        """
        connect_kwargs: dict[str, Any] = {
            "namespace": self.config.NAMESPACE,
            "lazy": self.config.LAZY_CONNECT,
            "keep_alive_config": KeepAliveConfig(
                interval_millis=self.config.KEEP_ALIVE_INTERVAL_MS,
                timeout_millis=self.config.KEEP_ALIVE_TIMEOUT_MS,
            ),
            "retry_config": RetryConfig(
                initial_interval_millis=self.config.CLIENT_RPC_RETRY_INITIAL_INTERVAL_MS,
                max_interval_millis=self.config.CLIENT_RPC_RETRY_MAX_INTERVAL_MS,
                max_retries=self.config.CLIENT_RPC_RETRY_MAX_RETRIES,
            ),
        }
        connect_kwargs.update(self._optional_connect_kwargs())

        # Configure Runtime with OTLP metrics if Temporal + OTel metrics are enabled, or a plain Runtime when only
        # environment info is disabled
        global_config = BaseConfig.global_config()
        otel = global_config.OTEL
        runtime: Runtime | None = None
        if self.config.ENABLE_METRICS and otel.IS_ENABLED and otel.METRICS_ENABLED:
            runtime = self._build_metrics_runtime(otel)
        elif self.config.RUNTIME_DISABLE_ENVIRONMENT_INFO:
            runtime = TemporalRuntimeManager().get_runtime(
                global_tags=self.config.METRICS_GLOBAL_TAGS,
                core_log_level=self.config.SDK_CORE_LOG_LEVEL,
                disable_environment_info=True,
            )
        if runtime is not None:
            connect_kwargs["runtime"] = runtime

        # Attach OTel tracing plugin when traces are enabled
        if otel.IS_ENABLED and otel.TRACES_ENABLED:
            from archipy.helpers.utils.otel_utils import OtelUtils

            OtelUtils.init_otel_if_needed(global_config)
            self._append_tracing_plugin(connect_kwargs)
        return connect_kwargs

    def _build_metrics_runtime(self, otel: OpentelemetryConfig) -> Runtime | None:
        """Build the Temporal Runtime that exports SDK metrics according to the OTel exporter mode.

        The Temporal SDK core can only export over OTLP or serve its own Prometheus scrape endpoint, so
        ``METRICS_EXPORTER=pull`` selects the scrape endpoint and every other mode uses OTLP.

        Args:
            otel: The global OpenTelemetry configuration.

        Returns:
            Runtime | None: The configured Runtime, or None when it could not be created.
        """
        from archipy.helpers.utils.otel_utils import OtelUtils

        otlp_endpoint = OtelUtils.resolve_metrics_endpoint(otel)
        prometheus_bind_address: str | None = None
        if otel.METRICS_EXPORTER == OtelMetricsExporter.PULL:
            prometheus_bind_address = f"{otel.METRICS_PULL_HOST}:{self.config.METRICS_PULL_PORT}"
        elif otel.METRICS_EXPORTER == OtelMetricsExporter.PUSHGATEWAY:
            logger.warning(
                "Temporal SDK metrics cannot be pushed to a Pushgateway; they are exported over OTLP to %s instead",
                otlp_endpoint,
            )

        return TemporalRuntimeManager().get_runtime(
            otel_metrics_enabled=True,
            otlp_endpoint=otlp_endpoint,
            headers=dict(otel.OTLP_HEADERS),
            use_http=(otel.PROTOCOL == "http/protobuf"),
            prometheus_bind_address=prometheus_bind_address,
            global_tags=self.config.METRICS_GLOBAL_TAGS,
            core_log_level=self.config.SDK_CORE_LOG_LEVEL,
            disable_environment_info=self.config.RUNTIME_DISABLE_ENVIRONMENT_INFO,
        )

    @staticmethod
    def _append_tracing_plugin(connect_kwargs: dict[str, Any]) -> None:
        """Append ArchiPy's Temporal OpenTelemetry plugin without discarding existing plugins.

        Skipped with a warning when the global tracer provider is not replay-safe.

        Args:
            connect_kwargs: Keyword arguments passed to ``Client.connect``.
        """
        from archipy.adapters.temporal.tracing import MessagingOpenTelemetryPlugin
        from archipy.helpers.utils.otel_utils import OtelUtils

        # The plugin needs a replay-safe provider; ArchiPy builds one itself, but a provider another library
        # registered first is adopted as is, and workers with workflows would then fail to start.
        if not isinstance(OtelUtils.tracer_provider(), ReplaySafeTracerProvider):
            logger.warning(
                "Temporal tracing is disabled: the global TracerProvider is not a temporalio ReplaySafeTracerProvider",
            )
            return
        connect_kwargs.setdefault("plugins", []).append(MessagingOpenTelemetryPlugin())

    def _has_tls_config(self) -> bool:
        """Check if any TLS file or server name is configured.

        Returns:
            bool: True if a CA certificate, a client certificate pair or a server name is set.
        """
        return any(
            [
                self.config.TLS_CA_CERT,
                self.config.TLS_CLIENT_CERT,
                self.config.TLS_SERVER_NAME,
            ],
        )

    def _tls_option(self) -> TLSConfig | bool | None:
        """Resolve the ``tls`` argument of ``Client.connect``.

        Returns:
            TLSConfig | bool | None: A TLS configuration when TLS settings are present (unless TLS is forced off),
                the ``TLS_ENABLED`` flag otherwise, or None to let the SDK enable TLS when an API key is set.
        """
        if self.config.TLS_ENABLED is False:
            return False
        if self._has_tls_config():
            return self._build_tls_config()
        return self.config.TLS_ENABLED

    def _build_tls_config(self) -> TLSConfig:
        """Build TLS configuration for secure connections from the configured files.

        Server-only TLS needs just the CA certificate (or none, to use the system CA bundle); mutual TLS adds the
        client certificate and key. ``TLS_VERIFICATION_SERVER_NAME`` checks the certificate name without changing
        the SNI that ``TLS_SERVER_NAME`` sets.

        Returns:
            TLSConfig: The TLS configuration object.

        Raises:
            InvalidArgumentError: If a configured TLS file cannot be read.
        """
        try:
            return TLSConfig(
                server_root_ca_cert=self._read_tls_file(self.config.TLS_CA_CERT),
                client_cert=self._read_tls_file(self.config.TLS_CLIENT_CERT),
                client_private_key=self._read_tls_file(self.config.TLS_CLIENT_KEY),
                domain=self.config.TLS_SERVER_NAME,
                verification_server_name=self.config.TLS_VERIFICATION_SERVER_NAME,
            )
        except OSError as error:
            raise InvalidArgumentError(additional_data={"original_error": str(error)}) from error

    @staticmethod
    def _read_tls_file(path: str | None) -> bytes | None:
        """Read a TLS file when a path is configured.

        Args:
            path (str | None): The file path.

        Returns:
            bytes | None: The file contents, or None when no path is set.
        """
        return Path(path).read_bytes() if path else None

    @staticmethod
    def _timeout(override: int | None, default: int | None) -> timedelta | None:
        """Resolve a timeout in seconds from a call override and its configured default.

        Args:
            override (int | None): The timeout passed to the call; 0 means no timeout, None uses the default.
            default (int | None): The configured timeout.

        Returns:
            timedelta | None: The timeout, or None when unbounded.
        """
        seconds = default if override is None else override
        return timedelta(seconds=seconds) if seconds else None

    @contextlib.asynccontextmanager
    async def _mapped_errors(self, operation: str, resource_id: str | None) -> AsyncGenerator[None]:
        """Map Temporal client errors raised in the block to ArchiPy domain errors.

        Workflow failures (``WorkflowFailureError``) are left as is: they carry the workflow's own error.

        Args:
            operation (str): The adapter operation, for error context.
            resource_id (str | None): The workflow or schedule ID the operation targets.

        Yields:
            None: Control to the wrapped block.

        Raises:
            AlreadyExistsError: If the workflow or schedule ID is already in use; for a workflow its
                ``additional_data`` holds the ``run_id`` and ``first_run_id`` of the running execution.
            NotFoundError: If the workflow or schedule does not exist.
            UnavailableError: If the Temporal server is unreachable.
            DeadlineExceededError: If the call timed out.
            TemporalError: For any other Temporal RPC error.
        """
        try:
            yield
        except WorkflowAlreadyStartedError as error:
            # The run IDs let callers reach the execution that already holds the workflow ID.
            raise AlreadyExistsError(
                resource_type=operation,
                additional_data={"id": resource_id, "run_id": error.run_id, "first_run_id": error.first_run_id},
            ) from error
        except ScheduleAlreadyRunningError as error:
            raise AlreadyExistsError(resource_type=operation, additional_data={"id": resource_id}) from error
        except RPCError as error:
            raise self._map_rpc_error(error, operation, resource_id) from error

    @staticmethod
    def _map_rpc_error(error: RPCError, operation: str, resource_id: str | None) -> BaseError:
        """Translate a Temporal RPC error into the matching ArchiPy error, without leaking the server message.

        Args:
            error (RPCError): The Temporal RPC error.
            operation (str): The adapter operation.
            resource_id (str | None): The workflow or schedule ID the operation targets.

        Returns:
            BaseError: The domain error to raise.
        """
        context = {"operation": operation, "id": resource_id, "status": error.status.name}
        match error.status:
            case RPCStatusCode.NOT_FOUND:
                return NotFoundError(resource_type=operation, additional_data=context)
            case RPCStatusCode.ALREADY_EXISTS:
                return AlreadyExistsError(resource_type=operation, additional_data=context)
            case RPCStatusCode.UNAVAILABLE:
                return UnavailableError(resource_type="temporal", additional_data=context)
            case RPCStatusCode.DEADLINE_EXCEEDED:
                return DeadlineExceededError(operation=operation, additional_data=context)
            case _:
                return TemporalError(additional_data=context)

    def _build_retry_policy(self) -> RetryPolicy:
        """Build default retry policy from configuration.

        Returns:
            RetryPolicy: The configured retry policy.
        """
        return RetryPolicy(
            initial_interval=timedelta(seconds=self.config.RETRY_INITIAL_INTERVAL),
            maximum_attempts=self.config.RETRY_MAXIMUM_ATTEMPTS,
            backoff_coefficient=self.config.RETRY_BACKOFF_COEFFICIENT,
            maximum_interval=timedelta(seconds=self.config.RETRY_MAXIMUM_INTERVAL),
            non_retryable_error_types=self.config.RETRY_NON_RETRYABLE_ERROR_TYPES,
        )

    def _build_workflow_retry_policy(self) -> RetryPolicy | None:
        """Build the workflow-level retry policy, only when enabled in configuration.

        A workflow retry reruns the whole workflow from the start, so it is opt-in.

        Returns:
            RetryPolicy | None: The configured retry policy, or None when workflow retries are disabled.
        """
        if not self.config.WORKFLOW_RETRY_ENABLED:
            return None
        return self._build_retry_policy()

    @staticmethod
    def _id_policies(
        id_reuse_policy: WorkflowIDReusePolicy | None,
        id_conflict_policy: WorkflowIDConflictPolicy | None,
    ) -> dict[str, Any]:
        """Build the workflow ID policy keyword arguments, leaving out unset ones so the SDK defaults apply.

        Args:
            id_reuse_policy (WorkflowIDReusePolicy | None): The ID reuse policy.
            id_conflict_policy (WorkflowIDConflictPolicy | None): The ID conflict policy.

        Returns:
            dict[str, Any]: The keyword arguments for ``start_workflow`` / ``execute_workflow``.
        """
        policies: dict[str, Any] = {}
        if id_reuse_policy is not None:
            policies["id_reuse_policy"] = id_reuse_policy
        if id_conflict_policy is not None:
            policies["id_conflict_policy"] = id_conflict_policy
        return policies

    @override
    async def start_workflow(
        self,
        workflow: str | Callable[..., Any],
        arg: Any = None,
        workflow_id: str | None = None,
        task_queue: str | None = None,
        execution_timeout: int | None = None,
        run_timeout: int | None = None,
        task_timeout: int | None = None,
        memo: dict[str, Any] | None = None,
        search_attributes: dict[str, Any] | None = None,
        id_reuse_policy: WorkflowIDReusePolicy | None = None,
        id_conflict_policy: WorkflowIDConflictPolicy | None = None,
    ) -> WorkflowHandle[T, Any]:
        """Start a workflow execution asynchronously.

        Args:
            workflow (str | Callable): The workflow function or workflow type name.
            arg (Any, optional): Input argument for the workflow. Defaults to None.
            workflow_id (str, optional): Unique identifier for the workflow execution.
                If None, a UUID will be generated. Defaults to None.
            task_queue (str, optional): Task queue name for workflow execution.
                If None, uses the default task queue. Defaults to None.
            execution_timeout (int, optional): Maximum workflow execution time in seconds.
                Overrides config default. Defaults to None.
            run_timeout (int, optional): Maximum single workflow run time in seconds.
                Overrides config default. Defaults to None.
            task_timeout (int, optional): Maximum workflow task processing time in seconds.
                Overrides config default. Defaults to None.
            memo (dict[str, Any], optional): Non-indexed metadata for the workflow.
                Defaults to None.
            search_attributes (dict[str, Any], optional): Indexed metadata for workflow search. Datetime values must
                be timezone-aware; ``temporalio`` 1.35+ raises on naive ones. Defaults to None.
            id_reuse_policy (WorkflowIDReusePolicy, optional): Whether a closed workflow's ID may be reused.
                Defaults to None (server default: allow duplicate).
            id_conflict_policy (WorkflowIDConflictPolicy, optional): What to do when a workflow with the same ID
                is still running, e.g. ``USE_EXISTING`` for idempotent starts. Defaults to None (server default: fail).

        Returns:
            WorkflowHandle[T, Any]: Handle to the started workflow execution.

        Note:
            The generated ID comes from the standard library ``uuid.uuid7()``, which is fine here, outside workflow
            code. Inside a workflow the sandbox blocks it: use ``temporalio.workflow.uuid7()`` instead.
        """
        async with self._mapped_errors("start_workflow", workflow_id):
            client = await self.get_client()

            workflow_id = workflow_id or str(uuid7())
            task_queue = task_queue or self.config.TASK_QUEUE

            # Build positional args: only include arg if it's not None,
            # so workflows that take no parameters aren't given an extra argument.
            positional_args: list[Any] = [] if arg is None else [arg]

            return await client.start_workflow(
                workflow,
                *positional_args,
                id=workflow_id,
                task_queue=task_queue,
                execution_timeout=self._timeout(execution_timeout, self.config.WORKFLOW_EXECUTION_TIMEOUT),
                run_timeout=self._timeout(run_timeout, self.config.WORKFLOW_RUN_TIMEOUT),
                task_timeout=self._timeout(task_timeout, self.config.WORKFLOW_TASK_TIMEOUT),
                retry_policy=self._build_workflow_retry_policy(),
                memo=memo,
                search_attributes=search_attributes,
                **self._id_policies(id_reuse_policy, id_conflict_policy),
            )

    @override
    async def execute_workflow(
        self,
        workflow: str | Callable[..., Any],
        arg: Any = None,
        workflow_id: str | None = None,
        task_queue: str | None = None,
        execution_timeout: int | None = None,
        run_timeout: int | None = None,
        task_timeout: int | None = None,
        id_reuse_policy: WorkflowIDReusePolicy | None = None,
        id_conflict_policy: WorkflowIDConflictPolicy | None = None,
    ) -> T:
        """Execute a workflow and wait for its completion.

        Args:
            workflow (str | Callable): The workflow function or workflow type name.
            arg (Any, optional): Input argument for the workflow. Defaults to None.
            workflow_id (str, optional): Unique identifier for the workflow execution.
                If None, a UUID will be generated. Defaults to None.
            task_queue (str, optional): Task queue name for workflow execution.
                If None, uses the default task queue. Defaults to None.
            execution_timeout (int, optional): Maximum workflow execution time in seconds.
                Overrides config default. Defaults to None.
            run_timeout (int, optional): Maximum single workflow run time in seconds.
                Overrides config default. Defaults to None.
            task_timeout (int, optional): Maximum workflow task processing time in seconds.
                Overrides config default. Defaults to None.
            id_reuse_policy (WorkflowIDReusePolicy, optional): Whether a closed workflow's ID may be reused.
                Defaults to None (server default: allow duplicate).
            id_conflict_policy (WorkflowIDConflictPolicy, optional): What to do when a workflow with the same ID
                is still running, e.g. ``USE_EXISTING`` for idempotent starts. Defaults to None (server default: fail).

        Returns:
            T: The workflow execution result.
        """
        async with self._mapped_errors("execute_workflow", workflow_id):
            client = await self.get_client()

            workflow_id = workflow_id or str(uuid7())
            task_queue = task_queue or self.config.TASK_QUEUE

            # Build positional args: only include arg if it's not None,
            # so workflows that take no parameters aren't given an extra argument.
            positional_args: list[Any] = [] if arg is None else [arg]

            return await client.execute_workflow(
                workflow,
                *positional_args,
                id=workflow_id,
                task_queue=task_queue,
                execution_timeout=self._timeout(execution_timeout, self.config.WORKFLOW_EXECUTION_TIMEOUT),
                run_timeout=self._timeout(run_timeout, self.config.WORKFLOW_RUN_TIMEOUT),
                task_timeout=self._timeout(task_timeout, self.config.WORKFLOW_TASK_TIMEOUT),
                retry_policy=self._build_workflow_retry_policy(),
                **self._id_policies(id_reuse_policy, id_conflict_policy),
            )

    @override
    async def get_workflow_handle(self, workflow_id: str, run_id: str | None = None) -> WorkflowHandle[T, Any]:
        """Get a handle to an existing workflow execution.

        Args:
            workflow_id (str): The unique identifier of the workflow execution.
            run_id (str, optional): The specific run identifier within the workflow.
                If None, gets the latest run. Defaults to None.

        Returns:
            WorkflowHandle[T, Any]: Handle to the workflow execution.
        """
        async with self._mapped_errors("get_workflow_handle", workflow_id):
            client = await self.get_client()
            return client.get_workflow_handle(workflow_id, run_id=run_id)

    @override
    async def cancel_workflow(self, workflow_id: str, run_id: str | None = None, reason: str | None = None) -> None:
        """Cancel a running workflow execution.

        Args:
            workflow_id (str): The unique identifier of the workflow execution.
            run_id (str, optional): The specific run identifier within the workflow.
                If None, cancels the latest run. Defaults to None.
            reason (str, optional): Reason for cancellation. Defaults to None.
        """
        async with self._mapped_errors("cancel_workflow", workflow_id):
            handle = await self.get_workflow_handle(workflow_id, run_id)
            await handle.cancel(reason=reason or "")

    @override
    async def terminate_workflow(self, workflow_id: str, run_id: str | None = None, reason: str | None = None) -> None:
        """Terminate a running workflow execution immediately.

        Args:
            workflow_id (str): The unique identifier of the workflow execution.
            run_id (str, optional): The specific run identifier within the workflow.
                If None, terminates the latest run. Defaults to None.
            reason (str, optional): Reason for termination. Defaults to None.
        """
        async with self._mapped_errors("terminate_workflow", workflow_id):
            handle = await self.get_workflow_handle(workflow_id, run_id)
            await handle.terminate(reason=reason)

    @override
    async def signal_workflow(
        self,
        workflow_id: str,
        signal_name: str,
        arg: Any = None,
        run_id: str | None = None,
    ) -> None:
        """Send a signal to a running workflow execution.

        Args:
            workflow_id (str): The unique identifier of the workflow execution.
            signal_name (str): The name of the signal to send.
            arg (Any, optional): Argument to pass with the signal. Defaults to None.
            run_id (str, optional): The specific run identifier within the workflow.
                If None, signals the latest run. Defaults to None.
        """
        async with self._mapped_errors("signal_workflow", workflow_id):
            handle = await self.get_workflow_handle(workflow_id, run_id)
            # Only pass arg if it's not None, so signal handlers with no parameters work correctly.
            positional_args: list[Any] = [] if arg is None else [arg]
            await handle.signal(signal_name, *positional_args)

    @override
    async def query_workflow(
        self,
        workflow_id: str,
        query_name: str,
        arg: Any = None,
        run_id: str | None = None,
    ) -> Any:
        """Query a running workflow execution for information.

        Args:
            workflow_id (str): The unique identifier of the workflow execution.
            query_name (str): The name of the query to execute.
            arg (Any, optional): Argument to pass with the query. Defaults to None.
            run_id (str, optional): The specific run identifier within the workflow.
                If None, queries the latest run. Defaults to None.

        Returns:
            Any: The query result from the workflow.
        """
        async with self._mapped_errors("query_workflow", workflow_id):
            handle = await self.get_workflow_handle(workflow_id, run_id)
            # Only pass arg if it's not None, so query handlers with no parameters work correctly.
            positional_args: list[Any] = [] if arg is None else [arg]
            return await handle.query(query_name, *positional_args)

    @override
    async def list_workflows(
        self,
        query: str | None = None,
        page_size: int | None = None,
        next_page_token: bytes | None = None,
        limit: int | None = None,
    ) -> list[WorkflowExecution]:
        """List workflow executions matching the given criteria.

        Args:
            query (str, optional): List filter query in Temporal SQL syntax.
                Defaults to None (no filter).
            page_size (int, optional): Maximum number of results per page.
                Defaults to None (server default).
            next_page_token (bytes, optional): Token for pagination.
                Defaults to None (first page).
            limit (int, optional): Maximum number of executions to return. Defaults to None, which loads every
                matching execution into memory; set it, or use ``iter_workflows``, for large namespaces.

        Returns:
            list[WorkflowExecution]: The workflow executions matching the query.
        """
        workflows_iter = await self.iter_workflows(
            query=query,
            page_size=page_size,
            next_page_token=next_page_token,
            limit=limit,
        )
        return [workflow async for workflow in workflows_iter]

    async def iter_workflows(
        self,
        query: str | None = None,
        page_size: int | None = None,
        next_page_token: bytes | None = None,
        limit: int | None = None,
    ) -> WorkflowExecutionAsyncIterator:
        """Iterate workflow executions page by page without loading them all into memory.

        Args:
            query (str, optional): List filter query in Temporal SQL syntax. Defaults to None (no filter).
            page_size (int, optional): Number of results fetched per page. Defaults to None (100).
            next_page_token (bytes, optional): Token to resume from. Defaults to None (first page).
            limit (int, optional): Maximum number of executions to yield. Defaults to None (no limit).

        Returns:
            WorkflowExecutionAsyncIterator: Async iterator of executions; its ``next_page_token`` resumes the listing.
        """
        async with self._mapped_errors("iter_workflows", None):
            client = await self.get_client()
            return client.list_workflows(
                query=query,
                limit=limit,
                page_size=page_size or 100,
                next_page_token=next_page_token,
            )

    @override
    async def describe_workflow(self, workflow_id: str, run_id: str | None = None) -> WorkflowExecutionDescription:
        """Get detailed information about a workflow execution.

        Args:
            workflow_id (str): The unique identifier of the workflow execution.
            run_id (str, optional): The specific run identifier within the workflow.
                If None, describes the latest run. Defaults to None.

        Returns:
            WorkflowExecutionDescription: Detailed workflow execution information.
        """
        async with self._mapped_errors("describe_workflow", workflow_id):
            handle = await self.get_workflow_handle(workflow_id, run_id)
            return await handle.describe()

    @override
    async def close(self) -> None:
        """Close the Temporal client connection.

        Performs cleanup of resources and closes the connection to the Temporal server.
        Should be called when the adapter is no longer needed.
        """
        if self._client:
            # Temporal client doesn't have a close method, just clear the reference
            self._client = None

    @override
    async def create_schedule(
        self,
        schedule_id: str,
        workflow_class: Any,
        spec: ScheduleSpec,
        task_queue: str,
        workflow_id: str | None = None,
        schedule_policy: SchedulePolicy | None = None,
        args: Sequence[Any] | None = None,
        memo: dict[str, Any] | None = None,
    ) -> None:
        """Create a schedule for a workflow.

        Args:
            schedule_id (str): Unique schedule ID.
            workflow_class (Any): The workflow class or workflow type name to start.
            spec (ScheduleSpec): When the schedule fires (intervals, calendars, cron).
            task_queue (str): Task queue of the started workflows.
            workflow_id (str, optional): Base ID of the started workflows; the server appends the fire time.
                Defaults to the schedule ID.
            schedule_policy (SchedulePolicy, optional): Overlap, catchup window and pause-on-failure policy.
                Defaults to skipping overlapping runs. ``SchedulePolicy.catchup_window`` defaults to the server value,
                so set it explicitly when missed runs must be caught up over a fixed window.
            args (Sequence[Any], optional): Arguments of the started workflows. Defaults to None.
            memo (dict[str, Any], optional): Memo of the started workflows. Defaults to None.
        """
        async with self._mapped_errors("create_schedule", schedule_id):
            client = await self.get_client()

            workflow_execution_id = workflow_id or schedule_id
            sched = Schedule(
                action=ScheduleActionStartWorkflow(
                    workflow=workflow_class,
                    args=list(args or []),
                    id=workflow_execution_id,
                    task_queue=task_queue,
                    memo=memo,
                ),
                spec=spec,
                policy=schedule_policy
                or SchedulePolicy(
                    overlap=ScheduleOverlapPolicy.SKIP,
                ),
            )

            await client.create_schedule(schedule_id, sched)

    @override
    async def stop_schedule(self, schedule_id: str) -> None:
        """Delete a schedule.

        Deprecated: the name hides that the schedule is deleted; use ``delete_schedule``, or ``pause_schedule`` to
        stop it while keeping it.

        Args:
            schedule_id (str): The schedule ID.
        """
        warnings.warn(
            "stop_schedule() deletes the schedule and is deprecated; use delete_schedule() or pause_schedule()",
            DeprecationWarning,
            stacklevel=2,
        )
        await self.delete_schedule(schedule_id)

    @override
    async def delete_schedule(self, schedule_id: str) -> None:
        """Delete a schedule permanently; ``pause_schedule`` stops it while keeping it.

        Args:
            schedule_id (str): The schedule ID.
        """
        async with self._mapped_errors("delete_schedule", schedule_id):
            await (await self._schedule_handle(schedule_id)).delete()

    async def _schedule_handle(self, schedule_id: str) -> ScheduleHandle:
        """Get a handle to an existing schedule.

        Args:
            schedule_id (str): The schedule ID.

        Returns:
            ScheduleHandle: Handle to the schedule.
        """
        client = await self.get_client()
        return client.get_schedule_handle(schedule_id)

    async def pause_schedule(self, schedule_id: str, note: str | None = None) -> None:
        """Pause a schedule so it stops firing.

        Args:
            schedule_id (str): The schedule ID.
            note (str, optional): Why the schedule is paused. Defaults to None.
        """
        async with self._mapped_errors("pause_schedule", schedule_id):
            await (await self._schedule_handle(schedule_id)).pause(note=note)

    async def unpause_schedule(self, schedule_id: str, note: str | None = None) -> None:
        """Resume a paused schedule.

        Args:
            schedule_id (str): The schedule ID.
            note (str, optional): Why the schedule is resumed. Defaults to None.
        """
        async with self._mapped_errors("unpause_schedule", schedule_id):
            await (await self._schedule_handle(schedule_id)).unpause(note=note)

    async def trigger_schedule(self, schedule_id: str, overlap: ScheduleOverlapPolicy | None = None) -> None:
        """Start the schedule's action now, outside its spec.

        Args:
            schedule_id (str): The schedule ID.
            overlap (ScheduleOverlapPolicy, optional): Overlap policy for this run. Defaults to the schedule's.
        """
        async with self._mapped_errors("trigger_schedule", schedule_id):
            await (await self._schedule_handle(schedule_id)).trigger(overlap=overlap)

    async def describe_schedule(self, schedule_id: str) -> ScheduleDescription:
        """Get a schedule's definition, state and recent actions.

        Args:
            schedule_id (str): The schedule ID.

        Returns:
            ScheduleDescription: The schedule description.
        """
        async with self._mapped_errors("describe_schedule", schedule_id):
            return await (await self._schedule_handle(schedule_id)).describe()

    async def update_schedule(
        self,
        schedule_id: str,
        spec: ScheduleSpec | None = None,
        schedule_policy: SchedulePolicy | None = None,
    ) -> None:
        """Replace a schedule's spec and/or policy, keeping its action and state.

        Args:
            schedule_id (str): The schedule ID.
            spec (ScheduleSpec, optional): New spec. Defaults to None (unchanged).
            schedule_policy (SchedulePolicy, optional): New policy. Defaults to None (unchanged).
        """

        def _update(update_input: ScheduleUpdateInput) -> ScheduleUpdate:
            schedule = update_input.description.schedule
            if spec is not None:
                schedule.spec = spec
            if schedule_policy is not None:
                schedule.policy = schedule_policy
            return ScheduleUpdate(schedule=schedule)

        async with self._mapped_errors("update_schedule", schedule_id):
            await (await self._schedule_handle(schedule_id)).update(_update)

    async def list_schedules(self, query: str | None = None, limit: int | None = None) -> list[ScheduleListDescription]:
        """List schedules, optionally filtered by a visibility query.

        Args:
            query (str, optional): Visibility list filter. Defaults to None (all schedules).
            limit (int, optional): Maximum number of schedules to return. Defaults to None (all).

        Returns:
            list[ScheduleListDescription]: The matching schedules.
        """
        async with self._mapped_errors("list_schedules", None):
            client = await self.get_client()
            schedules: list[ScheduleListDescription] = []
            async for schedule in await client.list_schedules(query):
                if limit is not None and len(schedules) >= limit:
                    break
                schedules.append(schedule)
            return schedules

archipy.adapters.temporal.adapters.TemporalAdapter.config instance-attribute

config = global_config.TEMPORAL

archipy.adapters.temporal.adapters.TemporalAdapter.get_client async

get_client() -> Client

Get or create the Temporal client connection.

Returns:

Name	Type	Description
Client	Client	The Temporal client instance.

Raises:

Type	Description
ConnectionError	If unable to connect to Temporal server.

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

async def get_client(self) -> Client:
    """Get or create the Temporal client connection.

    Returns:
        Client: The Temporal client instance.

    Raises:
        ConnectionError: If unable to connect to Temporal server.
    """
    if self._client is not None:
        return self._client
    # Concurrent first calls must share one connection.
    async with self._client_lock:
        if self._client is None:
            self._client = await self._connect()
    return self._client

archipy.adapters.temporal.adapters.TemporalAdapter.start_workflow async

start_workflow(
    workflow: str | Callable[..., Any],
    arg: Any = None,
    workflow_id: str | None = None,
    task_queue: str | None = None,
    execution_timeout: int | None = None,
    run_timeout: int | None = None,
    task_timeout: int | None = None,
    memo: dict[str, Any] | None = None,
    search_attributes: dict[str, Any] | None = None,
    id_reuse_policy: WorkflowIDReusePolicy | None = None,
    id_conflict_policy: WorkflowIDConflictPolicy
    | None = None,
) -> WorkflowHandle[T, Any]

Start a workflow execution asynchronously.

Parameters:

Name	Type	Description	Default
workflow	str | Callable	The workflow function or workflow type name.	required
arg	Any	Input argument for the workflow. Defaults to None.	None
workflow_id	str	Unique identifier for the workflow execution. If None, a UUID will be generated. Defaults to None.	None
task_queue	str	Task queue name for workflow execution. If None, uses the default task queue. Defaults to None.	None
execution_timeout	int	Maximum workflow execution time in seconds. Overrides config default. Defaults to None.	None
run_timeout	int	Maximum single workflow run time in seconds. Overrides config default. Defaults to None.	None
task_timeout	int	Maximum workflow task processing time in seconds. Overrides config default. Defaults to None.	None
memo	dict[str, Any]	Non-indexed metadata for the workflow. Defaults to None.	None
search_attributes	dict[str, Any]	Indexed metadata for workflow search. Datetime values must be timezone-aware; temporalio 1.35+ raises on naive ones. Defaults to None.	None
id_reuse_policy	WorkflowIDReusePolicy	Whether a closed workflow's ID may be reused. Defaults to None (server default: allow duplicate).	None
id_conflict_policy	WorkflowIDConflictPolicy	What to do when a workflow with the same ID is still running, e.g. USE_EXISTING for idempotent starts. Defaults to None (server default: fail).	None

Returns:

Type	Description
WorkflowHandle[T, Any]	WorkflowHandle[T, Any]: Handle to the started workflow execution.

Note

The generated ID comes from the standard library uuid.uuid7(), which is fine here, outside workflow code. Inside a workflow the sandbox blocks it: use temporalio.workflow.uuid7() instead.

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def start_workflow(
    self,
    workflow: str | Callable[..., Any],
    arg: Any = None,
    workflow_id: str | None = None,
    task_queue: str | None = None,
    execution_timeout: int | None = None,
    run_timeout: int | None = None,
    task_timeout: int | None = None,
    memo: dict[str, Any] | None = None,
    search_attributes: dict[str, Any] | None = None,
    id_reuse_policy: WorkflowIDReusePolicy | None = None,
    id_conflict_policy: WorkflowIDConflictPolicy | None = None,
) -> WorkflowHandle[T, Any]:
    """Start a workflow execution asynchronously.

    Args:
        workflow (str | Callable): The workflow function or workflow type name.
        arg (Any, optional): Input argument for the workflow. Defaults to None.
        workflow_id (str, optional): Unique identifier for the workflow execution.
            If None, a UUID will be generated. Defaults to None.
        task_queue (str, optional): Task queue name for workflow execution.
            If None, uses the default task queue. Defaults to None.
        execution_timeout (int, optional): Maximum workflow execution time in seconds.
            Overrides config default. Defaults to None.
        run_timeout (int, optional): Maximum single workflow run time in seconds.
            Overrides config default. Defaults to None.
        task_timeout (int, optional): Maximum workflow task processing time in seconds.
            Overrides config default. Defaults to None.
        memo (dict[str, Any], optional): Non-indexed metadata for the workflow.
            Defaults to None.
        search_attributes (dict[str, Any], optional): Indexed metadata for workflow search. Datetime values must
            be timezone-aware; ``temporalio`` 1.35+ raises on naive ones. Defaults to None.
        id_reuse_policy (WorkflowIDReusePolicy, optional): Whether a closed workflow's ID may be reused.
            Defaults to None (server default: allow duplicate).
        id_conflict_policy (WorkflowIDConflictPolicy, optional): What to do when a workflow with the same ID
            is still running, e.g. ``USE_EXISTING`` for idempotent starts. Defaults to None (server default: fail).

    Returns:
        WorkflowHandle[T, Any]: Handle to the started workflow execution.

    Note:
        The generated ID comes from the standard library ``uuid.uuid7()``, which is fine here, outside workflow
        code. Inside a workflow the sandbox blocks it: use ``temporalio.workflow.uuid7()`` instead.
    """
    async with self._mapped_errors("start_workflow", workflow_id):
        client = await self.get_client()

        workflow_id = workflow_id or str(uuid7())
        task_queue = task_queue or self.config.TASK_QUEUE

        # Build positional args: only include arg if it's not None,
        # so workflows that take no parameters aren't given an extra argument.
        positional_args: list[Any] = [] if arg is None else [arg]

        return await client.start_workflow(
            workflow,
            *positional_args,
            id=workflow_id,
            task_queue=task_queue,
            execution_timeout=self._timeout(execution_timeout, self.config.WORKFLOW_EXECUTION_TIMEOUT),
            run_timeout=self._timeout(run_timeout, self.config.WORKFLOW_RUN_TIMEOUT),
            task_timeout=self._timeout(task_timeout, self.config.WORKFLOW_TASK_TIMEOUT),
            retry_policy=self._build_workflow_retry_policy(),
            memo=memo,
            search_attributes=search_attributes,
            **self._id_policies(id_reuse_policy, id_conflict_policy),
        )

archipy.adapters.temporal.adapters.TemporalAdapter.execute_workflow async

execute_workflow(
    workflow: str | Callable[..., Any],
    arg: Any = None,
    workflow_id: str | None = None,
    task_queue: str | None = None,
    execution_timeout: int | None = None,
    run_timeout: int | None = None,
    task_timeout: int | None = None,
    id_reuse_policy: WorkflowIDReusePolicy | None = None,
    id_conflict_policy: WorkflowIDConflictPolicy
    | None = None,
) -> T

Execute a workflow and wait for its completion.

Parameters:

Name	Type	Description	Default
workflow	str | Callable	The workflow function or workflow type name.	required
arg	Any	Input argument for the workflow. Defaults to None.	None
workflow_id	str	Unique identifier for the workflow execution. If None, a UUID will be generated. Defaults to None.	None
task_queue	str	Task queue name for workflow execution. If None, uses the default task queue. Defaults to None.	None
execution_timeout	int	Maximum workflow execution time in seconds. Overrides config default. Defaults to None.	None
run_timeout	int	Maximum single workflow run time in seconds. Overrides config default. Defaults to None.	None
task_timeout	int	Maximum workflow task processing time in seconds. Overrides config default. Defaults to None.	None
id_reuse_policy	WorkflowIDReusePolicy	Whether a closed workflow's ID may be reused. Defaults to None (server default: allow duplicate).	None
id_conflict_policy	WorkflowIDConflictPolicy	What to do when a workflow with the same ID is still running, e.g. USE_EXISTING for idempotent starts. Defaults to None (server default: fail).	None

Returns:

Name	Type	Description
T	T	The workflow execution result.

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def execute_workflow(
    self,
    workflow: str | Callable[..., Any],
    arg: Any = None,
    workflow_id: str | None = None,
    task_queue: str | None = None,
    execution_timeout: int | None = None,
    run_timeout: int | None = None,
    task_timeout: int | None = None,
    id_reuse_policy: WorkflowIDReusePolicy | None = None,
    id_conflict_policy: WorkflowIDConflictPolicy | None = None,
) -> T:
    """Execute a workflow and wait for its completion.

    Args:
        workflow (str | Callable): The workflow function or workflow type name.
        arg (Any, optional): Input argument for the workflow. Defaults to None.
        workflow_id (str, optional): Unique identifier for the workflow execution.
            If None, a UUID will be generated. Defaults to None.
        task_queue (str, optional): Task queue name for workflow execution.
            If None, uses the default task queue. Defaults to None.
        execution_timeout (int, optional): Maximum workflow execution time in seconds.
            Overrides config default. Defaults to None.
        run_timeout (int, optional): Maximum single workflow run time in seconds.
            Overrides config default. Defaults to None.
        task_timeout (int, optional): Maximum workflow task processing time in seconds.
            Overrides config default. Defaults to None.
        id_reuse_policy (WorkflowIDReusePolicy, optional): Whether a closed workflow's ID may be reused.
            Defaults to None (server default: allow duplicate).
        id_conflict_policy (WorkflowIDConflictPolicy, optional): What to do when a workflow with the same ID
            is still running, e.g. ``USE_EXISTING`` for idempotent starts. Defaults to None (server default: fail).

    Returns:
        T: The workflow execution result.
    """
    async with self._mapped_errors("execute_workflow", workflow_id):
        client = await self.get_client()

        workflow_id = workflow_id or str(uuid7())
        task_queue = task_queue or self.config.TASK_QUEUE

        # Build positional args: only include arg if it's not None,
        # so workflows that take no parameters aren't given an extra argument.
        positional_args: list[Any] = [] if arg is None else [arg]

        return await client.execute_workflow(
            workflow,
            *positional_args,
            id=workflow_id,
            task_queue=task_queue,
            execution_timeout=self._timeout(execution_timeout, self.config.WORKFLOW_EXECUTION_TIMEOUT),
            run_timeout=self._timeout(run_timeout, self.config.WORKFLOW_RUN_TIMEOUT),
            task_timeout=self._timeout(task_timeout, self.config.WORKFLOW_TASK_TIMEOUT),
            retry_policy=self._build_workflow_retry_policy(),
            **self._id_policies(id_reuse_policy, id_conflict_policy),
        )

archipy.adapters.temporal.adapters.TemporalAdapter.get_workflow_handle async

get_workflow_handle(
    workflow_id: str, run_id: str | None = None
) -> WorkflowHandle[T, Any]

Get a handle to an existing workflow execution.

Parameters:

Name	Type	Description	Default
workflow_id	str	The unique identifier of the workflow execution.	required
run_id	str	The specific run identifier within the workflow. If None, gets the latest run. Defaults to None.	None

Returns:

Type	Description
WorkflowHandle[T, Any]	WorkflowHandle[T, Any]: Handle to the workflow execution.

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def get_workflow_handle(self, workflow_id: str, run_id: str | None = None) -> WorkflowHandle[T, Any]:
    """Get a handle to an existing workflow execution.

    Args:
        workflow_id (str): The unique identifier of the workflow execution.
        run_id (str, optional): The specific run identifier within the workflow.
            If None, gets the latest run. Defaults to None.

    Returns:
        WorkflowHandle[T, Any]: Handle to the workflow execution.
    """
    async with self._mapped_errors("get_workflow_handle", workflow_id):
        client = await self.get_client()
        return client.get_workflow_handle(workflow_id, run_id=run_id)

archipy.adapters.temporal.adapters.TemporalAdapter.cancel_workflow async

cancel_workflow(
    workflow_id: str,
    run_id: str | None = None,
    reason: str | None = None,
) -> None

Cancel a running workflow execution.

Parameters:

Name	Type	Description	Default
workflow_id	str	The unique identifier of the workflow execution.	required
run_id	str	The specific run identifier within the workflow. If None, cancels the latest run. Defaults to None.	None
reason	str	Reason for cancellation. Defaults to None.	None

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def cancel_workflow(self, workflow_id: str, run_id: str | None = None, reason: str | None = None) -> None:
    """Cancel a running workflow execution.

    Args:
        workflow_id (str): The unique identifier of the workflow execution.
        run_id (str, optional): The specific run identifier within the workflow.
            If None, cancels the latest run. Defaults to None.
        reason (str, optional): Reason for cancellation. Defaults to None.
    """
    async with self._mapped_errors("cancel_workflow", workflow_id):
        handle = await self.get_workflow_handle(workflow_id, run_id)
        await handle.cancel(reason=reason or "")

archipy.adapters.temporal.adapters.TemporalAdapter.terminate_workflow async

terminate_workflow(
    workflow_id: str,
    run_id: str | None = None,
    reason: str | None = None,
) -> None

Terminate a running workflow execution immediately.

Parameters:

Name	Type	Description	Default
workflow_id	str	The unique identifier of the workflow execution.	required
run_id	str	The specific run identifier within the workflow. If None, terminates the latest run. Defaults to None.	None
reason	str	Reason for termination. Defaults to None.	None

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def terminate_workflow(self, workflow_id: str, run_id: str | None = None, reason: str | None = None) -> None:
    """Terminate a running workflow execution immediately.

    Args:
        workflow_id (str): The unique identifier of the workflow execution.
        run_id (str, optional): The specific run identifier within the workflow.
            If None, terminates the latest run. Defaults to None.
        reason (str, optional): Reason for termination. Defaults to None.
    """
    async with self._mapped_errors("terminate_workflow", workflow_id):
        handle = await self.get_workflow_handle(workflow_id, run_id)
        await handle.terminate(reason=reason)

archipy.adapters.temporal.adapters.TemporalAdapter.signal_workflow async

signal_workflow(
    workflow_id: str,
    signal_name: str,
    arg: Any = None,
    run_id: str | None = None,
) -> None

Send a signal to a running workflow execution.

Parameters:

Name	Type	Description	Default
workflow_id	str	The unique identifier of the workflow execution.	required
signal_name	str	The name of the signal to send.	required
arg	Any	Argument to pass with the signal. Defaults to None.	None
run_id	str	The specific run identifier within the workflow. If None, signals the latest run. Defaults to None.	None

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def signal_workflow(
    self,
    workflow_id: str,
    signal_name: str,
    arg: Any = None,
    run_id: str | None = None,
) -> None:
    """Send a signal to a running workflow execution.

    Args:
        workflow_id (str): The unique identifier of the workflow execution.
        signal_name (str): The name of the signal to send.
        arg (Any, optional): Argument to pass with the signal. Defaults to None.
        run_id (str, optional): The specific run identifier within the workflow.
            If None, signals the latest run. Defaults to None.
    """
    async with self._mapped_errors("signal_workflow", workflow_id):
        handle = await self.get_workflow_handle(workflow_id, run_id)
        # Only pass arg if it's not None, so signal handlers with no parameters work correctly.
        positional_args: list[Any] = [] if arg is None else [arg]
        await handle.signal(signal_name, *positional_args)

archipy.adapters.temporal.adapters.TemporalAdapter.query_workflow async

query_workflow(
    workflow_id: str,
    query_name: str,
    arg: Any = None,
    run_id: str | None = None,
) -> Any

Query a running workflow execution for information.

Parameters:

Name	Type	Description	Default
workflow_id	str	The unique identifier of the workflow execution.	required
query_name	str	The name of the query to execute.	required
arg	Any	Argument to pass with the query. Defaults to None.	None
run_id	str	The specific run identifier within the workflow. If None, queries the latest run. Defaults to None.	None

Returns:

Name	Type	Description
Any	Any	The query result from the workflow.

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def query_workflow(
    self,
    workflow_id: str,
    query_name: str,
    arg: Any = None,
    run_id: str | None = None,
) -> Any:
    """Query a running workflow execution for information.

    Args:
        workflow_id (str): The unique identifier of the workflow execution.
        query_name (str): The name of the query to execute.
        arg (Any, optional): Argument to pass with the query. Defaults to None.
        run_id (str, optional): The specific run identifier within the workflow.
            If None, queries the latest run. Defaults to None.

    Returns:
        Any: The query result from the workflow.
    """
    async with self._mapped_errors("query_workflow", workflow_id):
        handle = await self.get_workflow_handle(workflow_id, run_id)
        # Only pass arg if it's not None, so query handlers with no parameters work correctly.
        positional_args: list[Any] = [] if arg is None else [arg]
        return await handle.query(query_name, *positional_args)

archipy.adapters.temporal.adapters.TemporalAdapter.list_workflows async

list_workflows(
    query: str | None = None,
    page_size: int | None = None,
    next_page_token: bytes | None = None,
    limit: int | None = None,
) -> list[WorkflowExecution]

List workflow executions matching the given criteria.

Parameters:

Name	Type	Description	Default
query	str	List filter query in Temporal SQL syntax. Defaults to None (no filter).	None
page_size	int	Maximum number of results per page. Defaults to None (server default).	None
next_page_token	bytes	Token for pagination. Defaults to None (first page).	None
limit	int	Maximum number of executions to return. Defaults to None, which loads every matching execution into memory; set it, or use iter_workflows, for large namespaces.	None

Returns:

Type	Description
list[WorkflowExecution]	list[WorkflowExecution]: The workflow executions matching the query.

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def list_workflows(
    self,
    query: str | None = None,
    page_size: int | None = None,
    next_page_token: bytes | None = None,
    limit: int | None = None,
) -> list[WorkflowExecution]:
    """List workflow executions matching the given criteria.

    Args:
        query (str, optional): List filter query in Temporal SQL syntax.
            Defaults to None (no filter).
        page_size (int, optional): Maximum number of results per page.
            Defaults to None (server default).
        next_page_token (bytes, optional): Token for pagination.
            Defaults to None (first page).
        limit (int, optional): Maximum number of executions to return. Defaults to None, which loads every
            matching execution into memory; set it, or use ``iter_workflows``, for large namespaces.

    Returns:
        list[WorkflowExecution]: The workflow executions matching the query.
    """
    workflows_iter = await self.iter_workflows(
        query=query,
        page_size=page_size,
        next_page_token=next_page_token,
        limit=limit,
    )
    return [workflow async for workflow in workflows_iter]

archipy.adapters.temporal.adapters.TemporalAdapter.iter_workflows async

iter_workflows(
    query: str | None = None,
    page_size: int | None = None,
    next_page_token: bytes | None = None,
    limit: int | None = None,
) -> WorkflowExecutionAsyncIterator

Iterate workflow executions page by page without loading them all into memory.

Parameters:

Name	Type	Description	Default
query	str	List filter query in Temporal SQL syntax. Defaults to None (no filter).	None
page_size	int	Number of results fetched per page. Defaults to None (100).	None
next_page_token	bytes	Token to resume from. Defaults to None (first page).	None
limit	int	Maximum number of executions to yield. Defaults to None (no limit).	None

Returns:

Name	Type	Description
WorkflowExecutionAsyncIterator	WorkflowExecutionAsyncIterator	Async iterator of executions; its next_page_token resumes the listing.

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

async def iter_workflows(
    self,
    query: str | None = None,
    page_size: int | None = None,
    next_page_token: bytes | None = None,
    limit: int | None = None,
) -> WorkflowExecutionAsyncIterator:
    """Iterate workflow executions page by page without loading them all into memory.

    Args:
        query (str, optional): List filter query in Temporal SQL syntax. Defaults to None (no filter).
        page_size (int, optional): Number of results fetched per page. Defaults to None (100).
        next_page_token (bytes, optional): Token to resume from. Defaults to None (first page).
        limit (int, optional): Maximum number of executions to yield. Defaults to None (no limit).

    Returns:
        WorkflowExecutionAsyncIterator: Async iterator of executions; its ``next_page_token`` resumes the listing.
    """
    async with self._mapped_errors("iter_workflows", None):
        client = await self.get_client()
        return client.list_workflows(
            query=query,
            limit=limit,
            page_size=page_size or 100,
            next_page_token=next_page_token,
        )

archipy.adapters.temporal.adapters.TemporalAdapter.describe_workflow async

describe_workflow(
    workflow_id: str, run_id: str | None = None
) -> WorkflowExecutionDescription

Get detailed information about a workflow execution.

Parameters:

Name	Type	Description	Default
workflow_id	str	The unique identifier of the workflow execution.	required
run_id	str	The specific run identifier within the workflow. If None, describes the latest run. Defaults to None.	None

Returns:

Name	Type	Description
WorkflowExecutionDescription	WorkflowExecutionDescription	Detailed workflow execution information.

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def describe_workflow(self, workflow_id: str, run_id: str | None = None) -> WorkflowExecutionDescription:
    """Get detailed information about a workflow execution.

    Args:
        workflow_id (str): The unique identifier of the workflow execution.
        run_id (str, optional): The specific run identifier within the workflow.
            If None, describes the latest run. Defaults to None.

    Returns:
        WorkflowExecutionDescription: Detailed workflow execution information.
    """
    async with self._mapped_errors("describe_workflow", workflow_id):
        handle = await self.get_workflow_handle(workflow_id, run_id)
        return await handle.describe()

archipy.adapters.temporal.adapters.TemporalAdapter.close async

close() -> None

Close the Temporal client connection.

Performs cleanup of resources and closes the connection to the Temporal server. Should be called when the adapter is no longer needed.

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def close(self) -> None:
    """Close the Temporal client connection.

    Performs cleanup of resources and closes the connection to the Temporal server.
    Should be called when the adapter is no longer needed.
    """
    if self._client:
        # Temporal client doesn't have a close method, just clear the reference
        self._client = None

archipy.adapters.temporal.adapters.TemporalAdapter.create_schedule async

create_schedule(
    schedule_id: str,
    workflow_class: Any,
    spec: ScheduleSpec,
    task_queue: str,
    workflow_id: str | None = None,
    schedule_policy: SchedulePolicy | None = None,
    args: Sequence[Any] | None = None,
    memo: dict[str, Any] | None = None,
) -> None

Create a schedule for a workflow.

Parameters:

Name	Type	Description	Default
schedule_id	str	Unique schedule ID.	required
workflow_class	Any	The workflow class or workflow type name to start.	required
spec	ScheduleSpec	When the schedule fires (intervals, calendars, cron).	required
task_queue	str	Task queue of the started workflows.	required
workflow_id	str	Base ID of the started workflows; the server appends the fire time. Defaults to the schedule ID.	None
schedule_policy	SchedulePolicy	Overlap, catchup window and pause-on-failure policy. Defaults to skipping overlapping runs. SchedulePolicy.catchup_window defaults to the server value, so set it explicitly when missed runs must be caught up over a fixed window.	None
args	Sequence[Any]	Arguments of the started workflows. Defaults to None.	None
memo	dict[str, Any]	Memo of the started workflows. Defaults to None.	None

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def create_schedule(
    self,
    schedule_id: str,
    workflow_class: Any,
    spec: ScheduleSpec,
    task_queue: str,
    workflow_id: str | None = None,
    schedule_policy: SchedulePolicy | None = None,
    args: Sequence[Any] | None = None,
    memo: dict[str, Any] | None = None,
) -> None:
    """Create a schedule for a workflow.

    Args:
        schedule_id (str): Unique schedule ID.
        workflow_class (Any): The workflow class or workflow type name to start.
        spec (ScheduleSpec): When the schedule fires (intervals, calendars, cron).
        task_queue (str): Task queue of the started workflows.
        workflow_id (str, optional): Base ID of the started workflows; the server appends the fire time.
            Defaults to the schedule ID.
        schedule_policy (SchedulePolicy, optional): Overlap, catchup window and pause-on-failure policy.
            Defaults to skipping overlapping runs. ``SchedulePolicy.catchup_window`` defaults to the server value,
            so set it explicitly when missed runs must be caught up over a fixed window.
        args (Sequence[Any], optional): Arguments of the started workflows. Defaults to None.
        memo (dict[str, Any], optional): Memo of the started workflows. Defaults to None.
    """
    async with self._mapped_errors("create_schedule", schedule_id):
        client = await self.get_client()

        workflow_execution_id = workflow_id or schedule_id
        sched = Schedule(
            action=ScheduleActionStartWorkflow(
                workflow=workflow_class,
                args=list(args or []),
                id=workflow_execution_id,
                task_queue=task_queue,
                memo=memo,
            ),
            spec=spec,
            policy=schedule_policy
            or SchedulePolicy(
                overlap=ScheduleOverlapPolicy.SKIP,
            ),
        )

        await client.create_schedule(schedule_id, sched)

archipy.adapters.temporal.adapters.TemporalAdapter.stop_schedule async

stop_schedule(schedule_id: str) -> None

Delete a schedule.

Deprecated: the name hides that the schedule is deleted; use delete_schedule, or pause_schedule to stop it while keeping it.

Parameters:

Name	Type	Description	Default
schedule_id	str	The schedule ID.	required

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def stop_schedule(self, schedule_id: str) -> None:
    """Delete a schedule.

    Deprecated: the name hides that the schedule is deleted; use ``delete_schedule``, or ``pause_schedule`` to
    stop it while keeping it.

    Args:
        schedule_id (str): The schedule ID.
    """
    warnings.warn(
        "stop_schedule() deletes the schedule and is deprecated; use delete_schedule() or pause_schedule()",
        DeprecationWarning,
        stacklevel=2,
    )
    await self.delete_schedule(schedule_id)

archipy.adapters.temporal.adapters.TemporalAdapter.delete_schedule async

delete_schedule(schedule_id: str) -> None

Delete a schedule permanently; pause_schedule stops it while keeping it.

Parameters:

Name	Type	Description	Default
schedule_id	str	The schedule ID.	required

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

@override
async def delete_schedule(self, schedule_id: str) -> None:
    """Delete a schedule permanently; ``pause_schedule`` stops it while keeping it.

    Args:
        schedule_id (str): The schedule ID.
    """
    async with self._mapped_errors("delete_schedule", schedule_id):
        await (await self._schedule_handle(schedule_id)).delete()

archipy.adapters.temporal.adapters.TemporalAdapter.pause_schedule async

pause_schedule(
    schedule_id: str, note: str | None = None
) -> None

Pause a schedule so it stops firing.

Parameters:

Name	Type	Description	Default
schedule_id	str	The schedule ID.	required
note	str	Why the schedule is paused. Defaults to None.	None

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

async def pause_schedule(self, schedule_id: str, note: str | None = None) -> None:
    """Pause a schedule so it stops firing.

    Args:
        schedule_id (str): The schedule ID.
        note (str, optional): Why the schedule is paused. Defaults to None.
    """
    async with self._mapped_errors("pause_schedule", schedule_id):
        await (await self._schedule_handle(schedule_id)).pause(note=note)

archipy.adapters.temporal.adapters.TemporalAdapter.unpause_schedule async

unpause_schedule(
    schedule_id: str, note: str | None = None
) -> None

Resume a paused schedule.

Parameters:

Name	Type	Description	Default
schedule_id	str	The schedule ID.	required
note	str	Why the schedule is resumed. Defaults to None.	None

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

async def unpause_schedule(self, schedule_id: str, note: str | None = None) -> None:
    """Resume a paused schedule.

    Args:
        schedule_id (str): The schedule ID.
        note (str, optional): Why the schedule is resumed. Defaults to None.
    """
    async with self._mapped_errors("unpause_schedule", schedule_id):
        await (await self._schedule_handle(schedule_id)).unpause(note=note)

archipy.adapters.temporal.adapters.TemporalAdapter.trigger_schedule async

trigger_schedule(
    schedule_id: str,
    overlap: ScheduleOverlapPolicy | None = None,
) -> None

Start the schedule's action now, outside its spec.

Parameters:

Name	Type	Description	Default
schedule_id	str	The schedule ID.	required
overlap	ScheduleOverlapPolicy	Overlap policy for this run. Defaults to the schedule's.	None

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

async def trigger_schedule(self, schedule_id: str, overlap: ScheduleOverlapPolicy | None = None) -> None:
    """Start the schedule's action now, outside its spec.

    Args:
        schedule_id (str): The schedule ID.
        overlap (ScheduleOverlapPolicy, optional): Overlap policy for this run. Defaults to the schedule's.
    """
    async with self._mapped_errors("trigger_schedule", schedule_id):
        await (await self._schedule_handle(schedule_id)).trigger(overlap=overlap)

archipy.adapters.temporal.adapters.TemporalAdapter.describe_schedule async

describe_schedule(schedule_id: str) -> ScheduleDescription

Get a schedule's definition, state and recent actions.

Parameters:

Name	Type	Description	Default
schedule_id	str	The schedule ID.	required

Returns:

Name	Type	Description
ScheduleDescription	ScheduleDescription	The schedule description.

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

async def describe_schedule(self, schedule_id: str) -> ScheduleDescription:
    """Get a schedule's definition, state and recent actions.

    Args:
        schedule_id (str): The schedule ID.

    Returns:
        ScheduleDescription: The schedule description.
    """
    async with self._mapped_errors("describe_schedule", schedule_id):
        return await (await self._schedule_handle(schedule_id)).describe()

archipy.adapters.temporal.adapters.TemporalAdapter.update_schedule async

update_schedule(
    schedule_id: str,
    spec: ScheduleSpec | None = None,
    schedule_policy: SchedulePolicy | None = None,
) -> None

Replace a schedule's spec and/or policy, keeping its action and state.

Parameters:

Name	Type	Description	Default
schedule_id	str	The schedule ID.	required
spec	ScheduleSpec	New spec. Defaults to None (unchanged).	None
schedule_policy	SchedulePolicy	New policy. Defaults to None (unchanged).	None

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

async def update_schedule(
    self,
    schedule_id: str,
    spec: ScheduleSpec | None = None,
    schedule_policy: SchedulePolicy | None = None,
) -> None:
    """Replace a schedule's spec and/or policy, keeping its action and state.

    Args:
        schedule_id (str): The schedule ID.
        spec (ScheduleSpec, optional): New spec. Defaults to None (unchanged).
        schedule_policy (SchedulePolicy, optional): New policy. Defaults to None (unchanged).
    """

    def _update(update_input: ScheduleUpdateInput) -> ScheduleUpdate:
        schedule = update_input.description.schedule
        if spec is not None:
            schedule.spec = spec
        if schedule_policy is not None:
            schedule.policy = schedule_policy
        return ScheduleUpdate(schedule=schedule)

    async with self._mapped_errors("update_schedule", schedule_id):
        await (await self._schedule_handle(schedule_id)).update(_update)

archipy.adapters.temporal.adapters.TemporalAdapter.list_schedules async

list_schedules(
    query: str | None = None, limit: int | None = None
) -> list[ScheduleListDescription]

List schedules, optionally filtered by a visibility query.

Parameters:

Name	Type	Description	Default
query	str	Visibility list filter. Defaults to None (all schedules).	None
limit	int	Maximum number of schedules to return. Defaults to None (all).	None

Returns:

Type	Description
list[ScheduleListDescription]	list[ScheduleListDescription]: The matching schedules.

Source code in packages/temporal/archipy/adapters/temporal/adapters.py

async def list_schedules(self, query: str | None = None, limit: int | None = None) -> list[ScheduleListDescription]:
    """List schedules, optionally filtered by a visibility query.

    Args:
        query (str, optional): Visibility list filter. Defaults to None (all schedules).
        limit (int, optional): Maximum number of schedules to return. Defaults to None (all).

    Returns:
        list[ScheduleListDescription]: The matching schedules.
    """
    async with self._mapped_errors("list_schedules", None):
        client = await self.get_client()
        schedules: list[ScheduleListDescription] = []
        async for schedule in await client.list_schedules(query):
            if limit is not None and len(schedules) >= limit:
                break
            schedules.append(schedule)
        return schedules

options: show_root_toc_entry: false heading_level: 3

Runtime

Temporal runtime configuration and initialization utilities.

Temporal Runtime singleton for managing Runtime instances with telemetry.

This module provides a singleton class for creating and managing Temporal Runtime instances with OpenTelemetry metrics integration.

archipy.adapters.temporal.runtime.logger module-attribute

logger = logging.getLogger(__name__)

archipy.adapters.temporal.runtime.TemporalRuntimeManager

Singleton manager for Temporal Runtime instances with telemetry configuration.

This class ensures only one Runtime instance is created and reused across all Temporal clients and workers. Once created with metrics enabled, the Runtime cannot be changed (Temporal SDK limitation).

Example

from archipy.adapters.temporal.runtime import TemporalRuntimeManager

# Get the singleton manager
manager = TemporalRuntimeManager()

# Get Runtime with OTLP metrics enabled
runtime = manager.get_runtime(
    otel_metrics_enabled=True,
    otlp_endpoint="http://localhost:4317",
)

Source code in packages/temporal/archipy/adapters/temporal/runtime.py

class TemporalRuntimeManager(metaclass=Singleton, thread_safe=True):
    """Singleton manager for Temporal Runtime instances with telemetry configuration.

    This class ensures only one Runtime instance is created and reused across all
    Temporal clients and workers. Once created with metrics enabled, the Runtime
    cannot be changed (Temporal SDK limitation).

    Example:
        ```python
        from archipy.adapters.temporal.runtime import TemporalRuntimeManager

        # Get the singleton manager
        manager = TemporalRuntimeManager()

        # Get Runtime with OTLP metrics enabled
        runtime = manager.get_runtime(
            otel_metrics_enabled=True,
            otlp_endpoint="http://localhost:4317",
        )
        ```
    """

    def __init__(self) -> None:
        """Initialize the TemporalRuntimeManager singleton."""
        self._runtime: Runtime | None = None

    def get_runtime(
        self,
        otel_metrics_enabled: bool = False,
        otlp_endpoint: str = "http://localhost:4317",
        headers: dict[str, str] | None = None,
        use_http: bool = False,
        prometheus_bind_address: str | None = None,
        global_tags: dict[str, str] | None = None,
        core_log_level: str | None = None,
        disable_environment_info: bool = False,
    ) -> Runtime | None:
        """Get or create a Runtime with OpenTelemetry telemetry.

        Args:
            otel_metrics_enabled: Whether to enable OTLP metrics export.
            otlp_endpoint: OTLP collector endpoint URL.
            headers: Optional headers for OTLP export (auth tokens, routing).
            use_http: If True, use HTTP/protobuf transport instead of gRPC.
            prometheus_bind_address: If set (``host:port``), the Runtime serves a Prometheus ``/metrics``
                scrape endpoint on that address instead of exporting over OTLP.
            global_tags: Tags added to every Temporal SDK metric (e.g. service, environment).
            core_log_level: Level of the SDK core's own logs (``ERROR``, ``WARN``, ``INFO``, ``DEBUG``, ``TRACE``);
                None keeps the SDK default.
            disable_environment_info: Keep runtime, hosting and platform information out of worker heartbeats.
                A Runtime is built for it even when metrics are disabled.

        Returns:
            Runtime | None: The configured Runtime instance if metrics are enabled or environment info is disabled,
                None otherwise (uses default Runtime).

        Raises:
            ConfigurationError: If the Runtime cannot be built from the telemetry settings (e.g. an invalid URL).

        Note:
            Once a Runtime is created with metrics enabled, it cannot be disabled
            or recreated with different settings due to Temporal SDK limitations.
            Subsequent calls will return the existing Runtime regardless of parameters.
        """
        if not otel_metrics_enabled and not disable_environment_info:
            logger.debug("OTLP metrics disabled for Temporal, using default runtime")
            return None

        # If Runtime already created, return it (can't change once bound)
        if self._runtime is not None:
            logger.debug("Returning existing Temporal Runtime instance")
            return self._runtime

        metrics: OpenTelemetryConfig | PrometheusConfig | None
        if not otel_metrics_enabled:
            logger.info("Creating Temporal Runtime without metrics to disable environment info")
            metrics = None
        elif prometheus_bind_address:
            logger.info("Creating Temporal Runtime with a Prometheus scrape endpoint on %s", prometheus_bind_address)
            metrics = PrometheusConfig(bind_address=prometheus_bind_address)
        else:
            logger.info("Creating Temporal Runtime with OTLP metrics at %s", otlp_endpoint)
            metrics = OpenTelemetryConfig(url=otlp_endpoint, headers=headers or None, http=use_http)

        telemetry_kwargs: dict[str, Any] = {"metrics": metrics, "global_tags": global_tags or {}}
        if core_log_level is not None:
            telemetry_kwargs["logging"] = LoggingConfig(
                filter=TelemetryFilter(core_level=core_log_level, other_level=core_log_level),
            )
        try:
            self._runtime = Runtime(
                telemetry=TelemetryConfig(**telemetry_kwargs),
                disable_environment_info=disable_environment_info,
            )
            logger.info("Temporal Runtime created successfully")
        except Exception as error:
            # Falling back to the default Runtime would drop every SDK metric without a trace, so fail loudly.
            raise ConfigurationError(
                operation="temporal_runtime",
                reason=f"cannot build the Temporal Runtime telemetry: {error}",
            ) from error

        return self._runtime

    def reset_runtime(self) -> None:
        """Reset the Runtime instance.

        Warning:
            This does NOT actually close the Runtime or release telemetry resources.
            The Temporal SDK does not support Runtime cleanup. This method only
            resets internal references for testing purposes.
        """
        logger.warning("Resetting Temporal Runtime reference (resources remain until process exit)")
        self._runtime = None

archipy.adapters.temporal.runtime.TemporalRuntimeManager.get_runtime

get_runtime(
    otel_metrics_enabled: bool = False,
    otlp_endpoint: str = "http://localhost:4317",
    headers: dict[str, str] | None = None,
    use_http: bool = False,
    prometheus_bind_address: str | None = None,
    global_tags: dict[str, str] | None = None,
    core_log_level: str | None = None,
    disable_environment_info: bool = False,
) -> Runtime | None

Get or create a Runtime with OpenTelemetry telemetry.

Parameters:

Name	Type	Description	Default
otel_metrics_enabled	bool	Whether to enable OTLP metrics export.	False
otlp_endpoint	str	OTLP collector endpoint URL.	'http://localhost:4317'
headers	dict[str, str] | None	Optional headers for OTLP export (auth tokens, routing).	None
use_http	bool	If True, use HTTP/protobuf transport instead of gRPC.	False
prometheus_bind_address	str | None	If set (host:port), the Runtime serves a Prometheus /metrics scrape endpoint on that address instead of exporting over OTLP.	None
global_tags	dict[str, str] | None	Tags added to every Temporal SDK metric (e.g. service, environment).	None
core_log_level	str | None	Level of the SDK core's own logs (ERROR, WARN, INFO, DEBUG, TRACE); None keeps the SDK default.	None
disable_environment_info	bool	Keep runtime, hosting and platform information out of worker heartbeats. A Runtime is built for it even when metrics are disabled.	False

Returns:

Type	Description
Runtime | None	Runtime | None: The configured Runtime instance if metrics are enabled or environment info is disabled, None otherwise (uses default Runtime).

Raises:

Type	Description
ConfigurationError	If the Runtime cannot be built from the telemetry settings (e.g. an invalid URL).

Note

Once a Runtime is created with metrics enabled, it cannot be disabled or recreated with different settings due to Temporal SDK limitations. Subsequent calls will return the existing Runtime regardless of parameters.

Source code in packages/temporal/archipy/adapters/temporal/runtime.py

def get_runtime(
    self,
    otel_metrics_enabled: bool = False,
    otlp_endpoint: str = "http://localhost:4317",
    headers: dict[str, str] | None = None,
    use_http: bool = False,
    prometheus_bind_address: str | None = None,
    global_tags: dict[str, str] | None = None,
    core_log_level: str | None = None,
    disable_environment_info: bool = False,
) -> Runtime | None:
    """Get or create a Runtime with OpenTelemetry telemetry.

    Args:
        otel_metrics_enabled: Whether to enable OTLP metrics export.
        otlp_endpoint: OTLP collector endpoint URL.
        headers: Optional headers for OTLP export (auth tokens, routing).
        use_http: If True, use HTTP/protobuf transport instead of gRPC.
        prometheus_bind_address: If set (``host:port``), the Runtime serves a Prometheus ``/metrics``
            scrape endpoint on that address instead of exporting over OTLP.
        global_tags: Tags added to every Temporal SDK metric (e.g. service, environment).
        core_log_level: Level of the SDK core's own logs (``ERROR``, ``WARN``, ``INFO``, ``DEBUG``, ``TRACE``);
            None keeps the SDK default.
        disable_environment_info: Keep runtime, hosting and platform information out of worker heartbeats.
            A Runtime is built for it even when metrics are disabled.

    Returns:
        Runtime | None: The configured Runtime instance if metrics are enabled or environment info is disabled,
            None otherwise (uses default Runtime).

    Raises:
        ConfigurationError: If the Runtime cannot be built from the telemetry settings (e.g. an invalid URL).

    Note:
        Once a Runtime is created with metrics enabled, it cannot be disabled
        or recreated with different settings due to Temporal SDK limitations.
        Subsequent calls will return the existing Runtime regardless of parameters.
    """
    if not otel_metrics_enabled and not disable_environment_info:
        logger.debug("OTLP metrics disabled for Temporal, using default runtime")
        return None

    # If Runtime already created, return it (can't change once bound)
    if self._runtime is not None:
        logger.debug("Returning existing Temporal Runtime instance")
        return self._runtime

    metrics: OpenTelemetryConfig | PrometheusConfig | None
    if not otel_metrics_enabled:
        logger.info("Creating Temporal Runtime without metrics to disable environment info")
        metrics = None
    elif prometheus_bind_address:
        logger.info("Creating Temporal Runtime with a Prometheus scrape endpoint on %s", prometheus_bind_address)
        metrics = PrometheusConfig(bind_address=prometheus_bind_address)
    else:
        logger.info("Creating Temporal Runtime with OTLP metrics at %s", otlp_endpoint)
        metrics = OpenTelemetryConfig(url=otlp_endpoint, headers=headers or None, http=use_http)

    telemetry_kwargs: dict[str, Any] = {"metrics": metrics, "global_tags": global_tags or {}}
    if core_log_level is not None:
        telemetry_kwargs["logging"] = LoggingConfig(
            filter=TelemetryFilter(core_level=core_log_level, other_level=core_log_level),
        )
    try:
        self._runtime = Runtime(
            telemetry=TelemetryConfig(**telemetry_kwargs),
            disable_environment_info=disable_environment_info,
        )
        logger.info("Temporal Runtime created successfully")
    except Exception as error:
        # Falling back to the default Runtime would drop every SDK metric without a trace, so fail loudly.
        raise ConfigurationError(
            operation="temporal_runtime",
            reason=f"cannot build the Temporal Runtime telemetry: {error}",
        ) from error

    return self._runtime

archipy.adapters.temporal.runtime.TemporalRuntimeManager.reset_runtime

reset_runtime() -> None

Reset the Runtime instance.

Warning

This does NOT actually close the Runtime or release telemetry resources. The Temporal SDK does not support Runtime cleanup. This method only resets internal references for testing purposes.

Source code in packages/temporal/archipy/adapters/temporal/runtime.py

def reset_runtime(self) -> None:
    """Reset the Runtime instance.

    Warning:
        This does NOT actually close the Runtime or release telemetry resources.
        The Temporal SDK does not support Runtime cleanup. This method only
        resets internal references for testing purposes.
    """
    logger.warning("Resetting Temporal Runtime reference (resources remain until process exit)")
    self._runtime = None

options: show_root_toc_entry: false heading_level: 3

Worker

Temporal worker setup for executing workflows and activities.

Worker management for Temporal workflow execution.

This module provides worker management functionality for Temporal workflow orchestration, including worker lifecycle management, task queue assignment, and integration with ArchiPy service adapters.

archipy.adapters.temporal.worker.SHUTDOWN_MARGIN_SECONDS module-attribute

SHUTDOWN_MARGIN_SECONDS = 10

archipy.adapters.temporal.worker.WorkerHandle

Bases: WorkerHandle

Handle for managing a Temporal worker instance.

Provides methods to control and monitor a running Temporal worker, including starting, stopping, and querying worker status.

Attributes:

Name	Type	Description
worker_id	str	Unique identifier for this worker instance.
task_queue	str	The task queue this worker polls from.
workflows	list[type]	List of workflow types registered with this worker.
activities	list[Callable]	List of activity callables registered with this worker.
build_id	str | None	Build identifier for worker versioning.
identity	str | None	Worker identity for debugging and monitoring.
max_concurrent_workflow_tasks	int	Maximum concurrent workflow tasks.
max_concurrent_activities	int	Maximum concurrent activity tasks.

Source code in packages/temporal/archipy/adapters/temporal/worker.py

class WorkerHandle(PortWorkerHandle):
    """Handle for managing a Temporal worker instance.

    Provides methods to control and monitor a running Temporal worker,
    including starting, stopping, and querying worker status.

    Attributes:
        worker_id (str): Unique identifier for this worker instance.
        task_queue (str): The task queue this worker polls from.
        workflows (list[type]): List of workflow types registered with this worker.
        activities (list[Callable]): List of activity callables registered with this worker.
        build_id (str | None): Build identifier for worker versioning.
        identity (str | None): Worker identity for debugging and monitoring.
        max_concurrent_workflow_tasks (int): Maximum concurrent workflow tasks.
        max_concurrent_activities (int): Maximum concurrent activity tasks.
    """

    def __init__(
        self,
        worker: Worker,
        worker_id: str,
        task_queue: str,
        workflows: list[type] | None = None,
        activities: list[Callable[..., Any]] | None = None,
        build_id: str | None = None,
        identity: str | None = None,
        max_concurrent_workflow_tasks: int | None = None,
        max_concurrent_activities: int | None = None,
        graceful_shutdown_seconds: int = 30,
    ) -> None:
        """Initialize the worker handle.

        Args:
            worker (Worker): The Temporal worker instance.
            worker_id (str): Unique identifier for this worker instance.
            task_queue (str): The task queue this worker polls from.
            workflows (list[type], optional): List of workflow types. Defaults to None.
            activities (list[Callable], optional): List of activity callables. Defaults to None.
            build_id (str, optional): Build identifier for worker versioning. Defaults to None.
            identity (str, optional): Worker identity. Defaults to None.
            max_concurrent_workflow_tasks (int, optional): Maximum concurrent workflow tasks.
                Defaults to None.
            max_concurrent_activities (int, optional): Maximum concurrent activity tasks.
                Defaults to None.
            graceful_shutdown_seconds (int): Default graceful shutdown timeout in seconds.
                Defaults to 30.
        """
        self._worker = worker
        self.worker_id = worker_id
        self.task_queue = task_queue
        self.workflows = workflows or []
        self.activities = activities or []
        self.build_id = build_id
        self.identity = identity
        self.max_concurrent_workflow_tasks = max_concurrent_workflow_tasks
        self.max_concurrent_activities = max_concurrent_activities
        self._graceful_shutdown_seconds = graceful_shutdown_seconds
        self._running = False
        self._run_error: BaseException | None = None
        self._logger = logging.getLogger(__name__)

    async def start(self) -> None:
        """Start the worker to begin polling for tasks.

        Raises:
            WorkerConnectionError: If the worker fails to start.
        """
        try:
            self._logger.info(
                "Starting worker",
                extra={
                    "worker_id": self.worker_id,
                    "task_queue": self.task_queue,
                    "identity": self.identity,
                },
            )

            # Start the worker in the background
            task = asyncio.create_task(self._worker.run())
            # Store task reference to avoid it being garbage collected
            self._background_task = task
            self._running = True
            task.add_done_callback(self._on_worker_done)

            self._logger.info(
                "Worker started successfully",
                extra={
                    "worker_id": self.worker_id,
                    "task_queue": self.task_queue,
                },
            )

        except Exception as error:  # worker lifecycle; temporal/runtime may raise broadly
            raise WorkerConnectionError(
                additional_data={
                    "message": f"Failed to start worker for task queue '{self.task_queue}'",
                    "task_queue": self.task_queue,
                    "worker_id": self.worker_id,
                    "error": str(error),
                },
            ) from error

    def _on_worker_done(self, task: asyncio.Task[None]) -> None:
        """Mark the worker stopped when ``Worker.run`` returns, and log the error that ended it.

        Args:
            task (asyncio.Task[None]): The finished ``Worker.run`` task.
        """
        self._running = False
        if task.cancelled():
            return
        error = task.exception()
        if error is not None:
            self._run_error = error
            self._logger.error(
                "Worker stopped unexpectedly",
                exc_info=error,
                extra={"worker_id": self.worker_id, "task_queue": self.task_queue},
            )

    @property
    def run_error(self) -> BaseException | None:
        """The error that ended ``Worker.run``, if it crashed.

        Returns:
            BaseException | None: The error, or None while running or after a clean stop.
        """
        return self._run_error

    async def stop(self, grace_period: int | None = None) -> None:
        """Stop the worker gracefully.

        Args:
            grace_period (int, optional): Seconds to let running activities finish before the SDK cancels them,
                as set by ``graceful_shutdown_timeout``. If None, uses the configured graceful shutdown timeout.
                The call gives up ``SHUTDOWN_MARGIN_SECONDS`` after that, so the SDK's own cancellation completes.

        Raises:
            WorkerShutdownError: If the worker fails to stop gracefully.
        """
        if grace_period is None:
            grace_period = self._graceful_shutdown_seconds

        if not self._running:
            return

        try:
            self._logger.info(
                "Stopping worker",
                extra={
                    "worker_id": self.worker_id,
                    "task_queue": self.task_queue,
                    "grace_period": grace_period,
                },
            )

            # The SDK enforces the grace period itself (cancelling activities when it elapses); the outer timeout only
            # guards against activities that ignore cancellation, so it must not cut the SDK's shutdown short.
            await asyncio.wait_for(self._worker.shutdown(), timeout=grace_period + SHUTDOWN_MARGIN_SECONDS)
            self._running = False

            self._logger.info(
                "Worker stopped successfully",
                extra={
                    "worker_id": self.worker_id,
                    "task_queue": self.task_queue,
                },
            )

        except TimeoutError as error:
            raise WorkerShutdownError(
                additional_data={
                    "message": f"Worker shutdown timeout after {grace_period} seconds",
                    "worker_identity": self.identity,
                    "task_queue": self.task_queue,
                    "worker_id": self.worker_id,
                    "grace_period": grace_period,
                },
            ) from error
        except Exception as error:  # worker lifecycle; temporal/runtime may raise broadly
            raise WorkerShutdownError(
                additional_data={
                    "message": f"Failed to stop worker for task queue '{self.task_queue}'",
                    "worker_identity": self.identity,
                    "task_queue": self.task_queue,
                    "worker_id": self.worker_id,
                    "error": str(error),
                },
            ) from error

    async def wait_until_stopped(self) -> None:
        """Wait until the worker is stopped."""
        await self._background_task

    @property
    def is_running(self) -> bool:
        """Check if the worker is currently running.

        Returns:
            bool: True if the worker is running, False otherwise.
        """
        return self._running

    def get_stats(self) -> dict[str, Any]:
        """Get worker statistics and status information.

        Returns:
            dict[str, Any]: Worker statistics and status.
        """
        return {
            "worker_id": self.worker_id,
            "task_queue": self.task_queue,
            "identity": self.identity,
            "build_id": self.build_id,
            "is_running": self.is_running,
            "workflow_count": len(self.workflows),
            "activity_count": len(self.activities),
            "max_concurrent_workflow_tasks": self.max_concurrent_workflow_tasks,
            "max_concurrent_activities": self.max_concurrent_activities,
        }

archipy.adapters.temporal.worker.WorkerHandle.worker_id instance-attribute

worker_id = worker_id

archipy.adapters.temporal.worker.WorkerHandle.task_queue instance-attribute

task_queue = task_queue

archipy.adapters.temporal.worker.WorkerHandle.workflows instance-attribute

workflows = workflows or []

archipy.adapters.temporal.worker.WorkerHandle.activities instance-attribute

activities = activities or []

archipy.adapters.temporal.worker.WorkerHandle.build_id instance-attribute

build_id = build_id

archipy.adapters.temporal.worker.WorkerHandle.identity instance-attribute

identity = identity

archipy.adapters.temporal.worker.WorkerHandle.max_concurrent_workflow_tasks instance-attribute

max_concurrent_workflow_tasks = (
    max_concurrent_workflow_tasks
)

archipy.adapters.temporal.worker.WorkerHandle.max_concurrent_activities instance-attribute

max_concurrent_activities = max_concurrent_activities

archipy.adapters.temporal.worker.WorkerHandle.run_error property

run_error: BaseException | None

The error that ended Worker.run, if it crashed.

Returns:

Type	Description
BaseException | None	BaseException | None: The error, or None while running or after a clean stop.

archipy.adapters.temporal.worker.WorkerHandle.is_running property

is_running: bool

Check if the worker is currently running.

Returns:

Name	Type	Description
bool	bool	True if the worker is running, False otherwise.

archipy.adapters.temporal.worker.WorkerHandle.start async

start() -> None

Start the worker to begin polling for tasks.

Raises:

Type	Description
WorkerConnectionError	If the worker fails to start.

Source code in packages/temporal/archipy/adapters/temporal/worker.py

async def start(self) -> None:
    """Start the worker to begin polling for tasks.

    Raises:
        WorkerConnectionError: If the worker fails to start.
    """
    try:
        self._logger.info(
            "Starting worker",
            extra={
                "worker_id": self.worker_id,
                "task_queue": self.task_queue,
                "identity": self.identity,
            },
        )

        # Start the worker in the background
        task = asyncio.create_task(self._worker.run())
        # Store task reference to avoid it being garbage collected
        self._background_task = task
        self._running = True
        task.add_done_callback(self._on_worker_done)

        self._logger.info(
            "Worker started successfully",
            extra={
                "worker_id": self.worker_id,
                "task_queue": self.task_queue,
            },
        )

    except Exception as error:  # worker lifecycle; temporal/runtime may raise broadly
        raise WorkerConnectionError(
            additional_data={
                "message": f"Failed to start worker for task queue '{self.task_queue}'",
                "task_queue": self.task_queue,
                "worker_id": self.worker_id,
                "error": str(error),
            },
        ) from error

archipy.adapters.temporal.worker.WorkerHandle.stop async

stop(grace_period: int | None = None) -> None

Stop the worker gracefully.

Parameters:

Name	Type	Description	Default
grace_period	int	Seconds to let running activities finish before the SDK cancels them, as set by graceful_shutdown_timeout. If None, uses the configured graceful shutdown timeout. The call gives up SHUTDOWN_MARGIN_SECONDS after that, so the SDK's own cancellation completes.	None

Raises:

Type	Description
WorkerShutdownError	If the worker fails to stop gracefully.

Source code in packages/temporal/archipy/adapters/temporal/worker.py

async def stop(self, grace_period: int | None = None) -> None:
    """Stop the worker gracefully.

    Args:
        grace_period (int, optional): Seconds to let running activities finish before the SDK cancels them,
            as set by ``graceful_shutdown_timeout``. If None, uses the configured graceful shutdown timeout.
            The call gives up ``SHUTDOWN_MARGIN_SECONDS`` after that, so the SDK's own cancellation completes.

    Raises:
        WorkerShutdownError: If the worker fails to stop gracefully.
    """
    if grace_period is None:
        grace_period = self._graceful_shutdown_seconds

    if not self._running:
        return

    try:
        self._logger.info(
            "Stopping worker",
            extra={
                "worker_id": self.worker_id,
                "task_queue": self.task_queue,
                "grace_period": grace_period,
            },
        )

        # The SDK enforces the grace period itself (cancelling activities when it elapses); the outer timeout only
        # guards against activities that ignore cancellation, so it must not cut the SDK's shutdown short.
        await asyncio.wait_for(self._worker.shutdown(), timeout=grace_period + SHUTDOWN_MARGIN_SECONDS)
        self._running = False

        self._logger.info(
            "Worker stopped successfully",
            extra={
                "worker_id": self.worker_id,
                "task_queue": self.task_queue,
            },
        )

    except TimeoutError as error:
        raise WorkerShutdownError(
            additional_data={
                "message": f"Worker shutdown timeout after {grace_period} seconds",
                "worker_identity": self.identity,
                "task_queue": self.task_queue,
                "worker_id": self.worker_id,
                "grace_period": grace_period,
            },
        ) from error
    except Exception as error:  # worker lifecycle; temporal/runtime may raise broadly
        raise WorkerShutdownError(
            additional_data={
                "message": f"Failed to stop worker for task queue '{self.task_queue}'",
                "worker_identity": self.identity,
                "task_queue": self.task_queue,
                "worker_id": self.worker_id,
                "error": str(error),
            },
        ) from error

archipy.adapters.temporal.worker.WorkerHandle.wait_until_stopped async

wait_until_stopped() -> None

Wait until the worker is stopped.

Source code in packages/temporal/archipy/adapters/temporal/worker.py

async def wait_until_stopped(self) -> None:
    """Wait until the worker is stopped."""
    await self._background_task

archipy.adapters.temporal.worker.WorkerHandle.get_stats

get_stats() -> dict[str, Any]

Get worker statistics and status information.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Worker statistics and status.

Source code in packages/temporal/archipy/adapters/temporal/worker.py

def get_stats(self) -> dict[str, Any]:
    """Get worker statistics and status information.

    Returns:
        dict[str, Any]: Worker statistics and status.
    """
    return {
        "worker_id": self.worker_id,
        "task_queue": self.task_queue,
        "identity": self.identity,
        "build_id": self.build_id,
        "is_running": self.is_running,
        "workflow_count": len(self.workflows),
        "activity_count": len(self.activities),
        "max_concurrent_workflow_tasks": self.max_concurrent_workflow_tasks,
        "max_concurrent_activities": self.max_concurrent_activities,
    }

archipy.adapters.temporal.worker.TemporalWorkerManager

Bases: WorkerPort

Manager for Temporal worker lifecycle and operations.

This class provides a high-level interface for managing Temporal workers, including creation, configuration, and lifecycle management. It integrates with ArchiPy configuration and service patterns.

Parameters:

Name	Type	Description	Default
temporal_config	TemporalConfig	Configuration settings for Temporal. If None, retrieves from global config. Defaults to None.	None

Source code in packages/temporal/archipy/adapters/temporal/worker.py

class TemporalWorkerManager(WorkerPort):
    """Manager for Temporal worker lifecycle and operations.

    This class provides a high-level interface for managing Temporal workers,
    including creation, configuration, and lifecycle management. It integrates
    with ArchiPy configuration and service patterns.

    Args:
        temporal_config (TemporalConfig, optional): Configuration settings for Temporal.
            If None, retrieves from global config. Defaults to None.
    """

    def __init__(
        self,
        temporal_config: TemporalConfig | None = None,
        data_converter: DataConverter | None = None,
        interceptors: Sequence[Interceptor] | None = None,
    ) -> None:
        """Initialize the worker manager.

        Args:
            temporal_config (TemporalConfig, optional): Configuration settings for Temporal.
                If None, retrieves from global config. Defaults to None.
            data_converter (DataConverter, optional): Data converter passed to the client the workers use.
                Defaults to None (see ``TemporalAdapter``).
            interceptors (Sequence[Interceptor], optional): Client interceptors; those that also implement
                ``temporalio.worker.Interceptor`` apply to every worker this manager starts. Defaults to None.
        """
        # Get temporal config from the global config or use provided one
        if temporal_config is None:
            global_config = BaseConfig.global_config()
            if hasattr(global_config, "TEMPORAL"):
                self.config = global_config.TEMPORAL
            else:
                # Create a default config if none exists

                self.config = TemporalConfig()
        else:
            self.config = temporal_config
        self._temporal_adapter = TemporalAdapter(temporal_config, data_converter, interceptors)
        self._workers: dict[str, WorkerHandle] = {}
        self._logger = logging.getLogger(__name__)

    async def _get_client(self) -> Client:
        """Get the Temporal client from the adapter.

        Returns:
            Client: The Temporal client instance.
        """
        return await self._temporal_adapter.get_client()

    def _poller_behavior(self) -> PollerBehavior:
        """Build the task poller behaviour, autoscaling when enabled in configuration.

        Returns:
            PollerBehavior: Autoscaling pollers, or the SDK default of at most 5 pollers.
        """
        if self.config.WORKER_POLLER_AUTOSCALING:
            return PollerBehaviorAutoscaling()
        return PollerBehaviorSimpleMaximum(maximum=5)

    def _eager_activity_reservations(self) -> int:
        """Resolve how many eager activities one workflow task may reserve.

        Returns:
            int: ``WORKER_MAX_EAGER_ACTIVITY_RESERVATIONS``, or the SDK default when it is unset.
        """
        reservations = self.config.WORKER_MAX_EAGER_ACTIVITY_RESERVATIONS
        return _SDK_EAGER_ACTIVITY_RESERVATIONS if reservations is None else reservations

    @override
    async def start_worker(
        self,
        task_queue: str,
        workflows: list[type] | None = None,
        activities: list[Callable[..., Any]] | None = None,
        build_id: str | None = None,
        identity: str | None = None,
        max_concurrent_workflow_tasks: int | None = None,
        max_concurrent_activities: int | None = None,
        tuner: WorkerTuner | None = None,
        activity_executor: Executor | None = None,
        deployment_config: WorkerDeploymentConfig | None = None,
        workflow_failure_exception_types: Sequence[type[BaseException]] = (),
        workflow_runner: WorkflowRunner | None = None,
    ) -> WorkerHandle:
        """Start a Temporal worker for the specified task queue.

        Args:
            task_queue (str): The task queue this worker will poll from.
            workflows (list[type], optional): List of workflow classes to register.
                Defaults to None.
            activities (list[Callable], optional): List of activity callables to register.
                Defaults to None.
            build_id (str, optional): Deprecated, use ``deployment_config``. Build identifier for worker
                versioning. Defaults to None.
            identity (str, optional): Unique worker identity. If None, auto-generated.
                Defaults to None.
            max_concurrent_workflow_tasks (int, optional): Maximum concurrent workflow tasks.
                Defaults to None (server default).
            max_concurrent_activities (int, optional): Maximum concurrent activity tasks.
                Defaults to None (server default).
            tuner (WorkerTuner, optional): Slot supplier tuner, e.g. ``WorkerTuner.create_resource_based(...)``.
                Cannot be combined with the ``max_concurrent_*`` limits. Defaults to None.
            activity_executor (Executor, optional): Executor that runs synchronous (non-async) activities.
                Required when registering sync activities. Defaults to None.
            deployment_config (WorkerDeploymentConfig, optional): Worker Deployment version this worker belongs
                to, replacing ``build_id``. Defaults to None.
            workflow_failure_exception_types (Sequence[type[BaseException]], optional): Exception types that fail
                the workflow instead of retrying the workflow task (e.g. ``[Exception]`` to fail on any error).
                Defaults to () (only ``FailureError`` subclasses fail it).
            workflow_runner (WorkflowRunner, optional): Workflow runner, e.g. a ``SandboxedWorkflowRunner`` with
                ``restrictions.with_passthrough_modules(...)``. Defaults to None (the SDK sandbox).

        Returns:
            WorkerHandle: Handle to the started worker.

        Raises:
            WorkerConnectionError: If the worker fails to start.
        """
        if not task_queue or not task_queue.strip():
            raise WorkerConnectionError(
                additional_data={
                    "message": "Task queue name cannot be empty",
                    "task_queue": task_queue,
                },
            )

        if build_id is not None:
            if deployment_config is not None:
                raise InvalidArgumentError(
                    argument_name="build_id",
                    additional_data={"reason": "build_id and deployment_config are mutually exclusive"},
                )
            warnings.warn(
                "start_worker(build_id=...) is deprecated; pass deployment_config=WorkerDeploymentConfig(...) instead",
                DeprecationWarning,
                stacklevel=2,
            )

        client = await self._get_client()
        worker_id = str(uuid7())
        # uuid7 starts with the timestamp, so the random tail keeps identities of workers started together apart.
        worker_identity = identity or f"worker-{worker_id[-12:]}"

        try:
            # Create the Temporal worker
            # Note: Worker inherits Runtime (including OTLP metrics config) from the Client
            worker = Worker(
                client,
                task_queue=task_queue,
                workflows=workflows or [],
                activities=activities or [],
                build_id=build_id,
                deployment_config=deployment_config,
                workflow_failure_exception_types=workflow_failure_exception_types,
                workflow_runner=workflow_runner or SandboxedWorkflowRunner(),
                identity=worker_identity,
                max_concurrent_workflow_tasks=max_concurrent_workflow_tasks,
                max_concurrent_activities=max_concurrent_activities,
                max_cached_workflows=self.config.WORKER_MAX_CACHED_WORKFLOWS,
                graceful_shutdown_timeout=timedelta(seconds=self.config.WORKER_GRACEFUL_SHUTDOWN_SECONDS),
                debug_mode=self.config.WORKER_DEBUG_MODE,
                disable_eager_activity_execution=self.config.WORKER_DISABLE_EAGER_ACTIVITY_EXECUTION,
                max_eager_activity_reservations_per_workflow_task=self._eager_activity_reservations(),
                max_concurrent_local_activities=self.config.WORKER_MAX_CONCURRENT_LOCAL_ACTIVITIES,
                tuner=tuner,
                activity_executor=activity_executor,
                max_activities_per_second=self.config.WORKER_MAX_ACTIVITIES_PER_SECOND,
                max_task_queue_activities_per_second=self.config.WORKER_MAX_TASK_QUEUE_ACTIVITIES_PER_SECOND,
                workflow_task_poller_behavior=self._poller_behavior(),
                activity_task_poller_behavior=self._poller_behavior(),
            )

            # Create worker handle
            worker_handle = WorkerHandle(
                worker=worker,
                worker_id=worker_id,
                task_queue=task_queue,
                workflows=workflows,
                activities=activities,
                build_id=build_id or (deployment_config.version.build_id if deployment_config else None),
                identity=worker_identity,
                max_concurrent_workflow_tasks=max_concurrent_workflow_tasks,
                max_concurrent_activities=max_concurrent_activities,
                graceful_shutdown_seconds=self.config.WORKER_GRACEFUL_SHUTDOWN_SECONDS,
            )

            # Start the worker
            await worker_handle.start()

            # Register the worker
            self._workers[worker_id] = worker_handle

            self._logger.info(
                "Worker created and started",
                extra={
                    "worker_id": worker_id,
                    "task_queue": task_queue,
                    "identity": worker_identity,
                    "workflow_count": len(workflows) if workflows else 0,
                    "activity_count": len(activities) if activities else 0,
                },
            )

        except Exception as error:  # worker lifecycle; temporal/runtime may raise broadly
            raise WorkerConnectionError(
                additional_data={
                    "message": f"Failed to start worker for task queue '{task_queue}'",
                    "task_queue": task_queue,
                    "worker_id": worker_id,
                    "identity": worker_identity,
                    "error": str(error),
                },
            ) from error
        else:
            return worker_handle

    @override
    async def stop_worker(self, worker_handle: PortWorkerHandle) -> None:
        """Stop a running Temporal worker.

        Args:
            worker_handle (PortWorkerHandle): Handle to the worker to stop.

        Raises:
            WorkerShutdownError: If the worker fails to stop gracefully.
        """
        if worker_handle.worker_id not in self._workers:
            return  # Worker already stopped or not managed by this manager

        try:
            await worker_handle.stop()
            del self._workers[worker_handle.worker_id]

            self._logger.info(
                "Worker stopped and removed",
                extra={
                    "worker_id": worker_handle.worker_id,
                    "task_queue": worker_handle.task_queue,
                },
            )

        except Exception:  # worker lifecycle; temporal/runtime may raise broadly
            # Remove from tracking even if shutdown failed
            if worker_handle.worker_id in self._workers:
                del self._workers[worker_handle.worker_id]
            raise

    @override
    async def shutdown_all_workers(self) -> None:
        """Shutdown all workers managed by this port.

        Performs graceful shutdown of all active workers, waiting for current
        tasks to complete before terminating.

        Raises:
            WorkerShutdownError: If any worker fails to shutdown gracefully.
        """
        if not self._workers:
            return

        self._logger.info(
            "Shutting down all workers",
            extra={
                "worker_count": len(self._workers),
            },
        )

        shutdown_errors = []
        workers_to_stop = list(self._workers.values())

        # Stop all workers concurrently
        results = await asyncio.gather(
            *(self.stop_worker(worker_handle) for worker_handle in workers_to_stop),
            return_exceptions=True,
        )
        for worker_handle, result in zip(workers_to_stop, results, strict=True):
            if isinstance(result, BaseException):
                shutdown_errors.append(
                    {
                        "worker_id": worker_handle.worker_id,
                        "task_queue": worker_handle.task_queue,
                        "error": str(result),
                    },
                )

        if shutdown_errors:
            raise WorkerShutdownError(
                additional_data={
                    "message": f"Failed to shutdown {len(shutdown_errors)} workers",
                    "worker_count": len(self._workers),
                    "failed_count": len(shutdown_errors),
                    "shutdown_errors": shutdown_errors,
                },
            )

        self._logger.info("All workers shut down successfully")

    def get_worker_stats(self) -> list[dict[str, Any]]:
        """Get statistics for all managed workers.

        Returns:
            list[dict[str, Any]]: List of worker statistics.
        """
        return [worker.get_stats() for worker in self._workers.values()]

    def get_worker_by_task_queue(self, task_queue: str) -> WorkerHandle | None:
        """Get a worker handle by task queue.

        Args:
            task_queue (str): The task queue to search for.

        Returns:
            WorkerHandle | None: Worker handle if found, None otherwise.
        """
        for worker in self._workers.values():
            if worker.task_queue == task_queue:
                return worker
        return None

    def list_workers(self) -> list[WorkerHandle]:
        """Get a list of all managed workers.

        Returns:
            list[WorkerHandle]: List of worker handles.
        """
        return list(self._workers.values())

    @property
    def worker_count(self) -> int:
        """Get the number of managed workers.

        Returns:
            int: Number of managed workers.
        """
        return len(self._workers)

    async def close(self) -> None:
        """Close the worker manager and all managed workers.

        Performs cleanup of all resources, including stopping all workers
        and closing the Temporal client connection.
        """
        await self.shutdown_all_workers()
        await self._temporal_adapter.close()
        self._logger.info("Worker manager closed")

archipy.adapters.temporal.worker.TemporalWorkerManager.config instance-attribute

config = global_config.TEMPORAL

archipy.adapters.temporal.worker.TemporalWorkerManager.worker_count property

worker_count: int

Get the number of managed workers.

Returns:

Name	Type	Description
int	int	Number of managed workers.

archipy.adapters.temporal.worker.TemporalWorkerManager.start_worker async

start_worker(
    task_queue: str,
    workflows: list[type] | None = None,
    activities: list[Callable[..., Any]] | None = None,
    build_id: str | None = None,
    identity: str | None = None,
    max_concurrent_workflow_tasks: int | None = None,
    max_concurrent_activities: int | None = None,
    tuner: WorkerTuner | None = None,
    activity_executor: Executor | None = None,
    deployment_config: WorkerDeploymentConfig | None = None,
    workflow_failure_exception_types: Sequence[
        type[BaseException]
    ] = (),
    workflow_runner: WorkflowRunner | None = None,
) -> WorkerHandle

Start a Temporal worker for the specified task queue.

Parameters:

Name	Type	Description	Default
task_queue	str	The task queue this worker will poll from.	required
workflows	list[type]	List of workflow classes to register. Defaults to None.	None
activities	list[Callable]	List of activity callables to register. Defaults to None.	None
build_id	str	Deprecated, use deployment_config. Build identifier for worker versioning. Defaults to None.	None
identity	str	Unique worker identity. If None, auto-generated. Defaults to None.	None
max_concurrent_workflow_tasks	int	Maximum concurrent workflow tasks. Defaults to None (server default).	None
max_concurrent_activities	int	Maximum concurrent activity tasks. Defaults to None (server default).	None
tuner	WorkerTuner	Slot supplier tuner, e.g. WorkerTuner.create_resource_based(...). Cannot be combined with the max_concurrent_* limits. Defaults to None.	None
activity_executor	Executor	Executor that runs synchronous (non-async) activities. Required when registering sync activities. Defaults to None.	None
deployment_config	WorkerDeploymentConfig	Worker Deployment version this worker belongs to, replacing build_id. Defaults to None.	None
workflow_failure_exception_types	Sequence[type[BaseException]]	Exception types that fail the workflow instead of retrying the workflow task (e.g. [Exception] to fail on any error). Defaults to () (only FailureError subclasses fail it).	()
workflow_runner	WorkflowRunner	Workflow runner, e.g. a SandboxedWorkflowRunner with restrictions.with_passthrough_modules(...). Defaults to None (the SDK sandbox).	None

Returns:

Name	Type	Description
WorkerHandle	WorkerHandle	Handle to the started worker.

Raises:

Type	Description
WorkerConnectionError	If the worker fails to start.

Source code in packages/temporal/archipy/adapters/temporal/worker.py

@override
async def start_worker(
    self,
    task_queue: str,
    workflows: list[type] | None = None,
    activities: list[Callable[..., Any]] | None = None,
    build_id: str | None = None,
    identity: str | None = None,
    max_concurrent_workflow_tasks: int | None = None,
    max_concurrent_activities: int | None = None,
    tuner: WorkerTuner | None = None,
    activity_executor: Executor | None = None,
    deployment_config: WorkerDeploymentConfig | None = None,
    workflow_failure_exception_types: Sequence[type[BaseException]] = (),
    workflow_runner: WorkflowRunner | None = None,
) -> WorkerHandle:
    """Start a Temporal worker for the specified task queue.

    Args:
        task_queue (str): The task queue this worker will poll from.
        workflows (list[type], optional): List of workflow classes to register.
            Defaults to None.
        activities (list[Callable], optional): List of activity callables to register.
            Defaults to None.
        build_id (str, optional): Deprecated, use ``deployment_config``. Build identifier for worker
            versioning. Defaults to None.
        identity (str, optional): Unique worker identity. If None, auto-generated.
            Defaults to None.
        max_concurrent_workflow_tasks (int, optional): Maximum concurrent workflow tasks.
            Defaults to None (server default).
        max_concurrent_activities (int, optional): Maximum concurrent activity tasks.
            Defaults to None (server default).
        tuner (WorkerTuner, optional): Slot supplier tuner, e.g. ``WorkerTuner.create_resource_based(...)``.
            Cannot be combined with the ``max_concurrent_*`` limits. Defaults to None.
        activity_executor (Executor, optional): Executor that runs synchronous (non-async) activities.
            Required when registering sync activities. Defaults to None.
        deployment_config (WorkerDeploymentConfig, optional): Worker Deployment version this worker belongs
            to, replacing ``build_id``. Defaults to None.
        workflow_failure_exception_types (Sequence[type[BaseException]], optional): Exception types that fail
            the workflow instead of retrying the workflow task (e.g. ``[Exception]`` to fail on any error).
            Defaults to () (only ``FailureError`` subclasses fail it).
        workflow_runner (WorkflowRunner, optional): Workflow runner, e.g. a ``SandboxedWorkflowRunner`` with
            ``restrictions.with_passthrough_modules(...)``. Defaults to None (the SDK sandbox).

    Returns:
        WorkerHandle: Handle to the started worker.

    Raises:
        WorkerConnectionError: If the worker fails to start.
    """
    if not task_queue or not task_queue.strip():
        raise WorkerConnectionError(
            additional_data={
                "message": "Task queue name cannot be empty",
                "task_queue": task_queue,
            },
        )

    if build_id is not None:
        if deployment_config is not None:
            raise InvalidArgumentError(
                argument_name="build_id",
                additional_data={"reason": "build_id and deployment_config are mutually exclusive"},
            )
        warnings.warn(
            "start_worker(build_id=...) is deprecated; pass deployment_config=WorkerDeploymentConfig(...) instead",
            DeprecationWarning,
            stacklevel=2,
        )

    client = await self._get_client()
    worker_id = str(uuid7())
    # uuid7 starts with the timestamp, so the random tail keeps identities of workers started together apart.
    worker_identity = identity or f"worker-{worker_id[-12:]}"

    try:
        # Create the Temporal worker
        # Note: Worker inherits Runtime (including OTLP metrics config) from the Client
        worker = Worker(
            client,
            task_queue=task_queue,
            workflows=workflows or [],
            activities=activities or [],
            build_id=build_id,
            deployment_config=deployment_config,
            workflow_failure_exception_types=workflow_failure_exception_types,
            workflow_runner=workflow_runner or SandboxedWorkflowRunner(),
            identity=worker_identity,
            max_concurrent_workflow_tasks=max_concurrent_workflow_tasks,
            max_concurrent_activities=max_concurrent_activities,
            max_cached_workflows=self.config.WORKER_MAX_CACHED_WORKFLOWS,
            graceful_shutdown_timeout=timedelta(seconds=self.config.WORKER_GRACEFUL_SHUTDOWN_SECONDS),
            debug_mode=self.config.WORKER_DEBUG_MODE,
            disable_eager_activity_execution=self.config.WORKER_DISABLE_EAGER_ACTIVITY_EXECUTION,
            max_eager_activity_reservations_per_workflow_task=self._eager_activity_reservations(),
            max_concurrent_local_activities=self.config.WORKER_MAX_CONCURRENT_LOCAL_ACTIVITIES,
            tuner=tuner,
            activity_executor=activity_executor,
            max_activities_per_second=self.config.WORKER_MAX_ACTIVITIES_PER_SECOND,
            max_task_queue_activities_per_second=self.config.WORKER_MAX_TASK_QUEUE_ACTIVITIES_PER_SECOND,
            workflow_task_poller_behavior=self._poller_behavior(),
            activity_task_poller_behavior=self._poller_behavior(),
        )

        # Create worker handle
        worker_handle = WorkerHandle(
            worker=worker,
            worker_id=worker_id,
            task_queue=task_queue,
            workflows=workflows,
            activities=activities,
            build_id=build_id or (deployment_config.version.build_id if deployment_config else None),
            identity=worker_identity,
            max_concurrent_workflow_tasks=max_concurrent_workflow_tasks,
            max_concurrent_activities=max_concurrent_activities,
            graceful_shutdown_seconds=self.config.WORKER_GRACEFUL_SHUTDOWN_SECONDS,
        )

        # Start the worker
        await worker_handle.start()

        # Register the worker
        self._workers[worker_id] = worker_handle

        self._logger.info(
            "Worker created and started",
            extra={
                "worker_id": worker_id,
                "task_queue": task_queue,
                "identity": worker_identity,
                "workflow_count": len(workflows) if workflows else 0,
                "activity_count": len(activities) if activities else 0,
            },
        )

    except Exception as error:  # worker lifecycle; temporal/runtime may raise broadly
        raise WorkerConnectionError(
            additional_data={
                "message": f"Failed to start worker for task queue '{task_queue}'",
                "task_queue": task_queue,
                "worker_id": worker_id,
                "identity": worker_identity,
                "error": str(error),
            },
        ) from error
    else:
        return worker_handle

archipy.adapters.temporal.worker.TemporalWorkerManager.stop_worker async

stop_worker(worker_handle: WorkerHandle) -> None

Stop a running Temporal worker.

Parameters:

Name	Type	Description	Default
worker_handle	WorkerHandle	Handle to the worker to stop.	required

Raises:

Type	Description
WorkerShutdownError	If the worker fails to stop gracefully.

Source code in packages/temporal/archipy/adapters/temporal/worker.py

@override
async def stop_worker(self, worker_handle: PortWorkerHandle) -> None:
    """Stop a running Temporal worker.

    Args:
        worker_handle (PortWorkerHandle): Handle to the worker to stop.

    Raises:
        WorkerShutdownError: If the worker fails to stop gracefully.
    """
    if worker_handle.worker_id not in self._workers:
        return  # Worker already stopped or not managed by this manager

    try:
        await worker_handle.stop()
        del self._workers[worker_handle.worker_id]

        self._logger.info(
            "Worker stopped and removed",
            extra={
                "worker_id": worker_handle.worker_id,
                "task_queue": worker_handle.task_queue,
            },
        )

    except Exception:  # worker lifecycle; temporal/runtime may raise broadly
        # Remove from tracking even if shutdown failed
        if worker_handle.worker_id in self._workers:
            del self._workers[worker_handle.worker_id]
        raise

archipy.adapters.temporal.worker.TemporalWorkerManager.shutdown_all_workers async

shutdown_all_workers() -> None

Shutdown all workers managed by this port.

Performs graceful shutdown of all active workers, waiting for current tasks to complete before terminating.

Raises:

Type	Description
WorkerShutdownError	If any worker fails to shutdown gracefully.

Source code in packages/temporal/archipy/adapters/temporal/worker.py

@override
async def shutdown_all_workers(self) -> None:
    """Shutdown all workers managed by this port.

    Performs graceful shutdown of all active workers, waiting for current
    tasks to complete before terminating.

    Raises:
        WorkerShutdownError: If any worker fails to shutdown gracefully.
    """
    if not self._workers:
        return

    self._logger.info(
        "Shutting down all workers",
        extra={
            "worker_count": len(self._workers),
        },
    )

    shutdown_errors = []
    workers_to_stop = list(self._workers.values())

    # Stop all workers concurrently
    results = await asyncio.gather(
        *(self.stop_worker(worker_handle) for worker_handle in workers_to_stop),
        return_exceptions=True,
    )
    for worker_handle, result in zip(workers_to_stop, results, strict=True):
        if isinstance(result, BaseException):
            shutdown_errors.append(
                {
                    "worker_id": worker_handle.worker_id,
                    "task_queue": worker_handle.task_queue,
                    "error": str(result),
                },
            )

    if shutdown_errors:
        raise WorkerShutdownError(
            additional_data={
                "message": f"Failed to shutdown {len(shutdown_errors)} workers",
                "worker_count": len(self._workers),
                "failed_count": len(shutdown_errors),
                "shutdown_errors": shutdown_errors,
            },
        )

    self._logger.info("All workers shut down successfully")

archipy.adapters.temporal.worker.TemporalWorkerManager.get_worker_stats

get_worker_stats() -> list[dict[str, Any]]

Get statistics for all managed workers.

Returns:

Type	Description
list[dict[str, Any]]	list[dict[str, Any]]: List of worker statistics.

Source code in packages/temporal/archipy/adapters/temporal/worker.py

def get_worker_stats(self) -> list[dict[str, Any]]:
    """Get statistics for all managed workers.

    Returns:
        list[dict[str, Any]]: List of worker statistics.
    """
    return [worker.get_stats() for worker in self._workers.values()]

archipy.adapters.temporal.worker.TemporalWorkerManager.get_worker_by_task_queue

get_worker_by_task_queue(
    task_queue: str,
) -> WorkerHandle | None

Get a worker handle by task queue.

Parameters:

Name	Type	Description	Default
task_queue	str	The task queue to search for.	required

Returns:

Type	Description
WorkerHandle | None	WorkerHandle | None: Worker handle if found, None otherwise.

Source code in packages/temporal/archipy/adapters/temporal/worker.py

def get_worker_by_task_queue(self, task_queue: str) -> WorkerHandle | None:
    """Get a worker handle by task queue.

    Args:
        task_queue (str): The task queue to search for.

    Returns:
        WorkerHandle | None: Worker handle if found, None otherwise.
    """
    for worker in self._workers.values():
        if worker.task_queue == task_queue:
            return worker
    return None

archipy.adapters.temporal.worker.TemporalWorkerManager.list_workers

list_workers() -> list[WorkerHandle]

Get a list of all managed workers.

Returns:

Type	Description
list[WorkerHandle]	list[WorkerHandle]: List of worker handles.

Source code in packages/temporal/archipy/adapters/temporal/worker.py

def list_workers(self) -> list[WorkerHandle]:
    """Get a list of all managed workers.

    Returns:
        list[WorkerHandle]: List of worker handles.
    """
    return list(self._workers.values())

archipy.adapters.temporal.worker.TemporalWorkerManager.close async

close() -> None

Close the worker manager and all managed workers.

Performs cleanup of all resources, including stopping all workers and closing the Temporal client connection.

Source code in packages/temporal/archipy/adapters/temporal/worker.py

async def close(self) -> None:
    """Close the worker manager and all managed workers.

    Performs cleanup of all resources, including stopping all workers
    and closing the Temporal client connection.
    """
    await self.shutdown_all_workers()
    await self._temporal_adapter.close()
    self._logger.info("Worker manager closed")

options: show_root_toc_entry: false heading_level: 3