Flytekit

Warnings

• breaking Flytekit v1.14.0 introduced a change to use MessagePack for serializing Python `dict`, `dataclass`, and `Pydantic BaseModel` types instead of JSON strings within a Protobuf struct. This change is not fully backward-compatible with Flyte backend versions older than 1.14. If referencing tasks registered with `flytekit>=1.14` from downstream tasks using `flytekit<1.14`, you may encounter `TypeTransformerFailedError`.
  Fix: Upgrade both your `flytekit` SDK and your Flyte backend deployment to version 1.14 or higher. For temporary backward compatibility, you can set the `FLYTE_USE_OLD_DC_FORMAT` environment variable to `true`.
• breaking When upgrading to `flytekit>=1.13.0`, users have reported `TaskReferenceNotFound` errors during workflow compilation and registration, especially when registering scripts to a Flyte cluster. Older versions (e.g., `flytekit==1.11.0`) might not exhibit this behavior.
  Fix: This issue is likely due to changes in how tasks and workflows are referenced and registered in newer versions. Consult the official Flytekit documentation and migration guides for your specific version to understand new registration patterns or configuration requirements. Ensure your Flyte backend is also compatible with your `flytekit` version.
• gotcha Flytekit strictly enforces type hints for all task inputs and outputs. If type annotations are incorrect or missing (and cannot be inferred), Flytekit will raise a compilation error during registration, preventing the workflow from running. This is a design choice to ensure robust and predictable workflows.
  Fix: Always provide explicit and correct type hints for all function arguments and return values in `@task` and `@workflow` decorated functions. Use Python's `typing` module for complex types. Refer to Flytekit's type system documentation for supported types and best practices.
• gotcha The body of a `@workflow` decorated function in Flytekit executes at *registration time* to construct the Directed Acyclic Graph (DAG) of tasks, not at runtime for actual computation. The actual computation logic is encapsulated within individual `@task` decorated functions.
  Fix: Understand that `workflow` definitions are primarily for orchestration. Avoid placing complex computational logic directly within a workflow body that is intended to run at execution time. Instead, encapsulate such logic within tasks.
• gotcha Flytekit's current stable versions (1.16.x) explicitly support Python versions 3.10, 3.11, and 3.12. Older Python versions, such as 3.8 or 3.9, may not be fully supported or tested, and could lead to unexpected behavior or installation issues.
  Fix: Ensure your development and execution environments use Python 3.10, 3.11, or 3.12. If you are on an older Python version, consider upgrading or using an older, compatible `flytekit` release at your own risk.
• gotcha In dynamic workflows, if one subtask fails, the default behavior is for all other running subtasks within that dynamic workflow to also fail. This is designed to conserve resources, assuming downstream tasks depend on all upstream outputs.
  Fix: If you require individual subtasks to complete even if others fail, you may need to explicitly configure the dynamic task with parameters that control this failure tolerance (e.g., a ratio of acceptable successful tasks, if supported by the specific dynamic task implementation).

Install

• pip install flytekit Install stable version

Imports

• task

  from flytekit import task

• workflow

  from flytekit import workflow

• current_context

  from flytekit import current_context

  Used to access execution context like current project, domain, or workflow ID.

Quickstart

This quickstart defines a simple Flyte task `say_hello` and a workflow `hello_world_wf` that orchestrates it. Tasks are Python functions decorated with `@task`, and workflows are Python functions decorated with `@workflow`. All inputs and outputs must be type-annotated. The workflow can be run locally like a regular Python function or via the `pyflyte` CLI.

from flytekit import task, workflow

@task
def say_hello(name: str) -> str:
    return f"Hello, {name}!"

@workflow
def hello_world_wf(name: str = "world") -> str:
    res = say_hello(name=name)
    return res

if __name__ == "__main__":
    # Run locally
    print(f"Local execution: {hello_world_wf(name='Flyte')}")
    # Alternatively, use pyflyte CLI for local or remote execution:
    # pyflyte run your_script.py hello_world_wf --name Flyte