orchestrator_from_str

Create a reusable sandboxed task from a code string.

def orchestrator_from_str(
    source: str,
    inputs: Dict[str, type],
    output: type = type(None),
    tasks: Optional[List[Any]] = None,
    name: str = "sandboxed-code",
    timeout_ms: int = 30_000,
    cache: CacheRequest = "disable",
    retries: int = 0,
    image: Optional[Any] = None
) - > CodeTaskTemplate

Create a reusable sandboxed task from a code string. The returned CodeTaskTemplate can be passed to flyte.run() just like a decorated task. The last expression in source becomes the return value.

Parameters

Name	Type	Description
source	str	Python code string to execute in the sandbox.
inputs	Dict[str, type]	Mapping of input names to their types.
output	type = type(None)	The return type (default NoneType).
tasks	Optional[List[Any]] = None	List of external functions (tasks, durable ops) available inside the sandbox. Each item's __name__ is used as the key.
name	str = "sandboxed-code"	Task name (default "sandboxed-code").
timeout_ms	int = 30_000	Sandbox execution timeout in milliseconds.
cache	CacheRequest = "disable"	Cache policy for the task.
retries	int = 0	Number of retries on failure.
image	Optional[Any] = None	Docker image to use. If not provided, a default Debian image with pydantic-monty is created automatically.

Returns

Type	Description
CodeTaskTemplate