flytekit.core.base_task
==============================
:mod:flytekit.core.base_task
.. currentmodule:: flytekit.core.base_task
.. autosummary::
:nosignatures:
:template: custom.rst
:toctree: generated/
kwtypes
PythonTask
Task
TaskResolverMixin
IgnoreOutputs
Directory
Classes
Class |
Description |
Any |
Special type indicating an unconstrained type. |
DeckField |
DeckField is used to specify the fields that will be rendered in the deck. |
Description |
Full user description with formatting preserved. |
Documentation |
DescriptionEntity contains detailed description for the task/workflow/launch plan. |
ExecutionParameters |
This is a run-time user-centric context object that is accessible to every @task method. |
ExecutionState |
This is the context that is active when executing a task or a local workflow. |
FlyteContext |
This is an internal-facing context object, that most users will not have to deal with. |
FlyteContextManager |
FlyteContextManager manages the execution context within Flytekit. |
FlyteEntities |
This is a global Object that tracks various tasks and workflows that are declared within a VM during the. |
Generic |
Abstract base class for generic types. |
Interface |
A Python native interface object, like inspect. |
LocalConfig |
Any configuration specific to local runs. |
LocalTaskCache |
This class implements a persistent store able to cache the result of local task executions. |
Promise |
This object is a wrapper and exists for three main reasons. |
PythonTask |
Base Class for all Tasks with a Python native Interface . |
SecurityContext |
This is a higher level wrapper object that for the most part users shouldn’t have to worry about. |
SerializationSettings |
These settings are provided while serializing a workflow and task, before registration. |
Task |
The base of all Tasks in flytekit. |
TaskMetadata |
Metadata for a Task. |
TaskResolverMixin |
Flytekit tasks interact with the Flyte platform very, very broadly in two steps. |
TrackedInstance |
Please see the notes for the metaclass above first. |
TypeEngine |
Core Extensible TypeEngine of Flytekit. |
TypeVar |
Type variable. |
Variable |
None. |
VoidPromise |
This object is returned for tasks that do not return any outputs (declared interface is empty). |
timeit |
A context manager and a decorator that measures the execution time of the wrapped code block or functions. |
Errors
flytekit.core.base_task.Any
Special type indicating an unconstrained type.
- Any is compatible with every type.
- Any assumed to have all methods.
- All values assumed to be instances of Any.
Note that all the above statements are true from the point of view of
static type checkers. At runtime, Any should not be used with instance
checks.
flytekit.core.base_task.DeckField
DeckField is used to specify the fields that will be rendered in the deck.
def DeckField(
args,
kwds,
):
Parameter |
Type |
args |
*args |
kwds |
|
flytekit.core.base_task.Description
Full user description with formatting preserved. This can be rendered
by clients, such as the console or command line tools with in-tact
formatting.
def Description(
value: typing.Optional[str],
uri: typing.Optional[str],
icon_link: typing.Optional[str],
format: <enum 'DescriptionFormat'>,
):
Parameter |
Type |
value |
typing.Optional[str] |
uri |
typing.Optional[str] |
icon_link |
typing.Optional[str] |
format |
<enum 'DescriptionFormat'> |
Methods
from_flyte_idl()
def from_flyte_idl(
pb2_object: flyteidl.admin.description_entity_pb2.Description,
):
Parameter |
Type |
pb2_object |
flyteidl.admin.description_entity_pb2.Description |
serialize_to_string()
def serialize_to_string()
short_string()
to_flyte_idl()
verbose_string()
Properties
Property |
Type |
Description |
is_empty |
|
|
flytekit.core.base_task.Documentation
DescriptionEntity contains detailed description for the task/workflow/launch plan.
Documentation could provide insight into the algorithms, business use case, etc.
def Documentation(
short_description: typing.Optional[str],
long_description: typing.Optional[flytekit.models.documentation.Description],
source_code: typing.Optional[flytekit.models.documentation.SourceCode],
):
Parameter |
Type |
short_description |
typing.Optional[str] |
long_description |
typing.Optional[flytekit.models.documentation.Description] |
source_code |
typing.Optional[flytekit.models.documentation.SourceCode] |
Methods
from_flyte_idl()
def from_flyte_idl(
pb2_object: flyteidl.admin.description_entity_pb2.DescriptionEntity,
):
Parameter |
Type |
pb2_object |
flyteidl.admin.description_entity_pb2.DescriptionEntity |
serialize_to_string()
def serialize_to_string()
short_string()
to_flyte_idl()
verbose_string()
Properties
Property |
Type |
Description |
is_empty |
|
|
flytekit.core.base_task.ExecutionParameters
This is a run-time user-centric context object that is accessible to every @task method. It can be accessed using
.. code-block:: python
flytekit.current_context()
This object provides the following
- a statsd handler
- a logging handler
- the execution ID as an :py:class:
flytekit.models.core.identifier.WorkflowExecutionIdentifier
object
- a working directory for the user to write arbitrary files to
Please do not confuse this object with the :py:class:flytekit.FlyteContext
object.
def ExecutionParameters(
execution_date,
tmp_dir,
stats,
execution_id: typing.Optional[_identifier.WorkflowExecutionIdentifier],
logging,
raw_output_prefix,
output_metadata_prefix,
checkpoint,
decks,
task_id: typing.Optional[_identifier.Identifier],
enable_deck: bool,
kwargs,
):
Parameter |
Type |
execution_date |
|
tmp_dir |
|
stats |
|
execution_id |
typing.Optional[_identifier.WorkflowExecutionIdentifier] |
logging |
|
raw_output_prefix |
|
output_metadata_prefix |
|
checkpoint |
|
decks |
|
task_id |
typing.Optional[_identifier.Identifier] |
enable_deck |
bool |
kwargs |
**kwargs |
Methods
builder()
get()
Returns task specific context if present else raise an error. The returned context will match the key
has_attr()
def has_attr(
attr_name: str,
):
Parameter |
Type |
attr_name |
str |
new_builder()
def new_builder(
current: Optional[ExecutionParameters],
):
Parameter |
Type |
current |
Optional[ExecutionParameters] |
with_enable_deck()
def with_enable_deck(
enable_deck: bool,
):
Parameter |
Type |
enable_deck |
bool |
with_task_sandbox()
Properties
Property |
Type |
Description |
checkpoint |
|
|
decks |
|
|
default_deck |
|
|
enable_deck |
|
|
execution_date |
|
|
execution_id |
|
|
logging |
|
|
output_metadata_prefix |
|
|
raw_output_prefix |
|
|
secrets |
|
|
stats |
|
|
task_id |
|
|
timeline_deck |
|
|
working_directory |
|
|
flytekit.core.base_task.ExecutionState
This is the context that is active when executing a task or a local workflow. This carries the necessary state to
execute.
Some required things during execution deal with temporary directories, ExecutionParameters that are passed to the
user etc.
Attributes:
mode (ExecutionState.Mode): Defines the context in which the task is executed (local, hosted, etc).
working_dir (os.PathLike): Specifies the remote, external directory where inputs, outputs and other protobufs
are uploaded
engine_dir (os.PathLike):
branch_eval_mode Optional[BranchEvalMode]: Used to determine whether a branch node should execute.
user_space_params Optional[ExecutionParameters]: Provides run-time, user-centric context such as a statsd
handler, a logging handler, the current execution id and a working directory.
def ExecutionState(
working_dir: Union[os.PathLike, str],
mode: Optional[ExecutionState.Mode],
engine_dir: Optional[Union[os.PathLike, str]],
branch_eval_mode: Optional[BranchEvalMode],
user_space_params: Optional[ExecutionParameters],
):
Parameter |
Type |
working_dir |
Union[os.PathLike, str] |
mode |
Optional[ExecutionState.Mode] |
engine_dir |
Optional[Union[os.PathLike, str]] |
branch_eval_mode |
Optional[BranchEvalMode] |
user_space_params |
Optional[ExecutionParameters] |
Methods
Method |
Description |
branch_complete() |
Indicates that we are within a conditional / ifelse block and the active branch is not done |
is_local_execution() |
None |
take_branch() |
Indicates that we are within an if-else block and the current branch has evaluated to true |
with_params() |
Produces a copy of the current execution state and overrides the copy’s parameters with passed parameter values |
branch_complete()
Indicates that we are within a conditional / ifelse block and the active branch is not done.
Default to SKIPPED
is_local_execution()
take_branch()
Indicates that we are within an if-else block and the current branch has evaluated to true.
Useful only in local execution mode
with_params()
def with_params(
working_dir: Optional[os.PathLike],
mode: Optional[Mode],
engine_dir: Optional[os.PathLike],
branch_eval_mode: Optional[BranchEvalMode],
user_space_params: Optional[ExecutionParameters],
):
Produces a copy of the current execution state and overrides the copy’s parameters with passed parameter values.
Parameter |
Type |
working_dir |
Optional[os.PathLike] |
mode |
Optional[Mode] |
engine_dir |
Optional[os.PathLike] |
branch_eval_mode |
Optional[BranchEvalMode] |
user_space_params |
Optional[ExecutionParameters] |
flytekit.core.base_task.FlyteContext
This is an internal-facing context object, that most users will not have to deal with. It’s essentially a globally
available grab bag of settings and objects that allows flytekit to do things like convert complex types, run and
compile workflows, serialize Flyte entities, etc.
Even though this object as a current_context
function on it, it should not be called directly. Please use the
:py:class:flytekit.FlyteContextManager
object instead.
Please do not confuse this object with the :py:class:flytekit.ExecutionParameters
object.
def FlyteContext(
file_access: FileAccessProvider,
level: int,
flyte_client: Optional['friendly_client.SynchronousFlyteClient'],
compilation_state: Optional[CompilationState],
execution_state: Optional[ExecutionState],
serialization_settings: Optional[SerializationSettings],
in_a_condition: bool,
origin_stackframe: Optional[traceback.FrameSummary],
output_metadata_tracker: Optional[OutputMetadataTracker],
worker_queue: Optional[Controller],
):
Parameter |
Type |
file_access |
FileAccessProvider |
level |
int |
flyte_client |
Optional['friendly_client.SynchronousFlyteClient'] |
compilation_state |
Optional[CompilationState] |
execution_state |
Optional[ExecutionState] |
serialization_settings |
Optional[SerializationSettings] |
in_a_condition |
bool |
origin_stackframe |
Optional[traceback.FrameSummary] |
output_metadata_tracker |
Optional[OutputMetadataTracker] |
worker_queue |
Optional[Controller] |
Methods
current_context()
This method exists only to maintain backwards compatibility. Please use
FlyteContextManager.current_context()
instead.
Users of flytekit should be wary not to confuse the object returned from this function
with :py:func:flytekit.current_context
enter_conditional_section()
def enter_conditional_section()
get_deck()
Returns the deck that was created as part of the last execution.
The return value depends on the execution environment. In a notebook, the return value is compatible with
IPython.display and should be rendered in the notebook.
.. code-block:: python
with flytekit.new_context() as ctx:
my_task(…)
ctx.get_deck()
OR if you wish to explicitly display
.. code-block:: python
from IPython import display
display(ctx.get_deck())
get_origin_stackframe_repr()
def get_origin_stackframe_repr()
new_builder()
new_compilation_state()
def new_compilation_state(
prefix: str,
):
Creates and returns a default compilation state. For most of the code this should be the entrypoint
of compilation, otherwise the code should always uses - with_compilation_state
Parameter |
Type |
prefix |
str |
new_execution_state()
def new_execution_state(
working_dir: Optional[os.PathLike],
):
Creates and returns a new default execution state. This should be used at the entrypoint of execution,
in all other cases it is preferable to use with_execution_state
Parameter |
Type |
working_dir |
Optional[os.PathLike] |
set_stackframe()
def set_stackframe(
s: traceback.FrameSummary,
):
Parameter |
Type |
s |
traceback.FrameSummary |
with_client()
def with_client(
c: SynchronousFlyteClient,
):
Parameter |
Type |
c |
SynchronousFlyteClient |
with_compilation_state()
def with_compilation_state(
c: CompilationState,
):
Parameter |
Type |
c |
CompilationState |
with_execution_state()
def with_execution_state(
es: ExecutionState,
):
Parameter |
Type |
es |
ExecutionState |
with_file_access()
def with_file_access(
fa: FileAccessProvider,
):
Parameter |
Type |
fa |
FileAccessProvider |
with_new_compilation_state()
def with_new_compilation_state()
def with_output_metadata_tracker(
t: OutputMetadataTracker,
):
Parameter |
Type |
t |
OutputMetadataTracker |
with_serialization_settings()
def with_serialization_settings(
ss: SerializationSettings,
):
Parameter |
Type |
ss |
SerializationSettings |
with_worker_queue()
def with_worker_queue(
wq: Controller,
):
Parameter |
Type |
wq |
Controller |
Properties
Property |
Type |
Description |
user_space_params |
|
|
flytekit.core.base_task.FlyteContextManager
FlyteContextManager manages the execution context within Flytekit. It holds global state of either compilation
or Execution. It is not thread-safe and can only be run as a single threaded application currently.
Context’s within Flytekit is useful to manage compilation state and execution state. Refer to CompilationState
and ExecutionState
for more information. FlyteContextManager provides a singleton stack to manage these contexts.
Typical usage is
.. code-block:: python
FlyteContextManager.initialize()
with FlyteContextManager.with_context(o) as ctx:
pass
If required - not recommended you can use
FlyteContextManager.push_context()
but correspondingly a pop_context should be called
FlyteContextManager.pop_context()
Methods
add_signal_handler()
def add_signal_handler(
handler: typing.Callable[[int, FrameType], typing.Any],
):
Parameter |
Type |
handler |
typing.Callable[[int, FrameType], typing.Any] |
current_context()
get_origin_stackframe()
def get_origin_stackframe(
limit,
):
initialize()
Re-initializes the context and erases the entire context
pop_context()
push_context()
def push_context(
ctx: FlyteContext,
f: Optional[traceback.FrameSummary],
):
Parameter |
Type |
ctx |
FlyteContext |
f |
Optional[traceback.FrameSummary] |
size()
with_context()
def with_context(
b: FlyteContext.Builder,
):
Parameter |
Type |
b |
FlyteContext.Builder |
flytekit.core.base_task.FlyteDownloadDataException
Common base class for all non-exit exceptions.
def FlyteDownloadDataException(
args,
timestamp: typing.Optional[float],
):
Parameter |
Type |
args |
*args |
timestamp |
typing.Optional[float] |
Properties
Property |
Type |
Description |
timestamp |
|
|
flytekit.core.base_task.FlyteEntities
This is a global Object that tracks various tasks and workflows that are declared within a VM during the
registration process
flytekit.core.base_task.FlyteNonRecoverableSystemException
Common base class for all non-exit exceptions.
def FlyteNonRecoverableSystemException(
exc_value: Exception,
):
FlyteNonRecoverableSystemException is thrown when a system code raises an exception.
Parameter |
Type |
exc_value |
Exception |
Properties
Property |
Type |
Description |
timestamp |
|
|
value |
|
|
flytekit.core.base_task.FlyteUploadDataException
Common base class for all non-exit exceptions.
def FlyteUploadDataException(
args,
timestamp: typing.Optional[float],
):
Parameter |
Type |
args |
*args |
timestamp |
typing.Optional[float] |
Properties
Property |
Type |
Description |
timestamp |
|
|
flytekit.core.base_task.FlyteUserRuntimeException
Common base class for all non-exit exceptions.
def FlyteUserRuntimeException(
exc_value: Exception,
timestamp: typing.Optional[float],
):
FlyteUserRuntimeException is thrown when a user code raises an exception.
Parameter |
Type |
exc_value |
Exception |
timestamp |
typing.Optional[float] |
Properties
Property |
Type |
Description |
timestamp |
|
|
value |
|
|
flytekit.core.base_task.Generic
Abstract base class for generic types.
On Python 3.12 and newer, generic classes implicitly inherit from
Generic when they declare a parameter list after the class’s name::
class Mapping[KT, VT]:
def getitem(self, key: KT) -> VT:
…
Etc.
On older versions of Python, however, generic classes have to
explicitly inherit from Generic.
After a class has been declared to be generic, it can then be used as
follows::
def lookup_name[KT, VT](mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
try:
return mapping[key]
except KeyError:
return default
flytekit.core.base_task.IgnoreOutputs
This exception should be used to indicate that the outputs generated by this can be safely ignored.
This is useful in case of distributed training or peer-to-peer parallel algorithms.
flytekit.core.base_task.Interface
A Python native interface object, like inspect.signature but simpler.
def Interface(
inputs: Union[Optional[Dict[str, Type]], Optional[Dict[str, Tuple[Type, Any]]]],
outputs: Union[Optional[Dict[str, Type]], Optional[Dict[str, Optional[Type]]]],
output_tuple_name: Optional[str],
docstring: Optional[Docstring],
):
Parameter |
Type |
inputs |
Union[Optional[Dict[str, Type]], Optional[Dict[str, Tuple[Type, Any]]]] |
outputs |
Union[Optional[Dict[str, Type]], Optional[Dict[str, Optional[Type]]]] |
output_tuple_name |
Optional[str] |
docstring |
Optional[Docstring] |
Methods
Method |
Description |
remove_inputs() |
This method is useful in removing some variables from the Flyte backend inputs specification, as these are |
with_inputs() |
Use this to add additional inputs to the interface |
with_outputs() |
This method allows addition of extra outputs are expected from a task specification |
def remove_inputs(
vars: Optional[List[str]],
):
This method is useful in removing some variables from the Flyte backend inputs specification, as these are
implicit local only inputs or will be supplied by the library at runtime. For example, spark-session etc
It creates a new instance of interface with the requested variables removed
Parameter |
Type |
vars |
Optional[List[str]] |
def with_inputs(
extra_inputs: Dict[str, Type],
):
Use this to add additional inputs to the interface. This is useful for adding additional implicit inputs that
are added without the user requesting for them
Parameter |
Type |
extra_inputs |
Dict[str, Type] |
with_outputs()
def with_outputs(
extra_outputs: Dict[str, Type],
):
This method allows addition of extra outputs are expected from a task specification
Parameter |
Type |
extra_outputs |
Dict[str, Type] |
Properties
Property |
Type |
Description |
default_inputs_as_kwargs |
|
|
docstring |
|
|
inputs |
|
|
inputs_with_defaults |
|
|
output_names |
|
|
output_tuple |
|
|
output_tuple_name |
|
|
outputs |
|
|
flytekit.core.base_task.LocalConfig
Any configuration specific to local runs.
def LocalConfig(
cache_enabled: bool,
cache_overwrite: bool,
):
Parameter |
Type |
cache_enabled |
bool |
cache_overwrite |
bool |
Methods
Method |
Description |
auto() |
None |
auto()
def auto(
config_file: typing.Union[str, ConfigFile],
):
Parameter |
Type |
config_file |
typing.Union[str, ConfigFile] |
flytekit.core.base_task.LocalTaskCache
This class implements a persistent store able to cache the result of local task executions.
Methods
clear()
get()
def get(
task_name: str,
cache_version: str,
input_literal_map: flytekit.models.literals.LiteralMap,
cache_ignore_input_vars: typing.Tuple[str, ...],
):
Parameter |
Type |
task_name |
str |
cache_version |
str |
input_literal_map |
flytekit.models.literals.LiteralMap |
cache_ignore_input_vars |
typing.Tuple[str, ...] |
initialize()
set()
def set(
task_name: str,
cache_version: str,
input_literal_map: flytekit.models.literals.LiteralMap,
cache_ignore_input_vars: typing.Tuple[str, ...],
value: flytekit.models.literals.LiteralMap,
):
Parameter |
Type |
task_name |
str |
cache_version |
str |
input_literal_map |
flytekit.models.literals.LiteralMap |
cache_ignore_input_vars |
typing.Tuple[str, ...] |
value |
flytekit.models.literals.LiteralMap |
flytekit.core.base_task.Promise
This object is a wrapper and exists for three main reasons. Let’s assume we’re dealing with a task like ::
@task
def t1() -> (int, str): …
#. Handling the duality between compilation and local execution - when the task function is run in a local execution
mode inside a workflow function, a Python integer and string are produced. When the task is being compiled as
part of the workflow, the task call creates a Node instead, and the task returns two Promise objects that
point to that Node.
#. One needs to be able to call ::
x = t1().with_overrides(…)
If the task returns an integer or a (int, str)
tuple like t1
above, calling with_overrides
on the
result would throw an error. This Promise object adds that.
#. Assorted handling for conditionals.
def Promise(
var: str,
val: Union[NodeOutput, _literals_models.Literal],
type: typing.Optional[_type_models.LiteralType],
):
Parameter |
Type |
var |
str |
val |
Union[NodeOutput, _literals_models.Literal] |
type |
typing.Optional[_type_models.LiteralType] |
Methods
deepcopy()
eval()
is_()
is_false()
is_none()
is_true()
with_overrides()
def with_overrides(
node_name: Optional[str],
aliases: Optional[Dict[str, str]],
requests: Optional[Resources],
limits: Optional[Resources],
timeout: Optional[Union[int, datetime.timedelta, object]],
retries: Optional[int],
interruptible: Optional[bool],
name: Optional[str],
task_config: Optional[Any],
container_image: Optional[str],
accelerator: Optional[BaseAccelerator],
cache: Optional[bool],
cache_version: Optional[str],
cache_serialize: Optional[bool],
args,
kwargs,
):
Parameter |
Type |
node_name |
Optional[str] |
aliases |
Optional[Dict[str, str]] |
requests |
Optional[Resources] |
limits |
Optional[Resources] |
timeout |
Optional[Union[int, datetime.timedelta, object]] |
retries |
Optional[int] |
interruptible |
Optional[bool] |
name |
Optional[str] |
task_config |
Optional[Any] |
container_image |
Optional[str] |
accelerator |
Optional[BaseAccelerator] |
cache |
Optional[bool] |
cache_version |
Optional[str] |
cache_serialize |
Optional[bool] |
args |
*args |
kwargs |
**kwargs |
with_var()
def with_var(
new_var: str,
):
Parameter |
Type |
new_var |
str |
Properties
Property |
Type |
Description |
attr_path |
|
|
is_ready |
|
|
ref |
|
|
val |
|
|
var |
|
|
flytekit.core.base_task.PythonTask
Base Class for all Tasks with a Python native Interface
. This should be directly used for task types, that do
not have a python function to be executed. Otherwise refer to :py:class:flytekit.PythonFunctionTask
.
def PythonTask(
task_type: str,
name: str,
task_config: typing.Optional[~T],
interface: typing.Optional[flytekit.core.interface.Interface],
environment: typing.Optional[typing.Dict[str, str]],
disable_deck: typing.Optional[bool],
enable_deck: typing.Optional[bool],
deck_fields: typing.Optional[typing.Tuple[flytekit.deck.deck.DeckField, ...]],
kwargs,
):
Parameter |
Type |
task_type |
str |
name |
str |
task_config |
typing.Optional[~T] |
interface |
typing.Optional[flytekit.core.interface.Interface] |
environment |
typing.Optional[typing.Dict[str, str]] |
disable_deck |
typing.Optional[bool] |
enable_deck |
typing.Optional[bool] |
deck_fields |
typing.Optional[typing.Tuple[flytekit.deck.deck.DeckField, ...]] |
kwargs |
**kwargs |
Methods
Method |
Description |
compile() |
Generates a node that encapsulates this task in a workflow definition |
construct_node_metadata() |
Used when constructing the node that encapsulates this task as part of a broader workflow definition |
dispatch_execute() |
This method translates Flyte’s Type system based input values and invokes the actual call to the executor |
execute() |
This method will be invoked to execute the task |
find_lhs() |
None |
get_config() |
Returns the task config as a serializable dictionary |
get_container() |
Returns the container definition (if any) that is used to run the task on hosted Flyte |
get_custom() |
Return additional plugin-specific custom data (if any) as a serializable dictionary |
get_extended_resources() |
Returns the extended resources to allocate to the task on hosted Flyte |
get_input_types() |
Returns the names and python types as a dictionary for the inputs of this task |
get_k8s_pod() |
Returns the kubernetes pod definition (if any) that is used to run the task on hosted Flyte |
get_sql() |
Returns the Sql definition (if any) that is used to run the task on hosted Flyte |
get_type_for_input_var() |
Returns the python type for an input variable by name |
get_type_for_output_var() |
Returns the python type for the specified output variable by name |
local_execute() |
This function is used only in the local execution path and is responsible for calling dispatch execute |
local_execution_mode() |
None |
post_execute() |
Post execute is called after the execution has completed, with the user_params and can be used to clean-up, |
pre_execute() |
This is the method that will be invoked directly before executing the task method and before all the inputs |
sandbox_execute() |
Call dispatch_execute, in the context of a local sandbox execution |
compile()
def compile(
ctx: flytekit.core.context_manager.FlyteContext,
args,
kwargs,
):
Generates a node that encapsulates this task in a workflow definition.
Parameter |
Type |
ctx |
flytekit.core.context_manager.FlyteContext |
args |
*args |
kwargs |
**kwargs |
def construct_node_metadata()
Used when constructing the node that encapsulates this task as part of a broader workflow definition.
dispatch_execute()
def dispatch_execute(
ctx: flytekit.core.context_manager.FlyteContext,
input_literal_map: flytekit.models.literals.LiteralMap,
):
This method translates Flyte’s Type system based input values and invokes the actual call to the executor
This method is also invoked during runtime.
VoidPromise
is returned in the case when the task itself declares no outputs.
Literal Map
is returned when the task returns either one more outputs in the declaration. Individual outputs
may be none
DynamicJobSpec
is returned when a dynamic workflow is executed
Parameter |
Type |
ctx |
flytekit.core.context_manager.FlyteContext |
input_literal_map |
flytekit.models.literals.LiteralMap |
execute()
This method will be invoked to execute the task.
Parameter |
Type |
kwargs |
**kwargs |
find_lhs()
get_config()
def get_config(
settings: flytekit.configuration.SerializationSettings,
):
Returns the task config as a serializable dictionary. This task config consists of metadata about the custom
defined for this task.
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
get_container()
def get_container(
settings: flytekit.configuration.SerializationSettings,
):
Returns the container definition (if any) that is used to run the task on hosted Flyte.
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
get_custom()
def get_custom(
settings: flytekit.configuration.SerializationSettings,
):
Return additional plugin-specific custom data (if any) as a serializable dictionary.
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
get_extended_resources()
def get_extended_resources(
settings: flytekit.configuration.SerializationSettings,
):
Returns the extended resources to allocate to the task on hosted Flyte.
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
Returns the names and python types as a dictionary for the inputs of this task.
get_k8s_pod()
def get_k8s_pod(
settings: flytekit.configuration.SerializationSettings,
):
Returns the kubernetes pod definition (if any) that is used to run the task on hosted Flyte.
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
get_sql()
def get_sql(
settings: flytekit.configuration.SerializationSettings,
):
Returns the Sql definition (if any) that is used to run the task on hosted Flyte.
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
def get_type_for_input_var(
k: str,
v: typing.Any,
):
Returns the python type for an input variable by name.
Parameter |
Type |
k |
str |
v |
typing.Any |
get_type_for_output_var()
def get_type_for_output_var(
k: str,
v: typing.Any,
):
Returns the python type for the specified output variable by name.
Parameter |
Type |
k |
str |
v |
typing.Any |
local_execute()
def local_execute(
ctx: flytekit.core.context_manager.FlyteContext,
kwargs,
):
This function is used only in the local execution path and is responsible for calling dispatch execute.
Use this function when calling a task with native values (or Promises containing Flyte literals derived from
Python native values).
Parameter |
Type |
ctx |
flytekit.core.context_manager.FlyteContext |
kwargs |
**kwargs |
local_execution_mode()
def local_execution_mode()
post_execute()
def post_execute(
user_params: typing.Optional[flytekit.core.context_manager.ExecutionParameters],
rval: typing.Any,
):
Post execute is called after the execution has completed, with the user_params and can be used to clean-up,
or alter the outputs to match the intended tasks outputs. If not overridden, then this function is a No-op
Parameter |
Type |
user_params |
typing.Optional[flytekit.core.context_manager.ExecutionParameters] |
rval |
typing.Any |
pre_execute()
def pre_execute(
user_params: typing.Optional[flytekit.core.context_manager.ExecutionParameters],
):
This is the method that will be invoked directly before executing the task method and before all the inputs
are converted. One particular case where this is useful is if the context is to be modified for the user process
to get some user space parameters. This also ensures that things like SparkSession are already correctly
setup before the type transformers are called
This should return either the same context of the mutated context
Parameter |
Type |
user_params |
typing.Optional[flytekit.core.context_manager.ExecutionParameters] |
sandbox_execute()
def sandbox_execute(
ctx: flytekit.core.context_manager.FlyteContext,
input_literal_map: flytekit.models.literals.LiteralMap,
):
Call dispatch_execute, in the context of a local sandbox execution. Not invoked during runtime.
Parameter |
Type |
ctx |
flytekit.core.context_manager.FlyteContext |
input_literal_map |
flytekit.models.literals.LiteralMap |
Properties
Property |
Type |
Description |
deck_fields |
|
|
disable_deck |
|
|
docs |
|
|
enable_deck |
|
|
environment |
|
|
instantiated_in |
|
|
interface |
|
|
lhs |
|
|
location |
|
|
metadata |
|
|
name |
|
|
python_interface |
|
|
security_context |
|
|
task_config |
|
|
task_type |
|
|
task_type_version |
|
|
flytekit.core.base_task.SecurityContext
This is a higher level wrapper object that for the most part users shouldn’t have to worry about. You should
be able to just use :py:class:flytekit.Secret
instead.
def SecurityContext(
run_as: typing.Optional[flytekit.models.security.Identity],
secrets: typing.Optional[typing.List[flytekit.models.security.Secret]],
tokens: typing.Optional[typing.List[flytekit.models.security.OAuth2TokenRequest]],
):
Parameter |
Type |
run_as |
typing.Optional[flytekit.models.security.Identity] |
secrets |
typing.Optional[typing.List[flytekit.models.security.Secret]] |
tokens |
typing.Optional[typing.List[flytekit.models.security.OAuth2TokenRequest]] |
Methods
from_flyte_idl()
def from_flyte_idl(
pb2_object: flyteidl.core.security_pb2.SecurityContext,
):
Parameter |
Type |
pb2_object |
flyteidl.core.security_pb2.SecurityContext |
serialize_to_string()
def serialize_to_string()
short_string()
to_flyte_idl()
verbose_string()
Properties
Property |
Type |
Description |
is_empty |
|
|
flytekit.core.base_task.SerializationSettings
These settings are provided while serializing a workflow and task, before registration. This is required to get
runtime information at serialization time, as well as some defaults.
Attributes:
project (str): The project (if any) with which to register entities under.
domain (str): The domain (if any) with which to register entities under.
version (str): The version (if any) with which to register entities under.
image_config (ImageConfig): The image config used to define task container images.
env (Optional[Dict[str, str]]): Environment variables injected into task container definitions.
flytekit_virtualenv_root (Optional[str]): During out of container serialize the absolute path of the flytekit
virtualenv at serialization time won’t match the in-container value at execution time. This optional value
is used to provide the in-container virtualenv path
python_interpreter (Optional[str]): The python executable to use. This is used for spark tasks in out of
container execution.
entrypoint_settings (Optional[EntrypointSettings]): Information about the command, path and version of the
entrypoint program.
fast_serialization_settings (Optional[FastSerializationSettings]): If the code is being serialized so that it
can be fast registered (and thus omit building a Docker image) this object contains additional parameters
for serialization.
source_root (Optional[str]): The root directory of the source code.
def SerializationSettings(
image_config: ImageConfig,
project: typing.Optional[str],
domain: typing.Optional[str],
version: typing.Optional[str],
env: Optional[Dict[str, str]],
git_repo: Optional[str],
python_interpreter: str,
flytekit_virtualenv_root: Optional[str],
fast_serialization_settings: Optional[FastSerializationSettings],
source_root: Optional[str],
):
Parameter |
Type |
image_config |
ImageConfig |
project |
typing.Optional[str] |
domain |
typing.Optional[str] |
version |
typing.Optional[str] |
env |
Optional[Dict[str, str]] |
git_repo |
Optional[str] |
python_interpreter |
str |
flytekit_virtualenv_root |
Optional[str] |
fast_serialization_settings |
Optional[FastSerializationSettings] |
source_root |
Optional[str] |
Methods
default_entrypoint_settings()
def default_entrypoint_settings(
interpreter_path: str,
):
Assumes the entrypoint is installed in a virtual-environment where the interpreter is
Parameter |
Type |
interpreter_path |
str |
for_image()
def for_image(
image: str,
version: str,
project: str,
domain: str,
python_interpreter_path: str,
):
Parameter |
Type |
image |
str |
version |
str |
project |
str |
domain |
str |
python_interpreter_path |
str |
from_dict()
def from_dict(
kvs: typing.Union[dict, list, str, int, float, bool, NoneType],
infer_missing,
):
Parameter |
Type |
kvs |
typing.Union[dict, list, str, int, float, bool, NoneType] |
infer_missing |
|
from_json()
def from_json(
s: typing.Union[str, bytes, bytearray],
parse_float,
parse_int,
parse_constant,
infer_missing,
kw,
):
Parameter |
Type |
s |
typing.Union[str, bytes, bytearray] |
parse_float |
|
parse_int |
|
parse_constant |
|
infer_missing |
|
kw |
|
from_transport()
def from_transport(
s: str,
):
new_builder()
Creates a SerializationSettings.Builder
that copies the existing serialization settings parameters and
allows for customization.
schema()
def schema(
infer_missing: bool,
only,
exclude,
many: bool,
context,
load_only,
dump_only,
partial: bool,
unknown,
):
Parameter |
Type |
infer_missing |
bool |
only |
|
exclude |
|
many |
bool |
context |
|
load_only |
|
dump_only |
|
partial |
bool |
unknown |
|
should_fast_serialize()
def should_fast_serialize()
Whether or not the serialization settings specify that entities should be serialized for fast registration.
to_dict()
def to_dict(
encode_json,
):
Parameter |
Type |
encode_json |
|
to_json()
def to_json(
skipkeys: bool,
ensure_ascii: bool,
check_circular: bool,
allow_nan: bool,
indent: typing.Union[int, str, NoneType],
separators: typing.Tuple[str, str],
default: typing.Callable,
sort_keys: bool,
kw,
):
Parameter |
Type |
skipkeys |
bool |
ensure_ascii |
bool |
check_circular |
bool |
allow_nan |
bool |
indent |
typing.Union[int, str, NoneType] |
separators |
typing.Tuple[str, str] |
default |
typing.Callable |
sort_keys |
bool |
kw |
|
venv_root_from_interpreter()
def venv_root_from_interpreter(
interpreter_path: str,
):
Computes the path of the virtual environment root, based on the passed in python interpreter path
for example /opt/venv/bin/python3 -> /opt/venv
Parameter |
Type |
interpreter_path |
str |
with_serialized_context()
def with_serialized_context()
Use this method to create a new SerializationSettings that has an environment variable set with the SerializedContext
This is useful in transporting SerializedContext to serialized and registered tasks.
The setting will be available in the env
field with the key SERIALIZED_CONTEXT_ENV_VAR
:return: A newly constructed SerializationSettings, or self, if it already has the serializationSettings
Properties
Property |
Type |
Description |
entrypoint_settings |
|
|
serialized_context |
|
|
flytekit.core.base_task.Task
The base of all Tasks in flytekit. This task is closest to the FlyteIDL TaskTemplate and captures information in
FlyteIDL specification and does not have python native interfaces associated. Refer to the derived classes for
examples of how to extend this class.
def Task(
task_type: str,
name: str,
interface: flytekit.models.interface.TypedInterface,
metadata: typing.Optional[flytekit.core.base_task.TaskMetadata],
task_type_version,
security_ctx: typing.Optional[flytekit.models.security.SecurityContext],
docs: typing.Optional[flytekit.models.documentation.Documentation],
kwargs,
):
Parameter |
Type |
task_type |
str |
name |
str |
interface |
flytekit.models.interface.TypedInterface |
metadata |
typing.Optional[flytekit.core.base_task.TaskMetadata] |
task_type_version |
|
security_ctx |
typing.Optional[flytekit.models.security.SecurityContext] |
docs |
typing.Optional[flytekit.models.documentation.Documentation] |
kwargs |
**kwargs |
Methods
Method |
Description |
compile() |
None |
dispatch_execute() |
This method translates Flyte’s Type system based input values and invokes the actual call to the executor |
execute() |
This method will be invoked to execute the task |
get_config() |
Returns the task config as a serializable dictionary |
get_container() |
Returns the container definition (if any) that is used to run the task on hosted Flyte |
get_custom() |
Return additional plugin-specific custom data (if any) as a serializable dictionary |
get_extended_resources() |
Returns the extended resources to allocate to the task on hosted Flyte |
get_input_types() |
Returns python native types for inputs |
get_k8s_pod() |
Returns the kubernetes pod definition (if any) that is used to run the task on hosted Flyte |
get_sql() |
Returns the Sql definition (if any) that is used to run the task on hosted Flyte |
get_type_for_input_var() |
Returns the python native type for the given input variable |
get_type_for_output_var() |
Returns the python native type for the given output variable |
local_execute() |
This function is used only in the local execution path and is responsible for calling dispatch execute |
local_execution_mode() |
None |
pre_execute() |
This is the method that will be invoked directly before executing the task method and before all the inputs |
sandbox_execute() |
Call dispatch_execute, in the context of a local sandbox execution |
compile()
def compile(
ctx: flytekit.core.context_manager.FlyteContext,
args,
kwargs,
):
Parameter |
Type |
ctx |
flytekit.core.context_manager.FlyteContext |
args |
*args |
kwargs |
**kwargs |
dispatch_execute()
def dispatch_execute(
ctx: flytekit.core.context_manager.FlyteContext,
input_literal_map: flytekit.models.literals.LiteralMap,
):
This method translates Flyte’s Type system based input values and invokes the actual call to the executor
This method is also invoked during runtime.
Parameter |
Type |
ctx |
flytekit.core.context_manager.FlyteContext |
input_literal_map |
flytekit.models.literals.LiteralMap |
execute()
This method will be invoked to execute the task.
Parameter |
Type |
kwargs |
**kwargs |
get_config()
def get_config(
settings: flytekit.configuration.SerializationSettings,
):
Returns the task config as a serializable dictionary. This task config consists of metadata about the custom
defined for this task.
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
get_container()
def get_container(
settings: flytekit.configuration.SerializationSettings,
):
Returns the container definition (if any) that is used to run the task on hosted Flyte.
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
get_custom()
def get_custom(
settings: flytekit.configuration.SerializationSettings,
):
Return additional plugin-specific custom data (if any) as a serializable dictionary.
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
get_extended_resources()
def get_extended_resources(
settings: flytekit.configuration.SerializationSettings,
):
Returns the extended resources to allocate to the task on hosted Flyte.
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
Returns python native types for inputs. In case this is not a python native task (base class) and hence
returns a None. we could deduce the type from literal types, but that is not a required exercise
TODO we could use literal type to determine this
get_k8s_pod()
def get_k8s_pod(
settings: flytekit.configuration.SerializationSettings,
):
Returns the kubernetes pod definition (if any) that is used to run the task on hosted Flyte.
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
get_sql()
def get_sql(
settings: flytekit.configuration.SerializationSettings,
):
Returns the Sql definition (if any) that is used to run the task on hosted Flyte.
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
def get_type_for_input_var(
k: str,
v: typing.Any,
):
Returns the python native type for the given input variable
TODO we could use literal type to determine this
Parameter |
Type |
k |
str |
v |
typing.Any |
get_type_for_output_var()
def get_type_for_output_var(
k: str,
v: typing.Any,
):
Returns the python native type for the given output variable
TODO we could use literal type to determine this
Parameter |
Type |
k |
str |
v |
typing.Any |
local_execute()
def local_execute(
ctx: flytekit.core.context_manager.FlyteContext,
kwargs,
):
This function is used only in the local execution path and is responsible for calling dispatch execute.
Use this function when calling a task with native values (or Promises containing Flyte literals derived from
Python native values).
Parameter |
Type |
ctx |
flytekit.core.context_manager.FlyteContext |
kwargs |
**kwargs |
local_execution_mode()
def local_execution_mode()
pre_execute()
def pre_execute(
user_params: flytekit.core.context_manager.ExecutionParameters,
):
This is the method that will be invoked directly before executing the task method and before all the inputs
are converted. One particular case where this is useful is if the context is to be modified for the user process
to get some user space parameters. This also ensures that things like SparkSession are already correctly
setup before the type transformers are called
This should return either the same context of the mutated context
Parameter |
Type |
user_params |
flytekit.core.context_manager.ExecutionParameters |
sandbox_execute()
def sandbox_execute(
ctx: flytekit.core.context_manager.FlyteContext,
input_literal_map: flytekit.models.literals.LiteralMap,
):
Call dispatch_execute, in the context of a local sandbox execution. Not invoked during runtime.
Parameter |
Type |
ctx |
flytekit.core.context_manager.FlyteContext |
input_literal_map |
flytekit.models.literals.LiteralMap |
Properties
Property |
Type |
Description |
docs |
|
|
interface |
|
|
metadata |
|
|
name |
|
|
python_interface |
|
|
security_context |
|
|
task_type |
|
|
task_type_version |
|
|
Metadata for a Task. Things like retries and whether or not caching is turned on, and cache version are specified
here.
See the :std:ref:IDL <idl:protos/docs/core/core:taskmetadata>
for the protobuf definition.
Attributes:
cache (bool): Indicates if caching should be enabled. See :std:ref:Caching <cookbook:caching>
.
cache_serialize (bool): Indicates if identical (i.e. same inputs) instances of this task should be executed in serial when caching is enabled. See :std:ref:Caching <cookbook:caching>
.
cache_version (str): Version to be used for the cached value.
cache_ignore_input_vars (Tuple[str, …]): Input variables that should not be included when calculating hash for cache.
interruptible (Optional[bool]): Indicates that this task can be interrupted and/or scheduled on nodes with lower QoS guarantees that can include pre-emption.
deprecated (str): Can be used to provide a warning message for a deprecated task. An absence or empty string indicates that the task is active and not deprecated.
retries (int): for retries=n; n > 0, on failures of this task, the task will be retried at-least n number of times.
timeout (Optional[Union[datetime.timedelta, int]]): The maximum duration for which one execution of this task should run. The execution will be terminated if the runtime exceeds this timeout.
pod_template_name (Optional[str]): The name of an existing PodTemplate resource in the cluster which will be used for this task.
generates_deck (bool): Indicates whether the task will generate a Deck URI.
is_eager (bool): Indicates whether the task should be treated as eager.
def TaskMetadata(
cache: bool,
cache_serialize: bool,
cache_version: str,
cache_ignore_input_vars: typing.Tuple[str, ...],
interruptible: typing.Optional[bool],
deprecated: str,
retries: int,
timeout: typing.Union[datetime.timedelta, int, NoneType],
pod_template_name: typing.Optional[str],
generates_deck: bool,
is_eager: bool,
):
Parameter |
Type |
cache |
bool |
cache_serialize |
bool |
cache_version |
str |
cache_ignore_input_vars |
typing.Tuple[str, ...] |
interruptible |
typing.Optional[bool] |
deprecated |
str |
retries |
int |
timeout |
typing.Union[datetime.timedelta, int, NoneType] |
pod_template_name |
typing.Optional[str] |
generates_deck |
bool |
is_eager |
bool |
Methods
def to_taskmetadata_model()
Converts to _task_model.TaskMetadata
Properties
Property |
Type |
Description |
retry_strategy |
|
|
flytekit.core.base_task.TaskResolverMixin
Flytekit tasks interact with the Flyte platform very, very broadly in two steps. They need to be uploaded to Admin,
and then they are run by the user upon request (either as a single task execution or as part of a workflow). In any
case, at execution time, for most tasks (that is those that generate a container target) the container image
containing the task needs to be spun up again at which point the container needs to know which task it’s supposed
to run and how to rehydrate the task object.
For example, the serialization of a simple task ::
in repo_root/workflows/example.py
@task
def t1(…) -> …: …
might result in a container with arguments like ::
pyflyte-execute –inputs s3://path/inputs.pb –output-prefix s3://outputs/location –raw-output-data-prefix /tmp/data –resolver flytekit.core.python_auto_container.default_task_resolver – task-module repo_root.workflows.example task-name t1
At serialization time, the container created for the task will start out automatically with the pyflyte-execute
bit, along with the requisite input/output args and the offloaded data prefix. Appended to that will be two things,
#. the location
of the task’s task resolver, followed by two dashes, followed by
#. the arguments provided by calling the loader_args
function below.
The default_task_resolver
declared below knows that
- When
loader_args
is called on a task, to look up the module the task is in, and the name of the task (the
key of the task in the module, either the function name, or the variable it was assigned to).
- When
load_task
is called, it interprets the first part of the command as the module to call
importlib.import_module
on, and then looks for a key t1
.
This is just the default behavior. Users should feel free to implement their own resolvers.
Methods
Method |
Description |
get_all_tasks() |
Future proof method |
load_task() |
Given the set of identifier keys, should return one Python Task or raise an error if not found |
loader_args() |
Return a list of strings that can help identify the parameter Task |
name() |
None |
task_name() |
Overridable function that can optionally return a custom name for a given task |
get_all_tasks()
Future proof method. Just making it easy to access all tasks (Not required today as we auto register them)
load_task()
def load_task(
loader_args: typing.List[str],
):
Given the set of identifier keys, should return one Python Task or raise an error if not found
Parameter |
Type |
loader_args |
typing.List[str] |
loader_args()
def loader_args(
settings: flytekit.configuration.SerializationSettings,
t: flytekit.core.base_task.Task,
):
Return a list of strings that can help identify the parameter Task
Parameter |
Type |
settings |
flytekit.configuration.SerializationSettings |
t |
flytekit.core.base_task.Task |
name()
task_name()
def task_name(
t: flytekit.core.base_task.Task,
):
Overridable function that can optionally return a custom name for a given task
Parameter |
Type |
t |
flytekit.core.base_task.Task |
Properties
Property |
Type |
Description |
location |
|
|
flytekit.core.base_task.TrackedInstance
Please see the notes for the metaclass above first.
This functionality has two use-cases currently,
- Keep track of naming for non-function
PythonAutoContainerTasks
. That is, things like the
:py:class:flytekit.extras.sqlite3.task.SQLite3Task
task.
- Task resolvers, because task resolvers are instances of :py:class:
flytekit.core.python_auto_container.TaskResolverMixin
classes, not the classes themselves, which means we need to look on the left hand side of them to see how to
find them at task execution time.
def TrackedInstance(
args,
kwargs,
):
Parameter |
Type |
args |
*args |
kwargs |
**kwargs |
Methods
find_lhs()
Properties
Property |
Type |
Description |
instantiated_in |
|
|
lhs |
|
|
location |
|
|
flytekit.core.base_task.TypeEngine
Core Extensible TypeEngine of Flytekit. This should be used to extend the capabilities of FlyteKits type system.
Users can implement their own TypeTransformers and register them with the TypeEngine. This will allow special handling
of user objects
Methods
async_to_literal()
def async_to_literal(
ctx: FlyteContext,
python_val: typing.Any,
python_type: Type[T],
expected: LiteralType,
):
Converts a python value of a given type and expected LiteralType
into a resolved Literal
value.
Parameter |
Type |
ctx |
FlyteContext |
python_val |
typing.Any |
python_type |
Type[T] |
expected |
LiteralType |
async_to_python_value()
def async_to_python_value(
ctx: FlyteContext,
lv: Literal,
expected_python_type: Type,
):
Parameter |
Type |
ctx |
FlyteContext |
lv |
Literal |
expected_python_type |
Type |
calculate_hash()
def calculate_hash(
python_val: typing.Any,
python_type: Type[T],
):
Parameter |
Type |
python_val |
typing.Any |
python_type |
Type[T] |
dict_to_literal_map()
def dict_to_literal_map(
ctx: FlyteContext,
d: typing.Dict[str, typing.Any],
type_hints: Optional[typing.Dict[str, type]],
):
Parameter |
Type |
ctx |
FlyteContext |
d |
typing.Dict[str, typing.Any] |
type_hints |
Optional[typing.Dict[str, type]] |
dict_to_literal_map_pb()
def dict_to_literal_map_pb(
ctx: FlyteContext,
d: typing.Dict[str, typing.Any],
type_hints: Optional[typing.Dict[str, type]],
):
Parameter |
Type |
ctx |
FlyteContext |
d |
typing.Dict[str, typing.Any] |
type_hints |
Optional[typing.Dict[str, type]] |
def get_available_transformers()
Returns all python types for which transformers are available
def get_transformer(
python_type: Type,
):
Implements a recursive search for the transformer.
Parameter |
Type |
python_type |
Type |
guess_python_type()
def guess_python_type(
flyte_type: LiteralType,
):
Transforms a flyte-specific LiteralType
to a regular python value.
Parameter |
Type |
flyte_type |
LiteralType |
guess_python_types()
def guess_python_types(
flyte_variable_dict: typing.Dict[str, _interface_models.Variable],
):
Transforms a dictionary of flyte-specific Variable
objects to a dictionary of regular python values.
Parameter |
Type |
flyte_variable_dict |
typing.Dict[str, _interface_models.Variable] |
def lazy_import_transformers()
Only load the transformers if needed.
literal_map_to_kwargs()
def literal_map_to_kwargs(
ctx: FlyteContext,
lm: LiteralMap,
python_types: typing.Optional[typing.Dict[str, type]],
literal_types: typing.Optional[typing.Dict[str, _interface_models.Variable]],
):
Parameter |
Type |
ctx |
FlyteContext |
lm |
LiteralMap |
python_types |
typing.Optional[typing.Dict[str, type]] |
literal_types |
typing.Optional[typing.Dict[str, _interface_models.Variable]] |
named_tuple_to_variable_map()
def named_tuple_to_variable_map(
t: typing.NamedTuple,
):
Converts a python-native NamedTuple
to a flyte-specific VariableMap of named literals.
Parameter |
Type |
t |
typing.NamedTuple |
register()
def register(
transformer: TypeTransformer,
additional_types: Optional[typing.List[Type]],
):
This should be used for all types that respond with the right type annotation when you use type(…) function
Parameter |
Type |
transformer |
TypeTransformer |
additional_types |
Optional[typing.List[Type]] |
register_additional_type()
def register_additional_type(
transformer: TypeTransformer[T],
additional_type: Type[T],
override,
):
Parameter |
Type |
transformer |
TypeTransformer[T] |
additional_type |
Type[T] |
override |
|
register_restricted_type()
def register_restricted_type(
name: str,
type: Type[T],
):
Parameter |
Type |
name |
str |
type |
Type[T] |
to_html()
def to_html(
ctx: FlyteContext,
python_val: typing.Any,
expected_python_type: Type[typing.Any],
):
Parameter |
Type |
ctx |
FlyteContext |
python_val |
typing.Any |
expected_python_type |
Type[typing.Any] |
to_literal()
def to_literal(
ctx: FlyteContext,
python_val: typing.Any,
python_type: Type[T],
expected: LiteralType,
):
The current dance is because we are allowing users to call from an async function, this synchronous
to_literal function, and allowing this to_literal function, to then invoke yet another async function,
namely an async transformer.
Parameter |
Type |
ctx |
FlyteContext |
python_val |
typing.Any |
python_type |
Type[T] |
expected |
LiteralType |
to_literal_checks()
def to_literal_checks(
python_val: typing.Any,
python_type: Type[T],
expected: LiteralType,
):
Parameter |
Type |
python_val |
typing.Any |
python_type |
Type[T] |
expected |
LiteralType |
to_literal_type()
def to_literal_type(
python_type: Type[T],
):
Converts a python type into a flyte specific LiteralType
Parameter |
Type |
python_type |
Type[T] |
to_python_value()
def to_python_value(
ctx: FlyteContext,
lv: Literal,
expected_python_type: Type,
):
Converts a Literal value with an expected python type into a python value.
Parameter |
Type |
ctx |
FlyteContext |
lv |
Literal |
expected_python_type |
Type |
unwrap_offloaded_literal()
def unwrap_offloaded_literal(
ctx: FlyteContext,
lv: Literal,
):
Parameter |
Type |
ctx |
FlyteContext |
lv |
Literal |
Inappropriate argument type.
flytekit.core.base_task.TypeVar
Type variable.
The preferred way to construct a type variable is via the dedicated
syntax for generic functions, classes, and type aliases::
class Sequence[T]: # T is a TypeVar
…
This syntax can also be used to create bound and constrained type
variables::
S is a TypeVar bound to str
class StrSequence[S: str]:
…
A is a TypeVar constrained to str or bytes
class StrOrBytesSequence[A: (str, bytes)]:
…
However, if desired, reusable type variables can also be constructed
manually, like so::
T = TypeVar(‘T’) # Can be anything
S = TypeVar(‘S’, bound=str) # Can be any subtype of str
A = TypeVar(‘A’, str, bytes) # Must be exactly str or bytes
Type variables exist primarily for the benefit of static type
checkers. They serve as the parameters for generic types as well
as for generic function and type alias definitions.
The variance of type variables is inferred by type checkers when they
are created through the type parameter syntax and when
infer_variance=True
is passed. Manually created type variables may
be explicitly marked covariant or contravariant by passing
covariant=True
or contravariant=True
. By default, manually
created type variables are invariant. See PEP 484 and PEP 695 for more
details.
flytekit.core.base_task.Variable
def Variable(
type,
description,
artifact_partial_id: typing.Optional[flyteidl.core.artifact_id_pb2.ArtifactID],
artifact_tag: typing.Optional[flyteidl.core.artifact_id_pb2.ArtifactTag],
):
Parameter |
Type |
type |
|
description |
|
artifact_partial_id |
typing.Optional[flyteidl.core.artifact_id_pb2.ArtifactID] |
artifact_tag |
typing.Optional[flyteidl.core.artifact_id_pb2.ArtifactTag] |
Methods
from_flyte_idl()
def from_flyte_idl(
variable_proto,
):
Parameter |
Type |
variable_proto |
|
serialize_to_string()
def serialize_to_string()
short_string()
to_flyte_idl()
to_flyte_idl_list()
verbose_string()
Properties
Property |
Type |
Description |
artifact_partial_id |
|
|
artifact_tag |
|
|
description |
|
|
is_empty |
|
|
type |
|
|
flytekit.core.base_task.VoidPromise
This object is returned for tasks that do not return any outputs (declared interface is empty)
VoidPromise cannot be interacted with and does not allow comparisons or any operations
def VoidPromise(
task_name: str,
ref: Optional[NodeOutput],
):
Parameter |
Type |
task_name |
str |
ref |
Optional[NodeOutput] |
Methods
runs_before()
def runs_before(
args,
kwargs,
):
This is a placeholder and should do nothing. It is only here to enable local execution of workflows
where a task returns nothing.
Parameter |
Type |
args |
*args |
kwargs |
**kwargs |
with_overrides()
def with_overrides(
args,
kwargs,
):
Parameter |
Type |
args |
*args |
kwargs |
**kwargs |
Properties
Property |
Type |
Description |
ref |
|
|
task_name |
|
|
flytekit.core.base_task.timeit
A context manager and a decorator that measures the execution time of the wrapped code block or functions.
It will append a timing information to TimeLineDeck. For instance:
@timeit(“Function description”)
def function()
with timeit(“Wrapped code block description”):
your code
def timeit(
name: str,
):