flytekit.core.class_based_resolver
Directory
Classes
Class | Description |
---|---|
ClassStorageTaskResolver |
Stores tasks inside a class variable. |
PythonAutoContainerTask |
A Python AutoContainer task should be used as the base for all extensions that want the user’s code to be in the. |
SerializationSettings |
These settings are provided while serializing a workflow and task, before registration. |
TaskResolverMixin |
Flytekit tasks interact with the Flyte platform very, very broadly in two steps. |
TrackedInstance |
Please see the notes for the metaclass above first. |
flytekit.core.class_based_resolver.ClassStorageTaskResolver
Stores tasks inside a class variable. The class must be inherited from at the point of usage because the task loading process basically relies on the same sequence of things happening.
def ClassStorageTaskResolver(
args,
kwargs,
):
Parameter | Type |
---|---|
args |
*args |
kwargs |
**kwargs |
Methods
Method | Description |
---|---|
add() |
None |
find_lhs() |
None |
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() |
This is responsible for turning an instance of a task into args that the load_task function can reconstitute |
name() |
None |
task_name() |
Overridable function that can optionally return a custom name for a given task |
add()
def add(
t: flytekit.core.python_auto_container.PythonAutoContainerTask,
):
Parameter | Type |
---|---|
t |
flytekit.core.python_auto_container.PythonAutoContainerTask |
find_lhs()
def find_lhs()
get_all_tasks()
def 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.python_auto_container.PythonAutoContainerTask,
):
This is responsible for turning an instance of a task into args that the load_task function can reconstitute.
Parameter | Type |
---|---|
settings |
flytekit.configuration.SerializationSettings |
t |
flytekit.core.python_auto_container.PythonAutoContainerTask |
name()
def 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 |
---|---|---|
instantiated_in | ||
lhs | ||
location |
flytekit.core.class_based_resolver.PythonAutoContainerTask
A Python AutoContainer task should be used as the base for all extensions that want the user’s code to be in the container and the container information to be automatically captured. This base will auto configure the image and image version to be used for all its derivatives.
If you are looking to extend, you might prefer to use PythonFunctionTask
or PythonInstanceTask
def PythonAutoContainerTask(
name: str,
task_config: T,
task_type,
container_image: Optional[Union[str, ImageSpec]],
requests: Optional[Resources],
limits: Optional[Resources],
environment: Optional[Dict[str, str]],
task_resolver: Optional[TaskResolverMixin],
secret_requests: Optional[List[Secret]],
pod_template: Optional[PodTemplate],
pod_template_name: Optional[str],
accelerator: Optional[BaseAccelerator],
shared_memory: Optional[Union[L[True], str]],
resources: Optional[Resources],
kwargs,
):
Parameter | Type |
---|---|
name |
str |
task_config |
T |
task_type |
|
container_image |
Optional[Union[str, ImageSpec]] |
requests |
Optional[Resources] |
limits |
Optional[Resources] |
environment |
Optional[Dict[str, str]] |
task_resolver |
Optional[TaskResolverMixin] |
secret_requests |
Optional[List[Secret]] |
pod_template |
Optional[PodTemplate] |
pod_template_name |
Optional[str] |
accelerator |
Optional[BaseAccelerator] |
shared_memory |
Optional[Union[L[True], str]] |
resources |
Optional[Resources] |
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_command() |
Returns the command which should be used in the container definition for the serialized version of this 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_default_command() |
Returns the default pyflyte-execute command used to run this on hosted Flyte platforms |
get_extended_resources() |
Returns the extended resources to allocate to the task on hosted Flyte |
get_image() |
Update image spec based on fast registration usage, and return string representing the image |
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 |
reset_command_fn() |
Resets the command which should be used in the container definition of this task to the default arguments |
sandbox_execute() |
Call dispatch_execute, in the context of a local sandbox execution |
set_command_fn() |
By default, the task will run on the Flyte platform using the pyflyte-execute command |
set_resolver() |
By default, flytekit uses the DefaultTaskResolver to resolve the task |
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 |
construct_node_metadata()
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 noneDynamicJobSpec
is returned when a dynamic workflow is executed
Parameter | Type |
---|---|
ctx |
flytekit.core.context_manager.FlyteContext |
input_literal_map |
flytekit.models.literals.LiteralMap |
execute()
def execute(
kwargs,
):
This method will be invoked to execute the task.
Parameter | Type |
---|---|
kwargs |
**kwargs |
find_lhs()
def find_lhs()
get_command()
def get_command(
settings: SerializationSettings,
):
Returns the command which should be used in the container definition for the serialized version of this task registered on a hosted Flyte platform.
Parameter | Type |
---|---|
settings |
SerializationSettings |
get_config()
def get_config(
settings: 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 |
SerializationSettings |
get_container()
def get_container(
settings: SerializationSettings,
):
Returns the container definition (if any) that is used to run the task on hosted Flyte.
Parameter | Type |
---|---|
settings |
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_default_command()
def get_default_command(
settings: SerializationSettings,
):
Returns the default pyflyte-execute command used to run this on hosted Flyte platforms.
Parameter | Type |
---|---|
settings |
SerializationSettings |
get_extended_resources()
def get_extended_resources(
settings: SerializationSettings,
):
Returns the extended resources to allocate to the task on hosted Flyte.
Parameter | Type |
---|---|
settings |
SerializationSettings |
get_image()
def get_image(
settings: SerializationSettings,
):
Update image spec based on fast registration usage, and return string representing the image
Parameter | Type |
---|---|
settings |
SerializationSettings |
get_input_types()
def get_input_types()
Returns the names and python types as a dictionary for the inputs of this task.
get_k8s_pod()
def get_k8s_pod(
settings: SerializationSettings,
):
Returns the kubernetes pod definition (if any) that is used to run the task on hosted Flyte.
Parameter | Type |
---|---|
settings |
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 |
get_type_for_input_var()
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] |
reset_command_fn()
def reset_command_fn()
Resets the command which should be used in the container definition of this task to the default arguments. This is useful when the command line is overridden at serialization time.
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 |
set_command_fn()
def set_command_fn(
get_command_fn: Optional[Callable[[SerializationSettings], List[str]]],
):
By default, the task will run on the Flyte platform using the pyflyte-execute command. However, it can be useful to update the command with which the task is serialized for specific cases like running map tasks (“pyflyte-map-execute”) or for fast-executed tasks.
Parameter | Type |
---|---|
get_command_fn |
Optional[Callable[[SerializationSettings], List[str]]] |
set_resolver()
def set_resolver(
resolver: TaskResolverMixin,
):
By default, flytekit uses the DefaultTaskResolver to resolve the task. This method allows the user to set a custom task resolver. It can be useful to override the task resolver for specific cases like running tasks in the jupyter notebook.
Parameter | Type |
---|---|
resolver |
TaskResolverMixin |
Properties
Property | Type | Description |
---|---|---|
container_image | ||
deck_fields | ||
disable_deck | ||
docs | ||
enable_deck | ||
environment | ||
instantiated_in | ||
interface | ||
lhs | ||
location | ||
metadata | ||
name | ||
python_interface | ||
resources | ||
security_context | ||
task_config | ||
task_resolver | ||
task_type | ||
task_type_version |
flytekit.core.class_based_resolver.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
Method | Description |
---|---|
default_entrypoint_settings() |
Assumes the entrypoint is installed in a virtual-environment where the interpreter is |
for_image() |
None |
from_dict() |
None |
from_json() |
None |
from_transport() |
None |
new_builder() |
Creates a ``SerializationSettings |
schema() |
None |
should_fast_serialize() |
Whether or not the serialization settings specify that entities should be serialized for fast registration |
to_dict() |
None |
to_json() |
None |
venv_root_from_interpreter() |
Computes the path of the virtual environment root, based on the passed in python interpreter path |
with_serialized_context() |
Use this method to create a new SerializationSettings that has an environment variable set with the SerializedContext |
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,
):
Parameter | Type |
---|---|
s |
str |
new_builder()
def 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.class_based_resolver.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 callimportlib.import_module
on, and then looks for a keyt1
.
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()
def 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()
def 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.class_based_resolver.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
Method | Description |
---|---|
find_lhs() |
None |
find_lhs()
def find_lhs()
Properties
Property | Type | Description |
---|---|---|
instantiated_in | ||
lhs | ||
location |