temporalio.contrib.strands._sandbox_activity.SandboxActivities

class documentation

class SandboxActivities: (source)

Constructor: SandboxActivities(factories, resolvable_worker_env_vars)

View In Hierarchy

Resolves Workflow-scoped sandboxes and exposes their activities.

Method __init__ Store named Workflow-scoped worker-side sandbox factories.
Method activities Build one stable activity set that dispatches by sandbox name.
Async Method _run_stream Undocumented
Async Method _sandbox Undocumented
Instance Variable _factories Undocumented
Instance Variable _worker_env_refs Undocumented
def __init__(self, factories: dict[str, SandboxFactory], resolvable_worker_env_vars: Collection[str] | AllowAllWorkerEnvVars = ()): (source)

Store named Workflow-scoped worker-side sandbox factories.

def activities(self) -> list[Callable[..., Any]]: (source)

Build one stable activity set that dispatches by sandbox name.

async def _run_stream(self, stream: AsyncGenerator[StreamChunk | ExecutionResult, None], *, timeout: float | None, sandbox_name: str, streaming_topic: str | None, streaming_batch_interval_seconds: float) -> list[_StreamItem]: (source)

Undocumented

async def _sandbox(self, input: _WorkflowScopedInput) -> AsyncGenerator[Sandbox, None]: (source)

Undocumented

_factories = (source)

Undocumented

_worker_env_refs = (source)

Undocumented