# Development cycle
> This bundle contains all pages in the Development cycle section.
> Source: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle ===

# Development cycle

This section covers developing production-ready workflows for Flyte.

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/project-structure ===

# Project structure

Organizing a workflow project repository effectively is key for ensuring scalability, collaboration, and easy maintenance.
Here are best practices for structuring a Flyte workflow project repo, covering task organization, workflow management, dependency handling, and documentation.

## Recommended Directory Structure

A typical Flyte workflow project structure could look like this:

```shell
├── .github/workflows/
├── .gitignore
├── docs/
│   └── README.md
├── src/
│   ├── core/                    # Core logic specific to the use case
│   │   ├── __init__.py
│   │   ├── model.py
│   │   ├── data.py
│   │   └── structs.py
│   ├── tasks/                   # Contains individual tasks
│   │   ├── __init__.py
│   │   ├── preprocess.py
│   │   ├── fit.py
│   │   ├── test.py
│   │   └── plot.py
│   ├── workflows/               # Contains workflow definitions
│   │   ├── __init__.py
│   │   ├── inference.py
│   │   └── train.py
│   └── orchestration/           # For helper constructs (e.g., secrets, images)
│       ├── __init__.py
│       └── constants.py
├── uv.lock
└── pyproject.toml

```

This structure is designed to ensure each project component has a clear, logical home, making it easy for team members to find and modify files.

## Organizing Tasks and Workflows

In Flyte, tasks are the building blocks of workflows, so it’s important to structure them intuitively:

* **Tasks**: Store each task in its own file within the `tasks/` directory. If multiple tasks are closely related, consider grouping them within a module. Alternatively, each task can have its own module to allow more granular organization and sub-directories could be used to group similar tasks.

* **Workflows**: Store workflows, which combine tasks into end-to-end processes, in the `workflows/` directory. This separation ensures workflows are organized independently from core task logic, promoting modularity and reuse.

## Orchestration Directory for Helper Constructs

Include a directory, such as `orchestration/` or `union_utils/`, for constructs that facilitate workflow orchestration. This can house helper files like:

* **Secrets**: Definitions for accessing secrets (e.g., API keys) in Flyte.

* **ImageSpec**: A tool that simplifies container management, allowing you to avoid writing Dockerfiles directly.

## Core Logic for Workflow-Specific Functionality

Use a `core/` directory for business logic specific to your workflows. This keeps the core application code separate from workflow orchestration code, improving maintainability and making it easier for new team members to understand core functionality.

## Importance of `__init__.py`

Adding `__init__.py` files within each directory is essential:

* **For Imports**: These files make the directory a Python package, enabling proper imports across modules.

* **For Flyte's Fast Registration**: When performing fast registration, Flyte considers the first directory without an `__init__.py` as the root. Flyte will then package the root and its contents into a tarball, streamlining the registration process and avoiding the need to rebuild the container image every time you make code changes.

## Monorepo vs Multi-repo: Choosing a structure

When working with multiple teams, you have two main options:

* **Monorepo**: A single repository shared across all teams, which can simplify dependency management and allow for shared constructs. However, it can introduce complexity in permissions and version control for different teams.

* **Multi-repo**: Separate repositories for each team or project can improve isolation and control. In this case, consider creating shared, installable packages for constructs that multiple teams use, ensuring consistency without merging codebases.

## CI/CD

The GitHub action should:
* Register (and promote if needed) on merge to domain branch.
* Execute on merge of input YAML.
* Inject git SHA as entity version.

## Documentation and Docstrings

Writing clear docstrings is encouraged, as they are automatically propagated to the Flyte UI. This provides useful context for anyone viewing the workflows and tasks in the UI, reducing the need to consult source code for explanations.

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/projects-and-domains ===

# Projects and domains

Projects and domains are the principle organizational categories into which you group your workflows in Flyte.

Projects define groups of task, workflows, launch plans and other entities that share a functional purpose.
Domains represent distinct steps through which the entities in a project transition as they proceed through the development cycle.

By default, Flyte provides three domains: `development`, `staging`, and `production`.
During onboarding, you can configure your Flyte instance to have different domains.
Speak to the Flyte team for more information.

Projects and domains are orthogonal to each other, meaning that a project has
multiple domains and a domain has multiple projects.

Here is an example arrangement:

|           | Development       | Staging           | Production        |
|-----------|-------------------|-------------------|-------------------|
| Project 1 | workflow_1 (v2.0) | workflow_1 (v1.0) | workflow_1 (v1.0) |
| Project 2 | workflow_2 (v2.0) | workflow_2 (v1.0) | workflow_2 (v1.0) |

## Projects

Projects represent independent workflows related to specific teams, business
areas, or applications.  Each project is isolated from others, but workflows can
reference entities (workflows or tasks) from other projects to reuse
generalizable resources.

## Domains

Domains represent distinct environments orthogonal to the set of projects in
your org within Flyte, such as development, staging, and production.  These
enable dedicated configurations, permissions, secrets, cached execution history,
and resource allocations for each environment, preventing unintended impact on
other projects and/or domains.

Using domains allows for a clear separation between environments, helping ensure
that development and testing don't interfere with production workflows.

A production domain ensures a “clean slate” so that cached development
executions do not result in unexpected behavior.  Additionally, secrets may be
configured for external production data sources.

## When to use different Flyte projects?

Projects help group independent workflows related to specific teams, business
areas, or applications.  Generally speaking, each independent team or ML product
should have its own Flyte project.  Even though these are isolated from one
another, teams may reference entities (workflows or tasks) from other Flyte
projects to reuse generalizable resources.  For example, one team may create a
generalizable task to train common model types.  However, this requires advanced
collaboration and common coding standards.

When setting up workflows in Flyte, effective use of **projects** and
**domains** is key to managing environments, permissions, and resource
allocation.  Below are best practices to consider when organizing workflows in
Flyte.

## Projects and Domains: The Power of the Project-Domain Pair

Flyte uses a project-domain pair to create isolated configurations for
workflows. This pairing allows for:

