TaskEnvironment

Define an execution environment for a set of tasks.

Attributes

Attribute	Type	Description
cache	CacheRequest = "disable"	Cache policy — "auto", "override", "disable", or a Cache object. Also settable in @env.task(cache=...) and task.override(cache=...).
reusable	`ReusePolicy	None` = None
plugin_config	Optional[Any] = None	Plugin configuration for custom task types (e.g., Ray, Spark). Cannot be combined with reusable. TaskEnvironment level only.
queue	Optional[str] = None	Queue name for scheduling. Queues identify specific partitions of your compute infrastructure (e.g., a particular cluster in a multi-cluster deployment) and are configured as part of your Flyte/Union deployment. Also settable in @env.task and task.override.

---

Methods

---

clone_with()

def clone_with(
        self,
        name: str,
        image: Optional[Union[str, Image, Literal["auto"]]] = None,
        resources: Optional[Resources] = None,
        env_vars: Optional[Dict[str, str]] = None,
        secrets: Optional[SecretRequest] = None,
        depends_on: Optional[List[Environment]] = None,
        description: Optional[str] = None,
        interruptible: Optional[bool] = None,
        include: Optional[Tuple[str, ...]] = None,
        **kwargs: Any,
    ) -> TaskEnvironment: ...

Create a new TaskEnvironment that shares most settings with this one but differs in name and selected overrides.

Parameters

Name	Type	Description
name	str	Name for the new environment (required).
image	Optional[Union[str, [Image](../../image/image.md?sid=flyte__image_image), Literal['auto']]] = None	Override the container image.
resources	Optional[[Resources](../../resources/resources.md?sid=flyte__resources_resources)] = None	Override compute resources.
env_vars	Optional[Dict[str, str]] = None	Override environment variables.
secrets	Optional[SecretRequest] = None	Override secrets.
depends_on	Optional[List[[Environment](../../environment/environment.md?sid=flyte__environment_environment)]] = None	Override deployment dependencies.
description	Optional[str] = None	Override the description.
interruptible	Optional[bool] = None	Override the interruptible setting.
include	Optional[Tuple[str, ...]] = None	Specific file or module inclusion paths for the new environment.
kwargs	Any	Additional TaskEnvironment-specific overrides (e.g., cache, reusable, plugin_config).

Returns

Type	Description
[TaskEnvironment](taskenvironment.md?sid=flyte__task_environment_taskenvironment)	A new TaskEnvironment instance incorporating inherited and overridden configuration settings.

---

task()

@overload
def task(
        self,
        *,
        short_name: Optional[str] = None,
        cache: CacheRequest | None = None,
        retries: Union[int, RetryStrategy] = 0,
        timeout: Union[timedelta, int] = 0,
        docs: Optional[Documentation] = None,
        pod_template: Optional[Union[str, PodTemplate]] = None,
        report: bool = False,
        interruptible: bool | None = None,
        max_inline_io_bytes: int = MAX_INLINE_IO_BYTES,
        queue: Optional[str] = None,
        triggers: Tuple[Trigger, ...] | Trigger = (),
        links: Tuple[Link, ...] | Link = (),
        task_resolver: Any | None = None,
        entrypoint: bool = False,
    ) -> Callable[[Callable[P, R]], AsyncFunctionTaskTemplate[P, R, Callable[P, R]]]: ...

@overload
    def task(
        self,
        _func: Callable[P, R],
        /,
    ) -> AsyncFunctionTaskTemplate[P, R, Callable[P, R]]: ...

def task(
        self,
        _func: F | None = None,
        *,
        short_name: Optional[str] = None,
        cache: CacheRequest | None = None,
        retries: Union[int, RetryStrategy] = 0,
        timeout: Union[timedelta, int] = 0,
        docs: Optional[Documentation] = None,
        pod_template: Optional[Union[str, PodTemplate]] = None,
        report: bool = False,
        interruptible: bool | None = None,
        max_inline_io_bytes: int = MAX_INLINE_IO_BYTES,
        queue: Optional[str] = None,
        triggers: Tuple[Trigger, ...] | Trigger = (),
        links: Tuple[Link, ...] | Link = (),
        task_resolver: Any | None = None,
        entrypoint: bool = False,
    ) -> Callable[[F], AsyncFunctionTaskTemplate[P, R, F]] | AsyncFunctionTaskTemplate[P, R, F]: ...

Decorate a function to be a task.

Parameters

Name	Type	Description
_func	`F	None` = None
short_name	Optional[str] = None	Optional friendly name for the task or action, used in parts of the UI (defaults to the function name). Overriding short_name does not change the task's fully-qualified name.
cache	`CacheRequest	None` = None
retries	Union[int, [RetryStrategy](../../retry/retrystrategy.md?sid=flyte__retry_retrystrategy)] = 0	Number of retries (int) or a RetryStrategy object that defines retry behavior. Defaults to 0 (no retries).
timeout	Union[timedelta, int] = 0	Task timeout, as a timedelta object or an integer number of seconds. 0 means no timeout.
docs	Optional[Documentation] = None	Optional The documentation for the task, if not provided the function docstring will be used.
pod_template	Optional[Union[str, PodTemplate]] = None	Optional The pod template for the task, if not provided the default pod template will be used.
report	bool = False	Optional Whether to generate the html report for the task, defaults to False.
interruptible	`bool	None` = None
max_inline_io_bytes	int = MAX_INLINE_IO_BYTES	Maximum allowed size (in bytes) for all inputs and outputs passed directly to the task (e.g., primitives, strings, dicts). Does not apply to files, directories, or dataframes. Default is 10 MiB.
queue	Optional[str] = None	Optional queue name to use for this task. If not set, the environment's queue will be used.
triggers	`Tuple[Trigger, ...]	Trigger` = ()
links	`Tuple[Link, ...]	Link` = ()
task_resolver	`Any	None` = None
entrypoint	bool = False	Optionally mark a task as an entrypoint task, defaults to False. This serves as a hint to the UI.

Returns

Type	Description
`Callable[[F], AsyncFunctionTaskTemplate[P, R, F]]	AsyncFunctionTaskTemplate[P, R, F]`

---

sandbox()

@property
def sandbox(self) -> _SandboxNamespace: ...

Access the sandbox namespace for creating sandboxed tasks.

Returns

Type	Description
_SandboxNamespace	A namespace wrapper bound to this environment for creating sandboxed tasks.

---

tasks()

@property
def tasks(self) -> Dict[str, TaskTemplate]: ...

Get all tasks defined in the environment.

Returns

Type	Description
Dict[str, TaskTemplate]	A dictionary mapping fully-qualified task names to their corresponding TaskTemplate instances.

---

from_task()

@classmethod
def from_task(
        cls,
        name: str,
        *tasks: TaskTemplate,
        depends_on: Optional[List["Environment"]] = None,
    ) -> TaskEnvironment: ...

Create a TaskEnvironment from a list of tasks. All tasks should have the same image or no Image defined. Similarity of Image is determined by the python reference, not by value. If images are different, an error is raised. If no image is defined, the image is set to "auto". For any other tasks that need to be use these tasks, the returned environment can be used in the depends_on attribute of the other TaskEnvironment.

Parameters

Name	Type	Description
name	str	The name of the environment.
tasks	TaskTemplate	The list of tasks to create the environment from.
depends_on	Optional[List[[Environment](../../environment/environment.md?sid=flyte__environment_environment)]] = None	Optional list of environments that this environment depends on.

Returns

Type	Description
[TaskEnvironment](taskenvironment.md?sid=flyte__task_environment_taskenvironment)	The created TaskEnvironment.

---