TaskContext

A context class to hold the current task executions context. This can be used to access various contextual parameters in the task execution by the user.

Attributes

Attribute	Type	Description
action	ActionID	The action ID of the current execution. This is always set, within a run.
version	str	The version of the executed task. This is set when the task is executed by an action and will be set on all sub-actions.
raw_data_path	RawDataPath	Storage location prefix used for reading and writing raw task data.
input_path	`str	None` = None
output_path	str	Path where task execution outputs and results are written.
run_base_dir	str	Root base directory allocated on the local filesystem for the current execution run.
report	Report	Report collector used to capture artifacts and render execution summaries.
group_data	`GroupData	None` = None
checkpoint_paths	`CheckpointPaths	None` = None
code_bundle	`CodeBundle	None` = None
compiled_image_cache	`ImageCache	None` = None
data	Dict[str, Any] = {}	Dictionary of runtime data and cached helpers shared across the task lifecycle.
mode	Literal["local", "remote", "hybrid"] = "remote"	Execution environment mode representing where the task is run. Must be one of 'local', 'remote', or 'hybrid'.
interactive_mode	bool = False	Flag indicating whether the task is executing within an interactive session.
custom_context	Dict[str, str] = {}	Context metadata for the action. If an action receives context, it'll automatically pass it to any actions it spawns. Context will not be used for cache key computation.
disable_run_cache	bool = False	Flag specifying whether caching across task runs is disabled.
in_driver_literal_conversion	bool = False	Set by the runtime during nested-task literal marshalling; type transformers may use it to skip duplicate side effects (e.g. report tabs) outside true task-body I/O.

---

Constructor

Signature

def TaskContext(
    action: ActionID,
    version: str,
    raw_data_path: RawDataPath,
    input_path: str | None = None,
    output_path: str,
    run_base_dir: str,
    report: Report,
    group_data: GroupData | None = None,
    checkpoint_paths: CheckpointPaths | None = None,
    code_bundle: CodeBundle | None = None,
    compiled_image_cache: ImageCache | None = None,
    data: Dict[str, Any] = <factory: dict>,
    mode: Literal['local', 'remote', 'hybrid'] = "remote",
    interactive_mode: bool = False,
    custom_context: Dict[str, str] = <factory: dict>,
    disable_run_cache: bool = False,
    in_driver_literal_conversion: bool = False
) -> None

Parameters

Name	Type	Description
action	ActionID	The action ID of the current execution. Always set within a run.
version	str	The version of the executed task.
raw_data_path	RawDataPath	The path for raw data.
input_path	`str	None` = None
output_path	str	The output path for the task execution.
run_base_dir	str	The base directory for the run.
report	Report	The report object for the execution.
group_data	`GroupData	None` = None
checkpoint_paths	`CheckpointPaths	None` = None
code_bundle	`CodeBundle	None` = None
compiled_image_cache	`ImageCache	None` = None
data	Dict[str, Any] = < factory: dict >	Dictionary holding contextual execution data.
mode	Literal['local', 'remote', 'hybrid'] = "remote"	The execution mode of the task.
interactive_mode	bool = False	Flag indicating whether interactive mode is enabled.
custom_context	Dict[str, str] = < factory: dict >	Context metadata for the action.
disable_run_cache	bool = False	Flag indicating whether run caching is disabled.
in_driver_literal_conversion	bool = False	Flag indicating whether the runtime is converting literals for nested-task / driver orchestration.

---

Methods

---

replace()

def replace(self, **kwargs) -> TaskContext: ...

Creates a new copy of the TaskContext with specified fields replaced or updated, merging any provided data dictionary with the existing context data.

Parameters

Name	Type	Description
**kwargs	Any	Field values to replace or update on the context instance.

Returns

Type	Description
[TaskContext](taskcontext.md?sid=flyte_models_taskcontext)	A new TaskContext instance containing the updated field values.

---

is_in_cluster()

def is_in_cluster(self): ...

Check if the task is running in a cluster.

Returns

Type	Description
bool	True if the task is running in remote mode on a cluster, False otherwise.

---

checkpoint()

@property
def checkpoint(self) -> Optional[Checkpoint]: ...

Task checkpoint helper for the runtime checkpoint_path / prev_checkpoint prefixes. Returns a lazily constructed flyte.Checkpoint cached on flyte.models.TaskContext.data, or None when no checkpoint output prefix is configured. In async tasks use flyte.Checkpoint.load and flyte.Checkpoint.save; in sync tasks use flyte.Checkpoint.load_sync and flyte.Checkpoint.save_sync. For a single raw blob, pass bytes to save; after a successful load, the blob is at checkpoint.path / "payload" when the remote object is not a tarball.

Returns

Type	Description
Optional[[Checkpoint](../checkpoint/checkpoint.md?sid=flyte__checkpoint_checkpoint)]	A lazily constructed Checkpoint instance cached on TaskContext.data, or None when no checkpoint output prefix is configured.

---

attempt_number()

@property
def attempt_number(self) -> int: ...

Get the attempt number for the current task.

Returns

Type	Description
int	The attempt number.

---