* **Dedicated Permissions**: Through Role-Based Access Control (RBAC), users can be assigned roles with tailored permissions—such as contributor or admin—specific to individual project-domain pairs. This allows fine-grained control over who can manage or execute workflows within each pair, ensuring that permissions are both targeted and secure. More details [here](https://www.union.ai/docs/v1/flyte/user-guide/administration/user-management).

* **Resource and Execution Monitoring**: Track and monitor resource utilization, executions, and performance metrics on a dashboard unique to each project-domain pair. This helps maintain visibility over workflow execution and ensures optimal performance. More details [here](https://www.union.ai/docs/v1/flyte/user-guide/administration/resources).

* **Resource Allocations and Quotas**: By setting quotas for each project-domain pair, Flyte can ensure that workflows do not exceed designated limits, preventing any project or domain from unintentionally impacting resources available to others. Additionally, you can configure unique resource defaults—such as memory, CPU, and storage allocations—for each project-domain pair. This allows each pair to meet the specific requirements of its workflows, which is particularly valuable given the unique needs across different projects. More details [here](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/tasks/task-hardware-environment/customizing-task-resources) and [here](https://www.union.ai/docs/v1/flyte/user-guide/administration/resources).

* **Configuring Secrets**: Flyte allows you to configure secrets at the project-domain level, ensuring sensitive information, such as API keys and tokens, is accessible only within the specific workflows that need them. This enhances security by isolating secrets according to the project and domain, reducing the risk of unauthorized access across environments. More details [here](./managing-secrets).

## Domains: Clear Environment Separation

Domains represent distinct environments within Flyte, allowing clear separation between development, staging, and production. This structure helps prevent cross-environment interference, ensuring that changes made in development or testing do not affect production workflows. Using domains for this separation ensures that workflows can evolve in a controlled manner across different stages, from initial development through to production deployment.

## Projects: Organizing Workflows by Teams, Business Areas, or Applications

Projects in Flyte are designed to group independent workflows around specific teams, business functions, or applications. By aligning projects to organizational structure, you can simplify access control and permissions while encouraging a clean separation of workflows across different teams or use cases. Although workflows can reference each other across projects, it's generally cleaner to maintain independent workflows within each project to avoid complexity.

Flyte’s CLI tools and SDKs provide options to specify projects and domains easily:

* **CLI Commands**: In most commands within the `pyflyte` and `uctl` CLIs, you can specify the project and domain by using the `--project` and `--domain` flags, enabling precise control over which project-domain pair a command applies to. More details [here](https://www.union.ai/docs/v1/flyte/api-reference/union-cli) and [here](https://www.union.ai/docs/v1/flyte/api-reference/uctl-cli).

* **Python SDK**: When working with the `flytekit` SDK, you can leverage `FlyteRemote` to define the project and domain for workflow interactions programmatically, ensuring that all actions occur in the intended environment. More details [here](union-remote).

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/building-workflows ===

# Building workflows

## When should I decompose tasks?

There are several reasons why one may choose to decompose a task into smaller tasks.
Doing so may result in better computational performance, improved cache performance, and taking advantage of interruptible tasks.
However, decomposition comes at the cost of the overhead among tasks, including spinning up nodes and downloading data.
In some cases, these costs may be remediated by using [Actors](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/actors).

### Differing runtime requirements

Firstly, decomposition provides support for heterogeneous environments among the operations in the task.
For example, you may have some large task that trains a machine learning model and then uses the model to run batch inference on your test data.
However, training a model typically requires significantly more memory than inference.
For that reason, given large enough scale, it could actually be beneficial to decompose this large task into two tasks that (1) train a model and then (2) run batch inference.
By doing so, you could request significantly less memory for the second task in order to save on the expense of this workflow.
If you are working with even more data, then you might benefit from decomposing the batch inference task via `map_task` such that you may further parallelize this operation, substantially reducing the runtime of this step.
Generally speaking, decomposition provides infrastructural flexibility regarding the ability to define resources, dependencies, and execution parallelism.

### Improved cache performance

Secondly, you may decompose large tasks into smaller tasks to enable “fine-grained” caching.
In other words, each unique task provides an automated “checkpoint” system.
Thus, by breaking down a large workflow into its many natural tasks, one may minimize redundant work among multiple serial workflow executions.
This is especially useful during rapid, iterative development, during which a user may attempt to run the same workflow multiple times in a short period of time.
“Fine-grained” caching will dramatically improve productivity while executing workflows both locally and remotely.

### Take advantage of interruptible tasks

Lastly, one may utilize “fine-grained” caching to leverage interruptible tasks.
Interruptible tasks will attempt to run on spot instances or spot VMs, where possible.
These nodes are interruptible, meaning that the task may occasionally fail due to another organization willing to pay more to use it.
However, these spot instances can be substantially cheaper than their non-interruptible counterparts (on-demand instances / VMs).
By utilizing “fine-grained” caching, one may reap the significant cost savings on interruptible tasks while minimizing the effects of having their tasks being interrupted.

## When should I parallelize tasks?

In general, parallelize early and often.
A lot of Flyte’s powerful ergonomics like caching and workflow recovery happen at the task level, as mentioned above.
Decomposing into smaller tasks and parallelizing enables for a performant and fault-tolerant workflow.

One caveat is for very short duration tasks, where the overhead of spinning up a pod and cleaning it up negates any benefits of parallelism.
With reusable containers via [Actors](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/actors), however, these overheads are transparently obviated, providing the best of both worlds at the cost of some up-front work in setting up that environment.
In any case, it may be useful to batch the inputs and outputs to amortize any overheads.
Please be mindful to keep the sequencing of inputs within a batch, and of the batches themselves, to ensure reliable cache hits.

### Parallelization constructs

The two main parallelization constructs in Flyte are the [map task](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/tasks/task-types) and the [dynamic workflow](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/workflows/dynamic-workflows).
They accomplish roughly the same goal but are implemented quite differently and have different advantages.

Dynamic tasks are more akin to a `for` loop, iterating over inputs sequentially.
The parallelism is controlled by the overall workflow parallelism.

Map tasks are more efficient and have no such sequencing guarantees.
They also have their own concurrency setting separate from the overall workflow and can have a minimum failure threshold of their constituent tasks.
A deeper explanation of their differences is available [here]() while examples of how to use them together can be found [here]().

## When should I use caching?

Caching should be enabled once the body of a task has stabilized.
Cache keys are implicitly derived from the task signature, most notably the inputs and outputs.
If the body of a task changes without a modification to the signature, and the same inputs are used, it will produce a cache hit.
This can result in unexpected behavior when iterating on the core functionality of the task and expecting different inputs downstream.
Moreover, caching will not introspect the contents of a `FlyteFile` for example.
If the same URI is used as input with completely different contents, it will also produce a cache hit.
For these reasons, it’s wise to add an explicit cache key so that it can be invalidated at any time.

Despite these caveats, caching is a huge time saver during workflow development.
Caching upstream tasks enable a rapid run through of the workflow up to the node you’re iterating on.
Additionally, caching can be valuable in complex parallelization scenarios where you’re debugging the failure state of large map tasks, for example.
In production, if your cluster is under heavy resource constraints, caching can allow a workflow to complete across re-runs as more and more tasks are able to return successfully with each run.
While not an ideal scenario, caching can help soften the blow of production failures.
With these caveats in mind, there are very few scenarios where caching isn’t warranted.

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/setting-up-a-project ===

# Setting up a production project

In Flyte, your work is organized in a hierarchy with the following structure:

* **Organization**: Your Flyte instance, accessible at a specific URL like `flyte.my-company.com`.
* **Domains** Within an organization there are (typically) three domains, `development`, `staging`, and `production`, used to organize your code during the development process.
You can configure a custom set of domains to suit your needs during [onboarding](https://www.union.ai/docs/v1/flyte/user-guide/deployment/configuring-your-data-plane).
* **Projects**: Orthogonal to domains, projects are used to organize your code into logical groups. You can create as many projects as you need.

A given workflow will reside in a specific project. For example, let's say `my_workflow` is a workflow in `my_project`.

When you start work on `my_workflow` you would typically register it in the project-domain `my_project/development`.

As you work on successive iterations of the workflow you might promote `my_workflow` to `my_project/staging` and eventually `my_project/production`.

Promotion is done simply by [re-registering the workflow to the new project-domain](./running-your-code).

## Terminology

In everyday use, the term "project" is often used to refer to not just the Flyte entity that holds a set of workflows,
but also to the local directory in which you are developing those workflows, and to the GitHub (or other SCM) repository that you are using to store the same workflow code.

To avoid confusion, in this guide we will stick to the following naming conventions:

* **Flyte project**: The entity in your Flyte instance that holds a set of workflows, as described above. Often referred to simply as a **project**.
* **Local project**: The local directory (usually the working directory of a GitHub repository) in which you are developing workflows.

## Create a Flyte project

Ensure that you have [`flytectl` CLI installed](https://www.union.ai/docs/v1/flyte/user-guide/getting-started/local-setup) and the connection to your Flyte cluster [properly configured](https://www.union.ai/docs/v1/flyte/user-guide/getting-started/local-setup).
Now, create a new project on your Flyte cluster:

```shell
$ flytectl create project \
      --id "my-project" \
      --labels "my-label=my-project" \
      --description "My Flyte project" \
      --name "My project"
```

## Creating a local production project directory using `pyflyte init`

Earlier, in the [Getting started](../getting-started/_index) section we used `pyflyte init`
to create a new local project based on the `flyte-simple`.

Here, we will do the same, but use the `flyte-production` template. Perform the following command:

```shell
$ pyflyte init --template union-production my-project
```

## Directory structure

In the `basic-example` directory you’ll see the following file structure:

```shell
├── LICENSE
├── README.md
├── docs
│   └── docs.md
├── pyproject.toml
├── src
│   ├── core
│   │   ├── __init__.py
│   │   └── core.py
│   ├── orchestration
│   │   ├── __init__.py
│   │   └── orchestration.py
│   ├── tasks
│   │   ├── __init__.py
│   │   └── say_hello.py
│   └── workflows
│       ├── __init__.py
│       └── hello_world.py
└── uv.lock
```

You can create your own conventions and file structure for your production projects, but this tempkate provides a good starting point.

However, the separate `workflows` subdirectory and the contained `__init__.py` file are significant.
We will discuss them when we cover the [registration process](./running-your-code).

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/local-dependencies ===

# Local dependencies

During the development cycle you will want to be able to run your workflows both locally on your machine and remotely on Flyte.
To enable this, you need to ensure that the required dependencies are installed in both places.
Here we will explain how to install your dependencies locally.
For information on how to make your dependencies available on Flyte, see [ImageSpec](./image-spec).

## Define your dependencies in your `pyproject.toml`

We recommend using the [`uv` tool](https://docs.astral.sh/uv/) for project and dependency management.

When using the best way declare your dependencies is to list them under `dependencies` in your `pyproject.toml` file, like this:

```toml
[project]
name = "union-simple"
version = "0.1.0"
description = "A simple Flyte project"
readme = "README.md"
requires-python = ">=3.9,<3.13"
dependencies = ["union"]
```

## Create a Python virtual environment

Ensure that your Python virtual environment is properly set up with the required dependencies.

Using `uv`, you can install the dependencies with the command:

```shell
$ uv sync
```

You can then activate the virtual environment with:

```shell
$ source .venv/bin/activate
```

> [!NOTE] `activate` vs `uv run`
> When running the Pyflyte CLI within your local project you must run it in the virtual environment _associated with_ that project.
>
> To run `pyflyte` within your project's virtual environment using `uv`, you can prefix it use the `uv run` command. For example:
>
> `uv run pyflyte ...`
>
> Alternatively, you can activate the virtual environment with `source .venv/bin/activate` and then run the `pyflyte` command directly.
> In our examples we assume that you are doing the latter.

Having installed your dependencies in your local environment, you can now [run your workflows locally using `pyflyte run`](./running-your-code).

The next step is to ensure that the same dependencies are also [available in the remote environment on Flyte](./image-spec).

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/image-spec ===

# ImageSpec

During the development cycle you will want to be able to run your workflows both locally on your machine and remotely on Flyte,
so you will need to ensure that the required dependencies are installed in both environments.

Here we will explain how to set up the dependencies for your workflow to run remotely on Flyte.
For information on how to make your dependencies available locally, see [Local dependencies](./local-dependencies).

When a workflow is deployed to Flyte, each task is set up to run in its own container in the Kubernetes cluster.
You specify the dependencies as part of the definition of the container image to be used for each task using the `ImageSpec` class.
For example::

```python
import flytekit as fl

image_spec = union.ImageSpec(

    name="say-hello-image",
    requirements="uv.lock",
)

@fl.task(container_image=image_spec)
def say_hello(name: str) -> str:
    return f"Hello, {name}!"

@fl.workflow
def hello_world_wf(name: str = "world") -> str:
    greeting = say_hello(name=name)
    return greeting
```

Here, the `ImageSpec` class is used to specify the container image to be used for the `say_hello` task.

* The `name` parameter specifies the name of the image. This name will be used to identify the image in the container registry.

* The `requirements` parameter specifies the path to a file (relative to the directory in which the `pyflyte run` or `pyflyte register` command is invoked) that specifies the dependencies to be installed in the image.
  The file may be:
  * A `requirements.txt` file.
  * A `uv.lock` file generated by the `uv sync` command.
  * A `poetry.lock` file generated by the `poetry install` command.
  * A `pyproject.toml` file.

When you execute the `pyflyte run` or `pyflyte register` command, Flyte will build the container image defined in `ImageSpec` block
(as well as registering the tasks and workflows defined in your code).

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/running-your-code ===

# Running your code

## Set up your development environment

If you have not already done so, follow the [Getting started](https://www.union.ai/docs/v1/flyte/user-guide/getting-started) section to sign in to Flyte, and set up your local environment.

## CLI commands for running your code

The Pyflyte CLI and Flytectl CLI provide commands that allow you to deploy and run your code at different stages of the development cycle:

<!-- TODO: Link to the union commands below to the reference section -->

1. `pyflyte run`: For deploying and running a single script immediately in your local Python environment.
2. `pyflyte run --remote`: For deploying and running a single script immediately in the cloud on Flyte.
3. `pyflyte register`: For deploying multiple scripts to Flyte and running them from the Web interface.

4. `pyflyte package` and `flytectl register`: For deploying workflows to production and for scripting within a CI/CD pipeline.

> [!NOTE]
> In some cases, you may want to test your code in a local cluster before deploying it to Flyte.
> This step corresponds to using the commands 2, 3, or 4, but targeting your local cluster instead of Flyte.
> For more details, see [Running in a local cluster](./running-in-a-local-cluster).

## Registration pattern summary

The following diagram provides a summarized view of the different registration patterns:

![Registration patterns](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/running-your-code/registration-patterns.png)

## Running a script in local Python with `pyflyte run` {#running-a-script-in-local-python}

During the development cycle you will want to run a specific workflow or task in your local Python environment to test it.
To quickly try out the code locally use `pyflyte run`:

```shell
$ pyflyte run workflows/example.py wf --name 'Albert'
```

Here you are invoking `pyflyte run` and passing the name of the Python file and the name of the workflow within that file that you want to run.
In addition, you are passing the named parameter `name` and its value.

This command is useful for quickly testing a workflow locally to check for basic errors.
For more details see [pyflyte run details](./details-of-pyflyte-run).

## Running a script on Flyte with `pyflyte run --remote`

To quickly run a workflow on Flyte, use `pyflyte run --remote`:

```shell
$ pyflyte run --remote --project basic-example --domain development workflows/example.py wf --name 'Albert'
```

Here we are invoking `pyflyte run --remote` and passing:
* The project, `basic-example`
* The domain, `development`
* The Python file, `workflows/example.py`
* The workflow within that file that you want to run, `wf`
* The named parameter `name`, and its value

This command will:
* Build the container image defined in your `ImageSpec`.

* Push the image to the container registry specified in that `ImageSpec`. Don't forget make the image accessible to Flyte. For example, if you are using GitHub Container Registry, you will need to make the image public.

* Package up your code and deploy it to the specified project and domain in Flyte.
* Run the workflow on Flyte.

This command is useful for quickly deploying and running a specific workflow on Flyte.
For more details see [pyflyte run details](./details-of-union-run).

This command is useful for quickly deploying and running a specific workflow on Flyte.
For more details see [pyflyte run details](./details-of-pyflyte-run).

## Running tasks through flytectl

This is a multi-step process where we create an execution spec file, update the spec file, and then create the execution.

### Generate execution spec file

```shell
$ flytectl launch task --project flytesnacks --domain development --name workflows.example.generate_normal_df --version v1
```

### Update the input spec file for arguments to the workflow

```yaml
iamRoleARN: 'arn:aws:iam::12345678:role/defaultrole'
inputs:
  n: 200
  mean: 0.0
  sigma: 1.0
kubeServiceAcct: ""
targetDomain: ""
targetProject: ""
task: workflows.example.generate_normal_df
version: "v1"
```

### Create execution using the exec spec file

```shell
$ flytectl create execution -p flytesnacks -d development --execFile exec_spec.yaml
```

### Monitor the execution by providing the execution id from create command

```shell
$ flytectl get execution -p flytesnacks -d development <execid>
```

## Running workflows through flytectl

Workflows on their own are not runnable directly. However, a launchplan is always bound to a workflow (at least the auto-create default launch plan) and you can use
launchplans to `launch` a workflow. The `default launchplan` for a workflow has the same name as its workflow and all argument defaults are also identical.

Tasks also can be executed using the launch command.
One difference between running a task and a workflow via launchplans is that launchplans cannot be associated with a
task. This is to avoid triggers and scheduling.

## Running launchplans through flytectl

This is multi-step process where we create an execution spec file, update the spec file and then create the execution.
More details can be found [here](https://www.union.ai/docs/v1/flyte/api-reference/flytectl-cli/flytectl-create/flytectl-create-execution).

### Generate an execution spec file

```shell
$ flytectl get launchplan -p flytesnacks -d development myapp.workflows.example.my_wf  --execFile exec_spec.yaml
```

### Update the input spec file for arguments to the workflow

```yaml
inputs:
    name: "adam"
```

### Create execution using the exec spec file

```shell
$ flytectl create execution -p flytesnacks -d development --execFile exec_spec.yaml
```

### Monitor the execution by providing the execution id from create command

```bash
$ flytectl get execution -p flytesnacks -d development <execid>
```

## Deploying your code to Flyte with `pyflyte register`

```shell
$ pyflyte register workflows --project basic-example --domain development
```

Here we are registering all the code in the `workflows` directory to the project `basic-example` in the domain `development`.

This command will:
* Build the container image defined in your `ImageSpec`.
* Package up your code and deploy it to the specified project and domain in Flyte.
  The package will contain the code in the Python package located in the `workflows` directory.
  Note that the presence of the `__init__.py` file in this directory is necessary in order to make it a Python package.

The command will not run the workflow. You can run it from the Web interface.

This command is useful for deploying your full set of workflows to Flyte for testing.

### Fast registration

`pyflyte register` packages up your code through a mechanism called fast registration.
Fast registration is useful when you already have a container image that’s hosted in your container registry of choice, and you change your workflow/task code without any changes in your system-level/Python dependencies. At a high level, fast registration:

* Packages and zips up the directory/file that you specify as the argument to `pyflyte register`, along with any files in the root directory of your project. The result of this is a tarball that is packaged into a `.tar.gz` file, which also includes the serialized task (in `protobuf` format) and workflow specifications defined in your workflow code.

* Registers the package to the specified cluster and uploads the tarball containing the user-defined code into the configured blob store (e.g. S3, GCS).

At workflow execution time, Flyte knows to automatically inject the zipped up task/workflow code into the running container, thereby overriding the user-defined tasks/workflows that were originally baked into the image.

<!-- TODO: determine if this section should be included. There was some discussion of flyteignore no longer being relevant

Ignoring files during fast registration

In step (1) of the fast registration process, by default Flyte will package up all user-defined code at the root of your project. In some cases, your project directory may contain datasets, model files, and other potentially large artifacts that you want to exclude from the tarball.

You can do so by specifying these files in a .flyteignore file in the root of your project. You can also use .gitignore or .dockerignore if you’d like to avoid adding another file.
-->

> [!NOTE] `WORKDIR`, `PYTHONPATH`, and `PATH`
> When executing any of the above commands, the archive that gets creates is extracted wherever the `WORKDIR` is set.
> This can be handled directly via the `WORKDIR` directive in a `Dockerfile`, or specified via `source_root` if using `ImageSpec`.
> This is important for discovering code and executables via `PATH` or `PYTHONPATH`.
> A common pattern for making your Python packages fully discoverable is to have a top-level `src` folder, adding that to your `PYTHONPATH`,
> and making all your imports absolute.
> This avoids having to “install” your Python project in the image at any point e.g. via `pip install -e`.

## Inspecting executions

Flytectl supports inspecting execution by retrieving its details. For a deeper dive, refer to the
[Reference](https://www.union.ai/docs/v1/flyte/user-guide/api-reference/flytectl-cli/_index) guide.

Monitor the execution by providing the execution id from create command which can be task or workflow execution.

```shell
$ flytectl get execution -p flytesnacks -d development <execid>
```

For more details use `--details` flag which shows node executions along with task executions on them.

```shell
$ flytectl get execution -p flytesnacks -d development <execid> --details
```

If you prefer to see yaml/json view for the details then change the output format using the -o flag.

```shell
$ flytectl get execution -p flytesnacks -d development <execid> --details -o yaml
```

To see the results of the execution you can inspect the node closure outputUri in detailed yaml output.

```shell
"outputUri": "s3://my-s3-bucket/metadata/propeller/flytesnacks-development-<execid>/n0/data/0/outputs.pb"
```

## Deploying your code to production

### Package your code with `pyflyte package`

The combination of `pyflyte package` and `flytectl register` is the standard way of deploying your code to production.
This method is often used in scripts to [build and deploy workflows in a CI/CD pipeline](./ci-cd-deployment).

First, package your workflows:

```shell
$ pyflyte --pkgs workflows package
```

This will create a tar file called `flyte-package.tgz` of the Python package located in the `workflows` directory.
Note that the presence of the `__init__.py` file in this directory is necessary in order to make it a Python package.

> [!NOTE]
> You can specify multiple workflow directories using the following command:
>
> `pyflyte --pkgs DIR1 --pkgs DIR2 package ...`
>
> This is useful in cases where you want to register two different projects that you maintain in a single place.
>
> If you encounter a ModuleNotFoundError when packaging, use the --source option to include the correct source paths. For instance:
>
> `pyflyte --pkgs <dir1> package --source ./src -f`

### Register the package with `flytectl register`

Once the code is packaged you register it using the `flytectl` CLI:

```shell
$ flytectl register files \
      --project basic-example \
      --domain development \
      --archive flyte-package.tgz \
      --version "$(git rev-parse HEAD)"
```

Let’s break down what each flag is doing here:

* `--project`: The target Flyte project.

* `--domain`:  The target domain. Usually one of `development`, `staging`, or `production`.

* `--archive`: This argument allows you to pass in a package file, which in this case is the `flyte-package.tgz` produced earlier.

* `--version`: This is a version string that can be any string, but we recommend using the Git SHA in general, especially in production use cases.

See [Flytectl CLI](https://www.union.ai/docs/v1/flyte/user-guide/api-reference/flytectl-cli/_index) for more details.

## Using pyflyte register versus pyflyte package + flytectl register

As a rule of thumb, `pyflyte register` works well when you are working on a single cluster and iterating quickly on your task/workflow code.

On the other hand, `pyflyte package` and `flytectl register` is appropriate if you are:

* Working with multiple clusters, since it uses a portable package

* Deploying workflows to a production context

* Testing your workflows in your CI/CD infrastructure.

> [!NOTE] Programmatic Python API
> You can also perform the equivalent of the three methods of registration using a [FlyteRemote object](../development-cycle/union-remote/_index).

## Image management and registration method

The `ImageSpec` construct available in `flytekit` also has a mechanism to copy files into the image being built.
Its behavior depends on the type of registration used:

* If fast register is used, then it’s assumed that you don’t also want to copy source files into the built image.

* If fast register is not used (which is the default for `pyflyte package`, or if `pyflyte register --copy none` is specified), then it’s assumed that you do want source files copied into the built image.

* If your `ImageSpec` constructor specifies a `source_root` and the `copy` argument is set to something other than `CopyFileDetection.NO_COPY`, then files will be copied regardless of fast registration status.

## Building your own images

While we recommend that you use `ImageSpec` and the `envd` image builder on registration, you can, if you wish build and deploy your own images separately.

You can start with `pyflyte init --template basic-template-dockerfile`, the resulting template project includes a `docker_build.sh` script that you can use to build and tag a container according to the recommended practice:

```shell
$ ./docker_build.sh
```

By default, the `docker_build.sh` script:

* Uses the `PROJECT_NAME` specified in the pyflyte  command, which in this case is my_project.

* Will not use any remote registry.

* Uses the Git SHA to version your tasks and workflows.

You can override the default values with the following flags:

```shell
$ ./docker_build.sh -p PROJECT_NAME -r REGISTRY -v VERSION
```

For example, if you want to push your Docker image to Github’s container registry you can specify the `-r ghcr.io` flag.

> [!NOTE]
> The `docker_build.sh` script is purely for convenience; you can always roll your own way of building Docker containers.

Once you’ve built the image, you can push it to the specified registry. For example, if you’re using Github container registry, do the following:

```shell
$ docker login ghcr.io
$ docker push TAG
```

## CI/CD with Flyte and GitHub Actions

You can use any of the commands we learned in this guide to register, execute, or test Flyte workflows in your CI/CD process.
Flyte provides two GitHub actions that facilitate this:

* `flyte-setup-action`: This action handles the installation of flytectl in your action runner.

* `flyte-register-action`: This action uses `flytectl register` under the hood to handle registration of packages, for example, the `.tgz` archives that are created by `pyflyte package`.

### Some CI/CD best practices

In the case where workflows are registered on each commit in your build pipelines, you can consider the following recommendations and approach:

* **Versioning Strategy** : Determining the version of the build for different types of commits makes them consistent and identifiable.
  For commits on feature branches, use `{branch-name}-{short-commit-hash}` and for the ones on main branches, use `main-{short-commit-hash}`.
  Use version numbers for the released (tagged) versions.

* **Workflow Serialization and Registration** : Workflows should be serialized and registered based on the versioning of the build and the container image.
  Depending on whether the build is for a feature branch or `main`, the registration domain should be adjusted accordingly.

* **Container Image Specification** : When managing multiple images across tasks within a workflow, use the `--image` flag during registration to specify which image to use.
  This avoids hardcoding the image within the task definition, promoting reusability and flexibility in workflows.

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/overriding-parameters ===

# Overriding parameters

The `with_overrides` method allows you to specify parameter overrides on [tasks](../core-concepts/tasks/_index),
[subworkflows, and sub-launch plans](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/workflows/subworkflows-and-sub-launch-plans) at execution time.
This is useful when you want to change the behavior of a task, subworkflow, or sub-launch plan without modifying the original definition.

## Task parameters

When calling a task, you can specify the following parameters in `with_overrides`:

* `accelerator`: Specify [accelerators](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/tasks/task-hardware-environment/accelerators).
* `cache_serialize`: Enable [cache serialization](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/caching).
* `cache_version`: Specify the [cache version](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/caching).
* `cache`: Enable [caching](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/caching).
* `container_image`: Specify a [container image](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/tasks/task-software-environment/image-spec).
* `interruptible`: Specify whether the task is [interruptible](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/tasks/task-hardware-environment/interruptible-instances).
* `limits`: Specify [resource limits](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/tasks/task-hardware-environment/customizing-task-resources).
* `name`: Give a specific name to this task execution. This will appear in the workflow flowchart in the UI (see **Development cycle > Overriding parameters > Task parameters > Using `with_overrides` with `name` and `node_name`**).
* `node_name`: Give a specific name to the DAG node for this task. This will appear in the workflow flowchart in the UI (see **Development cycle > Overriding parameters > Task parameters > Using `with_overrides` with `name` and `node_name`**).
* `requests`: Specify [resource requests](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/tasks/task-hardware-environment/customizing-task-resources).
* `retries`: Specify the [number of times to retry this task](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/tasks/task-parameters).
* `task_config`: Specify a [task config](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/tasks/task-parameters).
* `timeout`: Specify the [task timeout](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/tasks/task-parameters).

For example, if you have a task that does not have caching enabled, you can use `with_overrides` to enable caching at execution time as follows:

```python
my_task(a=1, b=2, c=3).with_overrides(cache=True)
```

### Using `with_overrides` with `name` and `node_name`

Using `with_overrides` with `name` on a task is a particularly useful feature.
For example, you can use `with_overrides(name="my_task")` to give a specific name to a task execution, which will appear in the UI.
The name specified can be chosen or generated at invocation time without modifying the task definition.

```python
@fl.workflow
def wf() -> int:
    my_task(a=1, b=1, c=1).with_overrides(name="my_task_1")
    my_task(a=2, b=2, c=2).with_overrides(name="my_task_2", node_name="my_node_2")
    return my_task(a=1, b=1, c=1)
```

The above code would produce the following workflow display in the UI:

![Overriding name](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/overriding-parameters/override-name.png)

There is also a related parameter called `node_name` that can be used to give a specific name to the DAG node for this task.
The DAG node name is usually autogenerated as `n0`, `n1`, `n2`, etc. It appears in the `node` column of the workflow table.
Overriding `node_name` results in the autogenerated name being replaced by the specified name:

![Overriding node name](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/overriding-parameters/override-node-name.png)

Note that the `node_name` was specified as `my_node_2` in the code but appears as `my_node_2` in the UI. This is to the fact that Kubernetes node names cannot contain underscores. Flyte automatically alters the name to be Kubernetes-compliant.

## Subworkflow and sub-launch plan parameters

When calling a workflow or launch plan from within a high-level workflow
(in other words, when invoking a subworkflow or sub-launch plan),
you can specify the following parameters in `with_overrides`:

* `cache_serialize`: Enable [cache serialization](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/caching).
* `cache_version`: Specify the [cache version](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/caching).
* `cache`: Enable [caching](https://www.union.ai/docs/v1/flyte/user-guide/core-concepts/caching).

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/run-details ===

# Run details

The `pyflyte run` command is used to run a specific workflow or task in your local Python environment or on Flyte.
In this section we will discuss some details of how and why to use it.

## Passing parameters

`pyflyte run` enables you to execute a specific workflow using the syntax:

```shell
$ pyflyte run <path/to/script.py> <workflow_or_task_function_name>
```

Keyword arguments can be supplied to `pyflyte run` by passing them in like this:

```shell
--<keyword> <value>
```

For example, above we invoked `pyflyte run` with script `example.py`, workflow `wf`, and named parameter `name`:

```shell
$ pyflyte run example.py wf --name 'Albert'
```

The value `Albert` is passed for the parameter `name`.

With `snake_case` argument names, you have to convert them to `kebab-case`. For example,
if the code were altered to accept a `last_name` parameter then the following command:

```shell
$ pyflyte run example.py wf --last-name 'Einstein'
```

This passes the value `Einstein` for that parameter.

## Why `pyflyte run` rather than `python`?

You could add a `main` guard at the end of the script like this:

```python
if __name__ == "__main__":
    training_workflow(hyperparameters={"C": 0.1})
```

This would let you run it with `python example.py`, though you have to hard code your arguments.

It would become even more verbose if you want to pass in your arguments:

```python
if __name__ == "__main__":
    import json
    from argparse import ArgumentParser

    parser = ArgumentParser()
    parser.add_argument("--hyperparameters", type=json.loads)
    ...  # add the other options

    args = parser.parse_args()
    training_workflow(hyperparameters=args.hyperparameters)Py
```

`pyflyte run` is less verbose and more convenient for running workflows with arguments.

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/debugging-with-interactive-tasks ===

# Debugging with interactive tasks

With interactive tasks you can inspect and debug live task code directly in the UI in an embedded Visual Studio Code IDE.

## Enabling interactive tasks in your code

To enable interactive tasks, you need to:

* Include `flytekitplugins-flyteinteractive` as a dependency
* Use the `@vscode` decorator on the tasks you want to make interactive.

The `@vscode` decorator, when applied, converts a task into a Visual Studio Code
server during runtime.  This process overrides the standard execution of the
task’s function body, initiating a command to start a Visual Studio Code server
instead.

<!-- TODO: Remove mention of flytesnacks and flytekit -->

> [!NOTE] No need for ingress or port forwarding
> The Flyte interactive tasks feature is an adaptation of the open-source
> [FlyteInteractive plugin](https://www.union.ai/docs/v1/flyte/integrations/external-service-backend-plugins/flyteinteractive-plugin).
> It improves on the open-source version by removing the need for ingress
> configuration or port forwarding, providing a more seamless debugging
> experience.

## Basic example

The following example demonstrates interactive tasks in a simple workflow.

### requirements.txt

This `requirements.txt` file is used by all the examples in this section:

```text
flytekit
flytekitplugins-flyteinteractive
```

### example.py

```python
"""Flyte workflow example of interactive tasks (@vscode)"""

import flytekit as fl
from flytekitplugins.flyteinteractive import vscode

image = fl.ImageSpec(
    registry="<my-image-registry>",
    name="interactive-tasks-example",
    base_image="ghcr.io/flyteorg/flytekit:py3.11-latest",
    requirements="requirements.txt"
)

@fl.task(container_image=image)
@vscode
def say_hello(name: str) -> str:
    s = f"Hello, {name}!"
    return s

@fl.workflow
def wf(name: str = "world") -> str:
    greeting = say_hello(name=name)
    return greeting
```

## Register and run the workflow

To register the code to a project on Flyte and run the workflow, follow the
directions in [Running your code](../development-cycle/running-your-code)

## Access the IDE

1. Select the first task in the workflow page (in this example the task is called `say_hello`).
   The task info pane will appear on the right side of the page.
2. Wait until the task is in the **Running** state and the **VSCode (User)** link appears.
3. Click the **VSCode (User)** link.

![VSCode link](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/debugging-with-interactive-tasks/vscode-link.png)

## Inspect the task code

Once the IDE opens, you will be able to see your task code in the editor.

![Inspect code](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/debugging-with-interactive-tasks/inspect-code.png)

## Interactive debugging

To run the task in VSCode, click the _Run and debug_ symbol on the left rail of the IDE and select the **Interactive Debugging** configuration.

![Interactive debugging](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/debugging-with-interactive-tasks/interactive-debugging.png)

Click the **Play** button beside the configuration drop-down to run the task.
This will run your task with inputs from the previous task. To inspect intermediate states, set breakpoints in the Python code and use the debugger for tracing.

> [!NOTE] No task output written to Flyte storage
> It’s important to note that during the debugging phase the task runs entirely within VSCode and does not write the output to Flyte storage.

## Update your code

You can edit your code in the VSCode environment and run the task again to see the changes.
Note, however, that the changes will not be automatically persisted anywhere.
You will have to manually copy and paste the changes back to your local environment.

## Resume task

After you finish debugging, you can resume your task with updated code by executing the **Resume Task** configuration.
This will terminate the code server, run the task with inputs from the previous task, and write the output to Flyte storage.

> [!NOTE] Remember to persist your code
> Remember to persist your code (for example, by checking it into GitHub) before resuming the task, since you will lose the connection to the VSCode server afterwards.

![Resume task](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/debugging-with-interactive-tasks/resume-task.png)

## Auxiliary Python files

You will notice that aside from your code, there are some additional files in the VSCode file explorer that have been automatically generated by the system:

### flyteinteractive_interactive_entrypoint.py

The `flyteinteractive_interactive_entrypoint.py` script implements the **Interactive Debugging** action that we used above:

![Interactive entrypoint](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/debugging-with-interactive-tasks/flyteinteractive-interactive-entrypoint-py.png)

### flyteinteractive_resume_task.py

The `flyteinteractive_resume_task.py` script implements the **Resume Task** action that we used above:

![Resume task](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/debugging-with-interactive-tasks/flyteinteractive-resume-task-py.png)

### launch.json

The `launch.json` file in the `.vscode` directory configures the **Interactive Debugging** and **Resume Task** actions.

![launch.json](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/debugging-with-interactive-tasks/launch-json.png)

## Integrated terminal

In addition to using the convenience functions defined by the auxiliary files, you can also run your Python code script directly from the integrated terminal using `python <script_name>.py` (in this example, `python hello.py`).

![Interactive terminal](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/debugging-with-interactive-tasks/interactive-terminal.png)

## Install extensions

As with local VSCode, you can install a variety of extensions to assist development.
Available extensions differ from official VSCode for legal reasons and are hosted on the [Open VSX Registry](https://open-vsx.org/).

Python and Jupyter extensions are installed by default.
Additional extensions can be added by defining a configuration object and passing it to the `@vscode` decorator, as shown below:

### example-extensions.py

```python
"""Flyte workflow example of interactive tasks (@vscode) with extensions"""

import flytekit as fl
from flytekitplugins.flyteinteractive import COPILOT_EXTENSION, VscodeConfig, vscode

image = fl.ImageSpec(
    registry="<my-image-registry>",
    name="interactive-tasks-example",
    base_image="ghcr.io/flyteorg/flytekit:py3.11-latest",
    requirements="requirements.txt"
)

config = VscodeConfig()
config.add_extensions(COPILOT_EXTENSION) # Use predefined URL
config.add_extensions(
    "https://open-vsx.org/api/vscodevim/vim/1.27.0/file/vscodevim.vim-1.27.0.vsix"
) # Copy raw URL from Open VSX

@fl.task(container_image=image)
@vscode(config=config)
def say_hello(name: str) -> str:
    s = f"Hello, {name}!"
    return s

@fl.workflow
def wf(name: str = "world") -> str:
    greeting = say_hello(name=name)
    return greeting
```

## Manage resources

To manage resources, the VSCode server is terminated after a period of idleness (no active HTTP connections).
Idleness is monitored via a heartbeat file.

The `max_idle_seconds` parameter can be used to set the maximum number of seconds the VSCode server can be idle before it is terminated.

### example-manage-resources.py

```python
"""Flyte workflow example of interactive tasks (@vscode) with max_idle_seconds"""

import flytekit as fl
from flytekitplugins.flyteinteractive import vscode

image = fl.ImageSpec(
    registry="<my-image-registry>",
    name="interactive-tasks-example",
    base_image="ghcr.io/flyteorg/flytekit:py3.11-latest",
    requirements="requirements.txt"
)

@fl.task(container_image=image)
@vscode(max_idle_seconds=60000)
def say_hello(name: str) -> str:
   s = f"Hello, {name}!"
   return s

@fl.workflow
def wf(name: str = "world") -> str:
    greeting = say_hello(name=name)
    return greeting
```

## Pre and post hooks

Interactive tasks also allow the registration of functions to be executed both before and after VSCode starts.
This can be used for tasks requiring setup or cleanup.

### example-pre-post-hooks.py

```python
"""Flyte workflow example of interactive tasks (@vscode) with pre and post hooks"""

import flytekit as fl
from flytekitplugins.flyteinteractive import vscode

image = fl.ImageSpec(
    registry="<my-image-registry>",
    name="interactive-tasks-example",
    base_image="ghcr.io/flyteorg/flytekit:py3.11-latest",
    requirements="requirements.txt"
)

def set_up_proxy():
    print("set up")

def push_code():
    print("push code")

@fl.task(container_image=image)
@vscode(pre_execute=set_up_proxy, post_execute=push_code)
def say_hello(name: str) -> str:
    s = f"Hello, {name}!"
    return s

@fl.workflow
def wf(name: str = "world") -> str:
    greeting = say_hello(name=name)
    return greeting
```

## Only initiate VSCode on task failure

The system can also be set to only initiate VSCode _after a task failure_, preventing task termination and thus enabling inspection.
This is done by setting the `run_task_first` parameter to `True`.

### example-run-task-first.py

```python
"""Flyte workflow example of interactive tasks (@vscode) with run_task_first"""

import flytekit as fl
from flytekitplugins.flyteinteractive import vscode

image = fl.ImageSpec(
    registry="<my-image-registry>",
    name="interactive-tasks-example",
    base_image="ghcr.io/flyteorg/flytekit:py3.11-latest",
    requirements="requirements.txt"
)

@fl.task(container_image=image)
@vscode(run_task_first=True)
def say_hello(name: str) -> str:
    s = f"Hello, {name}!"
    return s

@fl.workflow
def wf(name: str = "world") -> str:
    greeting = say_hello(name=name)
    return greeting
```

## Debugging execution issues

The inspection of task and workflow execution provides log links to debug things further.

Using `--details` flag you can view node executions with log links.

```shell
└── n1 - FAILED - 2021-06-30 08:51:07.3111846 +0000 UTC - 2021-06-30 08:51:17.192852 +0000 UTC
    └── Attempt :0
        └── Task - FAILED - 2021-06-30 08:51:07.3111846 +0000 UTC - 2021-06-30 08:51:17.192852 +0000 UTC
        └── Logs :
            └── Name :Kubernetes Logs (User)
            └── URI :http://localhost:30082/#/log/flytectldemo-development/f3a5a4034960f4aa1a09-n1-0/pod?namespace=flytectldemo-development
```

Additionally, you can check the pods launched in `\<project>-\<domain>` namespace

```shell
$ kubectl get pods -n <project>-<domain>
```

The launched pods will have a prefix of execution name along with suffix of `nodeId`:

```shell
NAME                        READY   STATUS             RESTARTS   AGE
f65009af77f284e50959-n0-0   0/1     ErrImagePull       0          18h
```

For example, above we see that the `STATUS` indicates an issue with pulling the image.

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/task-resource-validation ===

# Task resource validation

In Flyte, when you attempt to execute a workflow with unsatisfiable resource requests, we fail the execution immediately rather than allowing it to queue forever.

We intercept execution creation requests in executions service to validate that their resource requirements can be met and fast-fail if not. A failed validation returns a message similar to

```text
Request failed with status code 400 rpc error: code = InvalidArgument desc = no node satisfies task 'workflows.fotd.fotd_directory' resource requests
```

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/running-in-a-local-cluster ===

# Running in a local cluster

## Running in a local Kubernetes cluster

Ultimately you will be running your workflows in a Kubernetes cluster in Flyte. But it can be handy to try out a workflow in a cluster on your local machine.

First, ensure that you have [Docker](https://www.docker.com/products/docker-desktop/) (or a similar OCI-compliant container engine) installed locally and that _the daemon is running_.

Then start the demo cluster using `flytectl`:

```shell
$ flytectl demo start
```

### Configuration

When `flytectl` starts the cluster in your local container engine it also writes configuration information to the directory `~/.flyte/`.

Most importantly, it creates the file `~/.flyte/config-sandbox.yaml`. This file holds (among other things) the location of the Kubernetes cluster to which we will be deploying the workflow:

```yaml
admin:
  endpoint: localhost:30080
  authType: Pkce
  insecure: true
console:
  endpoint: http://localhost:30080
logger:
  show-source: true
  level: 0
```

Right now this file indicates that the target cluster is your local Docker instance (`localhost:30080`), but later we will change it to point to your Flyte cluster.

Later invocations of `flytectl` or `pyflyte` will need to know the location of the target cluster. This can be provided in two ways:

1. Explicitly passing the location of the config file on the command line
   * `flytectl --config ~/.flyte/config-sandbox.yaml <command>`
   * `pyflyte --config ~/.flyte/config-sandbox.yaml <command>`
2. Setting the environment variable `FLYTECTL_CONFIG`to the location of the config file:
   * `export FLYTECTL_CONFIG=~/.flyte/config-sandbox.yaml`

> [!NOTE]
> In this guide, we assume that you have set the `FLYTECTL_CONFIG` environment variable in your shell to the location of the configuration file.

### Start the workflow

Now you can run your workflow in the local cluster simply by adding the `--remote` flag to your `pyflyte` command:

```shell
$ pyflyte run --remote \
          workflows/example.py \
          training_workflow \
          --hyperparameters '{"C": 0.1}'
```

The output supplies a URL to your workflow execution in the UI.

### Inspect the results

Navigate to the URL produced by `pyflyte run` to see your workflow in the Flyte UI.

## Local cluster with default image

```shell
$ pyflyte run --remote my_file.py my_workflow
```

_Where `pyflyte` is configured to point to the local cluster started with `flytectl demo start`._

* Task code runs in the environment of the default image in your local cluster.
* Python code is dynamically overlaid into the container at runtime.
* Only supports Python code whose dependencies are installed in the default image (see here).
* Includes a local S3.
* Supports some plugins but not all.
* Single workflow runs immediately.
* Workflow is registered to a default project.
* Useful for demos.

## Local cluster with custom image

```shell
$ pyflyte run --remote \
              --image my_cr.io/my_org/my_image:latest \
              my_file.py \
              my_workflow
```

_Where `pyflyte` is configured to point to the local cluster started with `flytectl demo start`._

* Task code runs in the environment of your custom image (`my_cr.io/my_org/my_image:latest`) in your local cluster.
* Python code is dynamically overlaid into the container at runtime
* Supports any Python dependencies you wish, since you have full control of the image.
* Includes a local S3.
* Supports some plugins but not all.
* Single workflow runs immediately.
* Workflow is registered to a default project.
* Useful for advanced testing during the development cycle.

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/jupyter-notebooks ===

# Jupyter notebooks

Flyte supports the development, running, and debugging of tasks and workflows in an interactive Jupyter notebook environment, which accelerates the iteration speed when building data- or machine learning-driven applications.

## Write your workflows and tasks in cells

When building tasks and workflows in a notebook, you write the code in cells as you normally would.

From those cells you can run the code locally (i.e., in the notebook itself, not on Flyte) by clikcing the run button, as you would in any notebook.

## Enable the notebook to register workflows to Flyte

To enable the tasks and workflows in your notebok to be easily registered and run on your Flyte instance, you needdto set up an _interactive_ FlyteRemote object and then use to invoke the remote executions:

First, in a cell, create an interactive FlyteRemote  object:

```python
from flytekit.configuration import Config
from flytekit.remote import FlyteRemote

remote = FlyteRemote(
    config=Config.auto(),
    default_project="default",
    default_domain="development",
    interactive_mode_enabled=True,
)
```

The `interactive_mode_enabled` flag must be set to `True` when running in a Jupyter notebook environment, enabling interactive registration and execution of workflows.

Next, set up the execution invocation in another cell:

```python
execution = remote.execute(my_task, inputs={"name": "Joe"})
execution = remote.execute(my_wf, inputs={"name": "Anne"})
```

The interactive FlyteRemote client re-registers an entity whenever it’s redefined in the notebook, including when you re-execute a cell containing the entity definition, even if the entity remains unchanged. This behavior facilitates iterative development and debugging of tasks and workflows in a Jupyter notebook.

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/decks ===

# Decks

Decks lets you display customized data visualizations from within your task code.
Decks are rendered as HTML and appear right in the Flyte UI when you run your workflow.

> [!NOTE]
> Decks is an opt-in feature; to enable it, set `enable_deck` to `True` in the task parameters.

To begin, import the dependencies:

```python
import flytekit as fl
from flytekit.deck.renderer import MarkdownRenderer
from sklearn.decomposition import PCA
import plotly.express as px
import plotly
```

> [!NOTE]
> The renderers are packaged separately from `flytekit` itself.
> To enable the `MarkdownRenderer` imported above
> you first have to install the package `flytekitplugins-deck-standard`
> in your local Python environment and include it in your `ImageSpec` (as shown below).

We create a new deck named `pca` and render Markdown content along with a
[PCA](https://en.wikipedia.org/wiki/Principal_component_analysis) plot.

Now, declare the required dependnecies in an `ImageSpec`:

```python
custom_image = fl.ImageSpec(
    packages=[
        "flytekitplugins-deck-standard",
        "markdown",
        "pandas",
        "pillow",
        "plotly",
        "pyarrow",
        "scikit-learn",
        "ydata_profiling",
    ],
)
```

Next, we define the task that will construct the figure and create the Deck:

```python
@fl.task(enable_deck=True, container_image=custom_image)
def pca_plot():
    iris_df = px.data.iris()
    X = iris_df[["sepal_length", "sepal_width", "petal_length", "petal_width"]]
    pca = PCA(n_components=3)
    components = pca.fit_transform(X)
    total_var = pca.explained_variance_ratio_.sum() * 100
    fig = px.scatter_3d(
        components,
        x=0,
        y=1,
        z=2,
        color=iris_df["species"],
        title=f"Total Explained Variance: {total_var:.2f}%",
        labels={"0": "PC 1", "1": "PC 2", "2": "PC 3"},
    )
    main_deck = fl.Deck("pca", MarkdownRenderer().to_html("### Principal Component Analysis"))
    main_deck.append(plotly.io.to_html(fig))
```

Note the usage of `append` to append the Plotly figure to the Markdown deck.

The following is the expected output containing the path to the `deck.html` file:

```
{"asctime": "2023-07-11 13:16:04,558", "name": "flytekit", "levelname": "INFO", "message": "pca_plot task creates flyte deck html to file:///var/folders/6f/xcgm46ds59j7g__gfxmkgdf80000gn/T/flyte-0_8qfjdd/sandbox/local_flytekit/c085853af5a175edb17b11cd338cbd61/deck.html"}
```

![Union deck plot](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/decks/flyte-deck-plot-local.webp)

Once you execute this task on the Flyte instance, you can access the deck by going to the task view and clicking the _Deck_ button:

![Union deck button](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/decks/flyte-deck-button.png)

## Deck tabs

Each Deck has a minimum of three tabs: input, output and default.
The input and output tabs are used to render the input and output data of the task,
while the default deck can be used to creta cusom renderings such as line plots, scatter plots, Markdown text, etc.
Additionally, you can create other tabs as well.

## Deck renderers

> [!NOTE]
> The renderers are packaged separately from `flytekit` itself.
> To enable them you first have to install the package `flytekitplugins-deck-standard`
> in your local Python environment and include it in your `ImageSpec`.

### Frame profiling renderer

The frame profiling render creates a profile report from a Pandas DataFrame.

```python
import flytekit as fl
import pandas as pd
from flytekitplugins.deck.renderer import FrameProfilingRenderer

@fl.task(enable_deck=True, container_image=custom_image)
def frame_renderer() -> None:
    df = pd.DataFrame(data={"col1": [1, 2], "col2": [3, 4]})
    fl.Deck("Frame Renderer", FrameProfilingRenderer().to_html(df=df))
```

![Frame renderer](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/decks/flyte-decks-frame-renderer.png)

### Top-frame renderer

The top-fram renderer renders a DataFrame as an HTML table.

```python
import flytekit as fl
from typing import Annotated
from flytekit.deck import TopFrameRenderer

@fl.task(enable_deck=True, container_image=custom_image)
def top_frame_renderer() -> Annotated[pd.DataFrame, TopFrameRenderer(1)]:
    return pd.DataFrame(data={"col1": [1, 2], "col2": [3, 4]})
```

![Top frame renderer](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/decks/flyte-decks-top-frame-renderer.png)

### Markdown renderer

The Markdown renderer converts a Markdown string into HTML.

```python
import flytekit as fl
from flytekit.deck import MarkdownRenderer

@fl.task(enable_deck=True, container_image=custom_image)
def markdown_renderer() -> None:
    fl.current_context().default_deck.append(
        MarkdownRenderer().to_html("You can install flytekit using this command: ```import flytekit```")
    )
```

![Markdown renderer](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/decks/flyte-decks-markdown-renderer.png)

### Box renderer

The box renderer groups rows of a DataFrame together into a
box-and-whisker mark to visualize their distribution.

Each box extends from the first quartile (Q1) to the third quartile (Q3).
The median (Q2) is indicated by a line within the box.
Typically, the whiskers extend to the edges of the box,
plus or minus 1.5 times the interquartile range (IQR: Q3-Q1).

```python
import flytekit as fl
from flytekitplugins.deck.renderer import BoxRenderer

@fl.task(enable_deck=True, container_image=custom_image)
def box_renderer() -> None:
    iris_df = px.data.iris()
    fl.Deck("Box Plot", BoxRenderer("sepal_length").to_html(iris_df))
```

![Box renderer](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/decks/flyte-decks-box-renderer.png)

### Image renderer

The image renderer converts a `FlyteFile` or `PIL.Image.Image` object into an HTML displayable image,
where the image data is encoded as a base64 string.

```python
import flytekit as fl
from flytekitplugins.deck.renderer import ImageRenderer

@fl.task(enable_deck=True, container_image=custom_image)
def image_renderer(image: fl.FlyteFile) -> None:
    fl.Deck("Image Renderer", ImageRenderer().to_html(image_src=image))

@fl.workflow
def image_renderer_wf(image: fl.FlyteFile = "https://bit.ly/3KZ95q4",) -> None:
    image_renderer(image=image)
```

![Image renderer](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/decks/flyte-decks-image-renderer.png)

#### Table renderer

The table renderer converts a Pandas DataFrame into an HTML table.

```python
import flytekit as fl
from flytekitplugins.deck.renderer import TableRenderer

@fl.task(enable_deck=True, container_image=custom_image)
def table_renderer() -> None:
    fl.Deck(
        "Table Renderer",
        TableRenderer().to_html(df=pd.DataFrame(data={"col1": [1, 2], "col2": [3, 4]}), table_width=50),
    )
```

![Table renderer](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/decks/flyte-decks-table-renderer.png)

### Contribute to renderers

Don't hesitate to integrate a new renderer into
[renderer.py](https://github.com/flyteorg/flytekit/blob/master/plugins/flytekit-deck-standard/flytekitplugins/deck/renderer.py)
if your deck renderers can enhance data visibility.
Feel encouraged to open a pull request and play a part in enhancing the Flyte deck renderer ecosystem!

### Custom renderers

You can also create your own custom renderer.
A renderer is essentially a class with a `to_html` method.
Here we create custom renderer that summarizes the data from a Pandas `DataFrame` instead of showing raw values.

```python
class DataFrameSummaryRenderer:

    def to_html(self, df: pd.DataFrame) -> str:
        assert isinstance(df, pd.DataFrame)
        return df.describe().to_html()
```

Then we can use the Annotated type to override the default renderer of the `pandas.DataFrame` type:

```python
try:
    from typing import Annotated
except ImportError:
    from typing_extensions import Annotated

@task(enable_deck=True)
def iris_data(
    sample_frac: Optional[float] = None,
    random_state: Optional[int] = None,
) -> Annotated[pd.DataFrame, DataFrameSummaryRenderer()]:
    data = px.data.iris()
    if sample_frac is not None:
        data = data.sample(frac=sample_frac, random_state=random_state)

    md_text = (
        "# Iris Dataset\n"
        "This task loads the iris dataset using the  `plotly` package."
    )
    flytekit.current_context().default_deck.append(MarkdownRenderer().to_html(md_text))
    flytekit.Deck("box plot", BoxRenderer("sepal_length").to_html(data))
    return data
```

## Streaming Decks

You can stream a Deck directly using `Deck.publish()`:

```python
import flytekit as fl

@task(enable_deck=True)
def t_deck():
    fl.Deck.publish()
```

This will create a live deck that where you can click the refresh button and see the deck update until the task succeeds.

### Union Deck Succeed Video

📺 [Watch on YouTube](https://www.youtube.com/watch?v=LJaBP0mdFeE)

### Union Deck Fail Video

📺 [Watch on YouTube](https://www.youtube.com/watch?v=xaBF6Jlzjq0)

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/migrating_from_airflow_to_flyte ===

# Migrating from Airflow to Flyte

> [!WARNING]
> Many Airflow operators and sensors have been tested on Flyte, but some may not work as expected.
If you encounter any issues, please file an [issue](https://github.com/flyteorg/flyte/issues) or reach out to the Flyte community on [Slack](https://slack.flyte.org/).

Flyte can compile Airflow tasks into Flyte tasks without changing code, which allows you
to migrate your Airflow DAGs to Flyte with minimal effort.

In addition to migration capabilities, Flyte users can seamlessly integrate Airflow tasks into their workflows, leveraging the ecosystem of Airflow operators and sensors.
By combining the robust Airflow ecosystem with Flyte's capabilities such as scalability, versioning, and reproducibility, users can run more complex data and machine learning workflows with ease.
For more information, see the [Airflow connector documentation](https://www.union.ai/docs/v1/flyte/integrations/connectors/airflow-connector).

Even if you're already using Flyte and have no intentions of migrating from Airflow,
you can still incorporate Airflow tasks into your Flyte workflows. For instance, Airflow offers support
for Google Cloud [Dataproc Operators](https://airflow.apache.org/docs/apache-airflow-providers-google/stable/operators/cloud/dataproc.html), facilitating the execution of Spark jobs on Google Cloud Dataproc clusters. Rather than developing a custom plugin in Flyte, you can seamlessly integrate Airflow's Dataproc Operators into your Flyte workflows to execute Spark jobs.

## Prerequisites

- Install `flytekitplugins-airflow` in your Python environment.
- Enable an {ref}`Airflow connector<deployment-connector-setup-airflow>` in your Flyte cluster.

## Steps

### 1. Define your Airflow tasks in a Flyte workflow

Flytekit compiles Airflow tasks into Flyte tasks, so you can use
any Airflow sensor or operator in a Flyte workflow:

```python
from flytekit import task, workflow
from airflow.sensors.filesystem import FileSensor

@task
def say_hello() -> str:
    return "Hello, World!"

@workflow
def airflow_wf():
    flyte_task = say_hello()
    airflow_task = FileSensor(task_id="sensor", filepath="/")
    airflow_task >> flyte_task

if __name__ == "__main__":
    print(f"Running airflow_wf() {airflow_wf()}")
```

### 2. Test your workflow locally

> [!NOTE]
Before running your workflow locally, you must configure the [Airflow connection](https://airflow.apache.org/docs/apache-airflow/stable/howto/connection.html) by setting the `AIRFLOW_CONN_{CONN_ID}` environment variable.
For example,
```bash
export AIRFLOW_CONN_MY_PROD_DATABASE='my-conn-type://login:password@host:port/schema?param1=val1&param2=val2'
```

Although Airflow doesn't support local execution, you can run your workflow that contains Airflow tasks locally, which is helpful for testing and debugging your tasks before moving to production.

```bash
AIRFLOW_CONN_FS_DEFAULT="/" pyflyte run workflows.py airflow_wf
```

> [!WARNING]
> Some Airflow operators may require certain permissions to execute. For instance, `DataprocCreateClusterOperator` requires the `dataproc.clusters.create` permission.
> When running Airflow tasks locally, you may need to set the necessary permissions locally for the task to execute successfully.

### 3. Move your workflow to production

> [!NOTE]
> In production, we recommend storing connections in a [secrets backend](https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/secrets-backend/index.html).
> Make sure the connector pod has the right permission (IAM role) to access the secret from the external secrets backend.

After you have tested your workflow locally, you can execute it on a Flyte cluster using the `--remote` flag.
In this case, Flyte creates a pod in the Kubernetes cluster to run the `say_hello` task, and then runs
your Airflow `BashOperator` task on the Airflow connector.

```bash
pyflyte run --remote workflows.py airflow_wf
```

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/remote-management ===

# FlyteRemote

The `FlyteRemote` Python API supports functionality similar to that of the Pyflyte CLI, enabling you to manage Flyte workflows, tasks, launch plans and artifacts from within your Python code.

> [!NOTE]
> The primary use case of `FlyteRemote` is to automate the deployment of Flyte entities. As such, it is intended for use within scripts *external* to actual Flyte workflow and task code, for example CI/CD pipeline scripts.
>
> In other words: _Do not use `FlyteRemote` within task code._

## Creating a `FlyteRemote` object

Ensure that you have the Flytekit SDK installed, import the `FlyteRemote` class and create the object like this:

```python
import flytekit as fl
from flytekit.configuration import Config

remote = fl.FlyteRemote(config=Config.auto())
```

By default, when created with `config=Config.auto()`, `FlyteRemote` will use the prevailing configuration in the local environment to connect to Flyte,
that is, the same configuration as would be used by the Pyflyte CLI in that environment
(see [Pyflyte CLI configuration search path](https://www.union.ai/docs/v1/flyte/api-reference/pyflyte-cli/page.md)).

In the default case, as with the Pyflyte CLI, all operations will be applied to the default project, `flytesnacks` and default domain, `development`.

Alternatively, you can initialize `FlyteRemote` by explicitly specifying a `flytekit.configuration.Config` object with connection information to a Flyte instance, a project, and a domain. Additionall, the constructor supports specifying a file upload location (equivalent to a default raw data prefix):

```python
import flytekit as fl
from flytekit.configuration import Config

remote = fl.FlyteRemote(
    config=Config.for_endpoint(endpoint="union.example.com"),
    default_project="my-project",
    default_domain="my-domain",
    data_upload_location="<s3|gs|abs>://my-bucket/my-prefix",
)
```

Here we use the `Config.for_endpoint` method to specify the URL to connect to.
There are other ways to configure the `Config` object.
In general, you have all the same options as you would when specifying a connection for the Pyflyte CLI using a `config.yaml` file.

### Authenticating using a client secret

In some cases, you may be running a script with `FlyteRemote` in a CI/CD pipeline or via SSH, where you don't have access to a browser for the default authentication flow. In such scenarios, you can use the [client secret](../../development-cycle/authentication#3-clientsecret-best-for-cicd-and-automation) authentication method to establish a connection to Flyte. After [creating an API key](../managing-api-keys), you can initialize `FlyteRemote` as follows:

```python
import flytekit as fl
from flytekit.configuration import Config, PlatformConfig

remote = fl.FlyteRemote(
        config=Config(
            platform=PlatformConfig(
                endpoint="union.example.com",
                insecure=False,
                client_id="<your-client-id>",  # this is the api-key name
                client_credentials_secret="<your-client-secret>",  # this is the api-key
                auth_mode="client_credentials",
            )
        ),
    )
```

For details see [the API docs for `flytekit.configuration.Config`](https://www.union.ai/docs/v1/flyte/api-reference/flytekit-sdk/flytekit.configuration/page.md)

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/remote-management/remote-examples ===

# FlyteRemote examples

## Registering and running a workflow

In the following example we register and run a workflow and retrieve its output:

```shell
├── remote.py
└── workflow
    ├── __init__.py
    └── example.py
```

The workflow code that will be registered and run on Flyte resides in the `workflow` directory and consists of an empty `__init__.py` file and the workflow and task code in `example.py`:

```python
import os
import flytekit as fl

@fl.task()
def create_file(message: str) -> fl.FlyteFile:
    with open("data.txt", "w") as f:
        f.write(message)
    return fl.FlyteFile(path="data.txt")

@fl.workflow
def my_workflow(message: str) -> fl.FlyteFile:
    f = create_file(message)
    return f
```

The file `remote.py` contains the `FlyteRemote` logic. It is not part of the workflow code, and is meant to be run on your local machine.

```python
import flytekit as fl
from workflow.example import my_workflow

def run_workflow():
    remote = fl.FlyteRemote()
    remote.fast_register_workflow(entity=my_workflow)
    execution = remote.execute(
        entity=my_workflow,
        inputs={"message": "Hello, world!"},
        wait=True)
    output = execution.outputs["o0"]
    print(output)
    with open(output, "r") as f:
        read_lines = f.readlines()
    print(read_lines)
```

The `my_workflow` workflow and the `create_file` task is registered and run.
Once the workflow completes, the output is passed back to the `run_workflow` function and printed out.

The output is also be available via the UI, in the **Outputs** tab of the `create_file` task details view:

![Outputs](https://www.union.ai/docs/v1/flyte/_static/images/user-guide/development-cycle/union-remote/outputs.png)

The steps above demonstrates the simplest way of registering and running a workflow with `FlyteRemote`.
For more options and details see [Reference > FlyteRemote](https://www.union.ai/docs/v1/flyte/api-reference/flytekit-sdk/flytekit.remote.remote).

## Fetching outputs

By default, `FlyteRemote.execute` is non-blocking, but you can also pass in `wait=True` to make it synchronously wait for the task or workflow to complete, as we did above.

You can print out the Flyte console URL corresponding to your execution with:

```python
print(f"Execution url: {remote.generate_console_url(execution)}")
```

And you can synchronize the state of the execution object with the remote state with the `sync()` method:

```python
synced_execution = remote.sync(execution)
print(synced_execution.inputs)  # print out the inputs
```

You can also wait for the execution after you’ve launched it and access the outputs:

```shell
completed_execution = remote.wait(execution)
print(completed_execution.outputs)  # print out the outputs
```

## Terminating all running executions for a workflow

This example shows how to terminate all running executions in a given workflow name.

```python
import flytekit as fl
from dataclasses import dataclass
import json
from flytekit.configuration import Config
from flytekit.models.core.execution import NodeExecutionPhase

@dataclass
class Execution:
    name: str
    link: str

SOME_LARGE_LIMIT = 5000
PHASE = NodeExecutionPhase.RUNNING
WF_NAME = "your_workflow_name"
EXECUTIONS_TO_IGNORE = ["some_execution_name_to_ignore"]
PROJECT = "your_project"
DOMAIN = "production"
ENDPOINT = "union.example.com"

remote = fl.FlyteRemote(
    config=Config.for_endpoint(endpoint=ENDPOINT),
    default_project=PROJECT,
    default_domain=DOMAIN,
)

executions_of_interest = []

executions = remote.recent_executions(limit=SOME_LARGE_LIMIT)

for e in executions:
    if e.closure.phase == PHASE:
        if e.spec.launch_plan.name == WF_NAME:
            if e.id.name not in EXECUTIONS_TO_IGNORE:
                execution_on_interest = Execution(name=e.id.name, link=f"https://{ENDPOINT}/console/projects/{PROJECT}/domains/{DOMAIN}/executions/{e.id.name}")
                executions_of_interest.append(execution_on_interest)
                remote.terminate(e, cause="Terminated manually via script.")

with open('terminated_executions.json', 'w') as f:
    json.dump([{'name': e.name, 'link': e.link} for e in executions_of_interest], f, indent=2)

print(f"Terminated {len(executions_of_interest)} executions.")
```

## Rerunning all failed executions of a workflow

This example shows how to identify all failed executions from a given workflow since a certain time, and re-run them with the same inputs and a pinned workflow version.

```python
import datetime
import pytz
import flytekit as fl
from flytekit.models.core.execution import NodeExecutionPhase

SOME_LARGE_LIMIT = 5000
WF_NAME = "your_workflow_name"
PROJECT = "your_project"
DOMAIN = "production"
ENDPOINT = "union.example.com"
VERSION = "your_target_workflow_version"

remote = fl.FlyteRemote(
    config=Config.for_endpoint(endpoint=ENDPOINT),
    default_project=PROJECT,
    default_domain=DOMAIN,
)

executions = remote.recent_executions(limit=SOME_LARGE_LIMIT)

failures = [
    NodeExecutionPhase.FAILED,
    NodeExecutionPhase.ABORTED,
    NodeExecutionPhase.FAILING,
]

# time of the last successful execution
date = datetime.datetime(2024, 10, 30, tzinfo=pytz.UTC)

# filter executions by name
filtered = [execution for execution in executions if execution.spec.launch_plan.name == WF_NAME]

# filter executions by phase
failed = [execution for execution in filtered if execution.closure.phase in failures]

# filter executions by time
windowed = [execution for execution in failed if execution.closure.started_at > date]

# get inputs for each execution
inputs = [remote.sync(execution).inputs for execution in windowed]

# get new workflow version entity
workflow = remote.fetch_workflow(name=WF_NAME, version=VERSION)

# execute new workflow for each failed previous execution
[remote.execute(workflow, inputs=X) for X in inputs]
```

## Filtering for executions using a `Filter`

This example shows how to use a `Filter` to only query for the executions you want.

```python
from flytekit.models import filters
import flytekit as fl

WF_NAME = "your_workflow_name"
LP_NAME = "your_launchplan_name"
PROJECT = "your_project"
DOMAIN = "production"
ENDPOINT = "union.example.com"

remote = fl.FlyteRemote.for_endpoint(ENDPOINT)

# Only query executions from your project
project_filter = filters.Filter.from_python_std(f"eq(workflow.name,{WF_NAME})")
project_executions = remote.recent_executions(project=PROJECT, domain=DOMAIN, filters=[project_filter])

# Query for the latest execution that succeeded and was between 8 and 16 minutes
latest_success = remote.recent_executions(
    limit=1,
    filters=[
        filters.Equal("launch_plan.name", LP_NAME),
        filters.Equal("phase", "SUCCEEDED"),
        filters.GreaterThan("duration", 8 * 60),
        filters.LessThan("duration", 16 * 60),
    ],
)
```

## Launch task via FlyteRemote with a new version

```python
import flytekit as fl
from flytekit.remote import FlyteRemote
from flytekit.configuration import Config, SerializationSettings

# FlyteRemote object is the main entrypoint to API
remote = fl.FlyteRemote(
    config=Config.for_endpoint(endpoint="flyte.example.net"),
    default_project="flytesnacks",
    default_domain="development",
)

# Get Task
task = remote.fetch_task(name="workflows.example.generate_normal_df", version="v1")

task = remote.register_task(
    entity=flyte_task,
    serialization_settings=SerializationSettings(image_config=None),
    version="v2",
)

# Run Task
execution = remote.execute(
     task, inputs={"n": 200, "mean": 0.0, "sigma": 1.0}, execution_name="task-execution", wait=True
)

# Or use execution_name_prefix to avoid repeated execution names
execution = remote.execute(
     task, inputs={"n": 200, "mean": 0.0, "sigma": 1.0}, execution_name_prefix="flyte", wait=True
)

# Inspecting execution
# The 'inputs' and 'outputs' correspond to the task execution.
input_keys = execution.inputs.keys()
output_keys = execution.outputs.keys()
```

## Launch workflow via FlyteRemote

Workflows can be executed with `FlyteRemote` because under the hood it fetches and triggers a default launch plan.

```python
import flytekit as fl
from flytekit.configuration import Config

# UnionRemote object is the main entrypoint to API
remote = fl.FlyteRemote(
    config=Config.for_endpoint(endpoint="flyte.example.net"),
    default_project="flytesnacks",
    default_domain="development",
)

# Fetch workflow
workflow = remote.fetch_workflow(name="workflows.example.wf", version="v1")

# Execute
execution = remote.execute(
    workflow, inputs={"mean": 1}, execution_name="workflow-execution", wait=True
)

# Or use execution_name_prefix to avoid repeated execution names
execution = remote.execute(
    workflow, inputs={"mean": 1}, execution_name_prefix="flyte", wait=True
)
```

## Launch launchplan via FlyteRemote

A launch plan can be launched via FlyteRemote programmatically.

```python
import flytekit as fl
from flytekit.configuration import Config

# UnionRemote object is the main entrypoint to API
remote = fl.FlyteRemote(
    config=Config.for_endpoint(endpoint="flyte.example.net"),
    default_project="flytesnacks",
    default_domain="development",
)

# Fetch launch plan
lp = remote.fetch_launch_plan(
    name="workflows.example.wf", version="v1", project="flytesnacks", domain="development"
)

# Execute
execution = remote.execute(
    lp, inputs={"mean": 1}, execution_name="lp-execution", wait=True
)

# Or use execution_name_prefix to avoid repeated execution names
execution = remote.execute(
    lp, inputs={"mean": 1}, execution_name_prefix="flyte", wait=True
)
```

## Inspecting executions

With `FlyteRemote`, you can fetch the inputs and outputs of executions and inspect them.

```python
import flytekit as fl
from flytekit.configuration import Config

# UnionRemote object is the main entrypoint to API
remote = fl.FlyteRemote(
    config=Config.for_endpoint(endpoint="flyte.example.net"),
    default_project="flytesnacks",
    default_domain="development",
)

execution = remote.fetch_execution(
    name="fb22e306a0d91e1c6000", project="flytesnacks", domain="development"
)

input_keys = execution.inputs.keys()
output_keys = execution.outputs.keys()

# The inputs and outputs correspond to the top-level execution or the workflow itself.
# To fetch a specific output, say, a model file:
model_file = execution.outputs["model_file"]
with open(model_file) as f:
    ...

# You can use UnionRemote.sync() to sync the entity object's state with the remote state during the execution run.
synced_execution = remote.sync(execution, sync_nodes=True)
node_keys = synced_execution.node_executions.keys()

# node_executions will fetch all the underlying node executions recursively.
# To fetch output of a specific node execution:
node_execution_output = synced_execution.node_executions["n1"].outputs["model_file"]

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/testing ===

# Testing

The `flytekit` python SDK provides a few utilities for making it easier to test
your tasks and workflows in your test suite. For more details, you can also refer
to the [`flytekit.testing`](https://www.union.ai/docs/v1/flyte/api-reference/flytekit-sdk/flytekit.core.testing/page.md) module in the Reference section.

=== PAGE: https://www.union.ai/docs/v1/flyte/user-guide/development-cycle/testing/mocking-tasks ===

# Mocking tasks

A lot of the tasks that you write you can run locally, but some of them you will not be able to, usually because they are tasks that depend on a third-party only available on the backend. Hive tasks are a common example, as most users will not have access to the service that executes Hive queries from their development environment. However, it's still useful to be able to locally run a workflow that calls such a task. In these instances, flytekit provides a couple of utilities to help navigate this.

For example, this is a generic SQL task (and is by default not hooked up to any datastore nor handled by any plugin), and must be mocked if it is to be used in a unit test like so:

```python
import datetime

import pandas
from flytekit import SQLTask, TaskMetadata, kwtypes, task, workflow
from flytekit.testing import patch, task_mock
from flytekit.types.schema import FlyteSchema

sql = SQLTask(
    "my-query",
    query_template="SELECT * FROM hive.city.fact_airport_sessions WHERE ds = '{{ .Inputs.ds }}' LIMIT 10",
    inputs=kwtypes(ds=datetime.datetime),
    outputs=kwtypes(results=FlyteSchema),
    metadata=TaskMetadata(retries=2),
)

@task
def t1() -> datetime.datetime:
    return datetime.datetime.now()
```

Suppose you have a workflow that uses these two tasks:

```python
@workflow
def my_wf() -> FlyteSchema:
    dt = t1()
    return sql(ds=dt)
```

Without a mock, calling the workflow would typically raise an exception, but with the `task_mock` construct, which returns a `MagicMock` object, we can override the return value.

```python
def test_demonstrate_mock():
    with task_mock(sql) as mock:
        mock.return_value = pandas.DataFrame(data={"x": [1, 2], "y": ["3", "4"]})
        assert (my_wf().open().all() == pandas.DataFrame(data={"x": [1, 2], "y": ["3", "4"]})).all().all()
```

There is another utility as well called `patch` which offers the same functionality, but in the traditional Python patching style, where the first argument is the `MagicMock` object.

```python
@patch(sql)
def test_demonstrate_patch(mock_sql):
    mock_sql.return_value = pandas.DataFrame(data={"x": [1, 2], "y": ["3", "4"]})
    assert (my_wf().open().all() == pandas.DataFrame(data={"x": [1, 2], "y": ["3", "4"]})).all().all()
```