### Start Minikube Kubernetes Cluster

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/readme.md

This command initializes and starts the Minikube Kubernetes cluster on your local machine. dflow relies on Kubernetes for orchestrating workflows, making Minikube an essential component for local development and testing.

```Bash
minikube start
```

--------------------------------

### Start Minikube with China Image Mirror

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/readme.md

This command starts the Minikube Kubernetes cluster while utilizing an image mirror located in China. This option is recommended for users within China to significantly speed up the Minikube startup process due to network optimizations.

```Bash
minikube start --image-mirror-country=cn
```

--------------------------------

### Verify Minikube Installation on Windows

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_windowsos.md

This snippet provides the command to check the installed version of Minikube, confirming a successful installation.

```bash
minikube version
```

--------------------------------

### Install Argo Workflows in Kubernetes (China Mirror)

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/readme.md

These commands facilitate the installation of Argo Workflows using a local YAML file, which can be downloaded from a faster mirror for users in China. This approach helps to mitigate long wait times during the installation process by leveraging local resources.

```Bash
wget https://raw.githubusercontent.com/deepmodeling/dflow/master/manifests/quick-start-postgres-3.4.1-deepmodeling.yaml
kubectl apply -n argo -f quick-start-postgres-3.4.1-deepmodeling.yaml
# linux user might need to use `minikube kubectl --`
```

--------------------------------

### Verify Docker Installation on macOS

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_macos.md

This command runs the 'hello-world' Docker image to verify that Docker Desktop is correctly installed and functioning on macOS. It downloads a test image, runs it in a container, prints a confirmation message, and then exits, indicating a successful setup.

```bash
docker run hello-world
```

--------------------------------

### Install Argo Workflows in Kubernetes (Global Access)

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/readme.md

These commands set up the Argo Workflows engine within your Kubernetes cluster, which is the core orchestrator for dflow. The quick-start manifest includes Argo Workflows and other commonly used components for a complete setup.

```Bash
kubectl create ns argo
kubectl apply -n argo -f https://raw.githubusercontent.com/deepmodeling/dflow/master/manifests/quick-start-postgres-3.4.1-deepmodeling.yaml
```

--------------------------------

### Diagnose Minikube Startup Failures with Kubelet Logs

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/readme.md

To diagnose why Minikube fails to start, use these commands to SSH into the Minikube node and inspect the kubelet service logs. This helps identify issues like incorrect Kubernetes image versions being pulled, which often prevent the cluster from initializing correctly.

```bash
minikube ssh
sudo journalctl -xeu kubelet
```

--------------------------------

### Install Minikube Binary to System Path

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

This command moves the downloaded Minikube binary to `/usr/local/bin`, making it executable from any directory in the system's PATH. This completes the Minikube installation process, allowing you to run `minikube` commands globally.

```bash
sudo install minikube /usr/local/bin/minikube
```

--------------------------------

### Add New User and Grant Sudo Privileges for Minikube

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

These commands create a new user account (e.g., 'developer') and then add that user to the 'sudo' group. This is necessary because Minikube typically does not allow starting as the root user, requiring a non-root user with administrative privileges.

```bash
adduser developer 
#add the user. Follow the prompt
usermod -aG sudo developer
```

--------------------------------

### Install dflow Python Package with pip

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/readme.md

This command installs the pydflow library, which is the Python interface for interacting with dflow. It is a fundamental dependency for developing and running dflow workflows.

```Bash
pip install pydflow
```

--------------------------------

### Verify Minikube Installation and Version

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

This command checks the installed Minikube version, confirming that the installation was successful and the Minikube binary is accessible in the system's PATH. A successful output indicates Minikube is ready for use.

```bash
minikube version
```

--------------------------------

### Verify Docker Engine Installation with Hello-World Image

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

This command runs the official `hello-world` Docker image, which downloads a test image and executes it in a container. A successful run, indicated by a printed message, confirms that Docker Engine is correctly installed and operational.

```bash
docker run hello-world
```

--------------------------------

### Switch to Newly Created User Account

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

This command switches the current session to the newly created user account (e.g., 'developer'). It's important to log in as this user before proceeding with Minikube installation and usage to avoid permission issues.

```bash
su - developer
```

--------------------------------

### Access Minio UI via Kubectl Port Forwarding

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/readme.md

This command forwards port 9001 from the Minio deployment to your local machine, enabling access to the Minio web UI. After executing, you can log in at http://127.0.0.1:9001 using the default credentials: admin for username and password for password.

```bash
kubectl -n argo port-forward deployment/minio 9001:9001 --address 0.0.0.0
```

--------------------------------

### Monitor Argo Workflow Pod Status

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/readme.md

This command allows you to check the status of all pods within the 'argo' namespace in your Kubernetes cluster. It is crucial for monitoring the installation and setup progress of Argo Workflows, ensuring all components are running correctly before proceeding.

```Bash
kubectl get pod -n argo
# linux user might need to use `minikube kubectl --`
```

--------------------------------

### Verify Minikube Installation on macOS

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_macos.md

This command checks the installed version of Minikube, confirming that the installation was successful and the binary is correctly configured in the system's PATH. It outputs the Minikube version number, indicating readiness for use.

```bash
minikube version
```

--------------------------------

### Access Argo Workflow UI via Port Forwarding

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/readme.md

This command establishes a port-forward from your local machine to the Argo server deployment within the Kubernetes cluster. It enables you to access the Argo Workflows user interface in your web browser, typically at `http://localhost:2746`, for managing and monitoring workflows.

```Bash
kubectl -n argo port-forward deployment/argo-server 2746:2746 --address 0.0.0.0
```

--------------------------------

### Download Minikube Older Version (1.25.2) via Mirror

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_macos.md

This command downloads a specific older version (1.25.2) of the Minikube binary for Linux AMD64 from a Chinese mirror. This method is recommended for users in China due to potential issues with the up-to-date Minikube image repository, which can lead to start failures.

```bash
curl -o minikube -L https://registry.npmmirror.com/-/binary/minikube/v1.25.2/minikube-linux-amd64
```

--------------------------------

### Access Minio API via Kubectl Port Forwarding

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/readme.md

This command establishes a port-forward from your local machine to the Minio deployment within the Argo namespace, allowing access to the Minio API on port 9000. It's crucial to run this in a separate terminal to keep the connection active for artifact operations like downloading or uploading.

```bash
kubectl -n argo port-forward deployment/minio 9000:9000 --address 0.0.0.0
```

--------------------------------

### Set up Docker Repository: Update APT and Install Dependencies

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

This command updates the APT package index and installs necessary packages like ca-certificates, curl, gnupg, and lsb-release. These packages are required for APT to use a repository over HTTPS, which is essential for securely installing Docker.

```console
sudo apt-get update

sudo apt-get install \
    ca-certificates \
    curl \
    gnupg \
    lsb-release
```

--------------------------------

### Install Minikube Binary to System Path on macOS

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_macos.md

This command moves the downloaded Minikube binary to the '/usr/local/bin' directory, making it accessible from any location in the terminal. It requires superuser privileges ('sudo') to write to this system directory, ensuring Minikube is properly installed and executable.

```bash
sudo install minikube /usr/local/bin/minikube
```

--------------------------------

### Manually Resolve Minikube Incorrect Image Pull Issues

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/readme.md

This solution addresses Minikube startup failures caused by incorrect Kubernetes image versions. By SSHing into the Minikube node, you can manually pull the correct pause image from a specified registry and then retag it to match the expected k8s.gcr.io path, resolving the image mismatch.

```bash
minikube ssh
docker pull registry.aliyuncs.com/google_containers/pause:3.6
docker tag registry.aliyuncs.com/google_containers/pause:3.6 k8s.gcr.io/pause:3.6
exit
```

--------------------------------

### List Available Docker Engine Versions from APT Repository

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

This command queries the APT cache to list all available versions of the docker-ce package in the configured repositories. This is useful for identifying specific versions to install or for troubleshooting compatibility issues.

```console
apt-cache madison docker-ce
```

--------------------------------

### Install Specific Docker Engine Version

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

This command installs a specific version of Docker Engine, Docker CLI, containerd.io, and the Docker Compose plugin. Replace `<VERSION_STRING>` with the desired version obtained from `apt-cache madison docker-ce` to ensure a precise installation.

```console
sudo apt-get install docker-ce=<VERSION_STRING> docker-ce-cli=<VERSION_STRING> containerd.io docker-compose-plugin
```

--------------------------------

### Install Latest Docker Engine, Containerd, and Docker Compose Plugin

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

This command updates the APT package index and installs the latest stable versions of Docker Engine (docker-ce), Docker CLI (docker-ce-cli), containerd.io, and the Docker Compose plugin. This provides a complete and up-to-date Docker environment.

```console
sudo apt-get update
sudo apt-get install docker-ce docker-ce-cli containerd.io docker-compose-plugin
```

--------------------------------

### Add Docker's Official GPG Key for APT Repository

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

This command creates a directory for APT keyrings and downloads Docker's official GPG key. The key is then de-armored and saved to the specified path, which allows APT to verify the authenticity of Docker packages.

```console
sudo mkdir -p /etc/apt/keyrings
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /etc/apt/keyrings/docker.gpg
```

--------------------------------

### Configure Docker APT Repository Source List

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

This command adds the Docker stable repository to your APT sources list. It uses the system's architecture and Ubuntu release name to specify the correct repository, enabling your system to find and install Docker Engine packages.

```console
echo \
  "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.gpg] https://download.docker.com/linux/ubuntu \
  $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null
```

--------------------------------

### Download Specific Minikube Version (1.25.2) from Mirror

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

This command downloads a specific older version (1.25.2) of Minikube for Linux AMD64 from a Chinese mirror. This is recommended for users in China due to potential issues with the official repository for newer versions, ensuring a successful download.

```bash
curl -o minikube -L https://registry.npmmirror.com/-/binary/minikube/v1.25.2/minikube-linux-amd64
```

--------------------------------

### Install dflow Python Package

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This section describes how to install the dflow Python package. Ensure your Python version is not less than 3.6 before proceeding with the installation.

```Shell
pip install pydflow
```

--------------------------------

### Install Cookiecutter and Initialize a DFLOW OP Package

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This snippet demonstrates how to install the `cookiecutter` utility and then use it to initialize a new DFLOW OP (Operation) package from the `dflow-op-cutter` template. This process allows developers to quickly set up a new, reusable operation for their DFLOW workflows, streamlining the development of custom computational steps.

```python
pip install cookiecutter
cookiecutter https://github.com/deepmodeling/dflow-op-cutter.git
```

--------------------------------

### Add User to Docker Group for Minikube Access

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/install_manual_linuxos.md

These commands ensure the 'docker' group exists and then add the current user to it. This allows the user to run Docker commands without `sudo`, which is often required for Minikube to interact with the Docker daemon.

```bash
sudo groupadd docker
sudo usermod -aG docker $USER
- Re-Login or Restart the Server
```

--------------------------------

### Configure Dflow Steps to Continue on Failure in Python

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This example shows how to allow a workflow to proceed even if a specific `Step` fails. Setting `continue_on_failed=True` ensures that the overall workflow execution is not halted by individual step failures, useful for resilient workflows.

```python
Step(..., continue_on_failed=True)
```

--------------------------------

### Redefine Dflow Steps for Output Modification Example

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-reuse.ipynb

Reinitializes the 'Steps' object and redefines the 'addition' and 'loop' steps, similar to the reuse example, but specifically for demonstrating output modification. The 'limit' is set to 6.

```python
steps = Steps(
    name="iter",
    inputs=Inputs(
        parameters={"iter": InputParameter(value=0), "limit": InputParameter(value=6)}
    ),
)
addition = Step(name="add",
                template=PythonOPTemplate(Plus1,
                                        image=f"python:{sys.version_info.major}.{sys.version_info.minor}"),
                parameters={"iter": steps.inputs.parameters["iter"]},
                key="iter-%s" % steps.inputs.parameters["iter"]) # do not modify the key!!!!
loop = Step(
    name="loop",
    template=steps,
    parameters={"iter": addition.outputs.parameters["iter"]},
    when="%
```

--------------------------------

### Submit Slurm Job via Dflow Virtual Node in Kubernetes

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This section demonstrates how to submit a Slurm job through a Dflow step by leveraging virtual nodes in Kubernetes. It requires prior installation and configuration of the wlm-operator to expose Slurm partitions as Kubernetes nodes. The example shows how to define a Dflow step with a SlurmJobTemplate executor, directing it to a specific virtual node.

```bash
kubectl get nodes
NAME                            STATUS   ROLES                  AGE    VERSION
minikube                        Ready    control-plane,master   49d    v1.22.3
slurm-minikube-cpu              Ready    agent                  131m   v1.13.1-vk-N/A
slurm-minikube-dplc-ai-v100x8   Ready    agent                  131m   v1.13.1-vk-N/A
slurm-minikube-v100             Ready    agent                  131m   v1.13.1-vk-N/A
```

```python
step = Step(
    ...
    executor=SlurmJobTemplate(
        header="#!/bin/sh\n#SBATCH --nodes=1",
        node_selector={"kubernetes.io/hostname": "slurm-minikube-v100"}
    )
)
```

--------------------------------

### Python dflow OP: Write File and Get Length

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-python.ipynb

This `WriteFile` OP class writes a given string message to a file named 'msg.txt' and outputs the file as an artifact, along with the length of the input message as an integer. It defines input and output signatures using `OPIOSign` for type validation.

```python
class WriteFile(OP):
    def __init__(self):
        pass

    @classmethod
    def get_input_sign(cls):
        return OPIOSign({
            "msg": str,
        })

    @classmethod
    def get_output_sign(cls):
        return OPIOSign({
            "out_art": Artifact(Path),
            "length": int,
        })

    @OP.exec_sign_check
    def execute(
            self,
            op_in: OPIO,
    ) -> OPIO:
        with open("msg.txt","w") as f:
            f.write(op_in["msg"])
        
        op_out = OPIO({
            "out_art": Path("msg.txt"),
            "length": len(op_in["msg"]),
        })
        return op_out
```

--------------------------------

### Dflow Component Naming Conventions and Restrictions

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This section details the specific naming restrictions and conventions for various Dflow components. It differentiates between static and dynamic names and provides examples for Workflow/OP template names, Step/Task names, Parameter/Artifact names, and Key names, ensuring compliance with Dflow's internal requirements.

```APIDOC
{
  "component_naming_rules": [
    {
      "name": "Workflow/OP template name",
      "type": "Static",
      "restrictions": "Lowercase RFC 1123 subdomain (must consist of lower case alphanumeric characters, '-' or '.', and must start and end with an alphanumeric character), not more than 57 characters long",
      "example": "my-name"
    },
    {
      "name": "Step/Task name",
      "type": "Static",
      "restrictions": "Must consist of alpha-numeric characters or '-', and must start with an alpha-numeric character",
      "example": "My-name1-2, 123-NAME"
    },
    {
      "name": "Parameter/Artifact name",
      "type": "Static",
      "restrictions": "Must consist of alpha-numeric characters, '_' or '-'",
      "example": "my_param_1, MY-PARAM-1"
    },
    {
      "name": "Key name",
      "type": "Dynamic",
      "restrictions": "Lowercase RFC 1123 subdomain (must consist of lower case alphanumeric characters, '-' or '.', and must start and end with an alphanumeric character)",
      "example": "my-name"
    }
  ]
}
```

--------------------------------

### Generate Parallel Dflow Steps with `with_sequence` in Python

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This example shows how to use `with_sequence` to generate parallel steps based on a numeric sequence. Combined with `argo_sequence`, it's ideal for creating iterative or indexed parallel tasks, where each step processes a value from the sequence.

```python
step = Step(
    ...
    parameters={"i": "{{item}}"},
    with_sequence=argo_sequence(argo_len(steps.inputs.parameters["msg_list"]))
)
```

--------------------------------

### Execute Dflow Workflow with DispatcherExecutor on Remote Cluster

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-slurm.ipynb

This snippet demonstrates how to set up and use Dflow's `DispatcherExecutor` to run a workflow step on a remote cluster. It configures the executor with host, username, and password for SSH authentication. The example creates a `Step` using the `Duplicate` operator, uploads an artifact, and submits the workflow, showcasing remote execution capabilities.

```python
dispatcher_executor = DispatcherExecutor(
    host="your-cluster-address",
    username="your-login-username",
    password="your-login-password",
)

with open("foo.txt", "w") as f:
    f.write("Hello world!")

step = Step(
    "duplicate",
    PythonOPTemplate(Duplicate),
    parameters={"msg": "Hello", "num": 2},
    artifacts={"foo": upload_artifact("foo.txt")},
    executor=dispatcher_executor,
)

wf = Workflow(name="slurm")
wf.add(step)
wf.submit()
```

--------------------------------

### dflow OP: Write File and Get Length (Python)

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-function.ipynb

This dflow OP function, `write_file`, writes a given string message to a file named 'msg.txt'. It also calculates and returns the length of the input message. It takes a string `msg` as input and outputs a file artifact `out_art` and an integer `length`.

```python
@OP.function(image=f"python:{sys.version_info.major}.{sys.version_info.minor}")
def write_file(msg: str) -> {"out_art": Artifact(Path), "length": int}:
    with open("msg.txt","w") as f:
        f.write(msg)
    
    return {
        "out_art": Path("msg.txt"),
        "length": len(msg)
    }
```

--------------------------------

### Assign Conditional Output Parameter in Dflow Steps/DAG in Python

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This example demonstrates how to assign an output parameter conditionally within a `steps` or `dag` template. Using `if_expression`, the output value is determined based on a boolean condition, allowing for flexible output generation.

```python
steps.outputs.parameters["msg"].value_from_expression = if_expression(
    _if=par1 < par2,
    _then=par3,
    _else=par4
)
```

--------------------------------

### Assign Output Parameter from Nested Step in Dflow Steps

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This example illustrates how to propagate an output parameter from a contained `step` to the parent `steps` template. This mechanism is crucial for managing data flow and ensuring that results from inner operations are accessible at the higher workflow level.

```python
steps.outputs.parameters["msg"].value_from_parameter = step.outputs.parameters["msg"]
```

--------------------------------

### Add `addition` and `loop` Steps to `steps` Template

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-recurse.ipynb

These lines add the previously defined 'addition' and 'loop' steps to the `steps` template. By adding the 'loop' step (which uses `steps` as its template) back into `steps`, a self-referential, recursive workflow is established. This setup allows the workflow to iterate until the specified 'when' condition in the 'loop' step is no longer met.

```python
steps.add(addition)
steps.add(loop)
```

--------------------------------

### dflow Hello OP API Reference

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-slices.ipynb

API documentation for the `Hello` dflow Operation, detailing its input parameters (`filename`) and output artifacts (`out_art`).

```APIDOC
class Hello(OP):
    # Input:
    #   filename: str (name of the file to generate)
    # Output:
    #   out_art: Artifact(str) (path to the generated file)
    # Methods:
    #   get_input_sign(): Returns OPIOSign for 'filename'
    #   get_output_sign(): Returns OPIOSign for 'out_art'
    #   execute(op_in: OPIO) -> OPIO: Writes 'Hello' * file_num to filename, returns filename as out_art
```

--------------------------------

### Import dflow Python Workflow Packages

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-python.ipynb

This snippet imports necessary classes from the `dflow` library, including `Step`, `Workflow`, `OP`, `OPIO`, `OPIOSign`, and `PythonOPTemplate`, along with `Path` from `pathlib`, to construct and manage dflow workflows.

```python
from pathlib import Path

from dflow import Step, Workflow
from dflow.python import OP, OPIO, Artifact, OPIOSign, PythonOPTemplate
```

--------------------------------

### Define and Submit dflow Python Workflow Steps

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-python.ipynb

This snippet defines two `dflow` steps: `step0` uses `WriteFile` to write 'HelloWorld!' and `step1` uses `Duplicate` to process outputs from `step0`. It then constructs a `Workflow` named 'python', adds both steps, and submits the workflow to Argo.

```python
import sys
step0 = Step(
    name="step0",
    template=PythonOPTemplate(WriteFile, image=f"python:{sys.version_info.major}.{sys.version_info.minor}"),
    parameters={"msg": "HelloWorld!"},
)

step1 = Step(
    name="step1",
    template=PythonOPTemplate(Duplicate, image=f"python:{sys.version_info.major}.{sys.version_info.minor}"),
    parameters={"in_num": step0.outputs.parameters["length"]},
    artifacts={"in_art": step0.outputs.artifacts["out_art"]},
)
wf = Workflow(name="python")
wf.add(step0)
wf.add(step1)
wf.submit();
```

--------------------------------

### Define Class-based Python OP Template with dflow

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This Python code demonstrates how to define a class-based OP (`SimpleExample`) using `dflow.python.OP`. It illustrates the use of `get_input_sign` and `get_output_sign` to declare input/output parameters (string) and artifacts (Path), and the `execute` method for the operation logic, including file copying and parameter handling. Type checking is enforced via `@OP.exec_sign_check`.

```python
from dflow.python import OP, OPIO, OPIOSign, Artifact
from pathlib import Path
import shutil


class SimpleExample(OP):
    def __init__(self):
        pass

    @classmethod
    def get_input_sign(cls):
        return OPIOSign(
            {
                "msg": str,
                "foo": Artifact(Path),
            }
        )

    @classmethod
    def get_output_sign(cls):
        return OPIOSign(
            {
                "msg": str,
                "bar": Artifact(Path),
            }
        )

    @OP.exec_sign_check
    def execute(
        self,
        op_in: OPIO,
    ) -> OPIO:
        shutil.copy(op_in["foo"], "bar.txt")
        out_msg = op_in["msg"]
        op_out = OPIO(
            {
                "msg": out_msg,
                "bar": Path("bar.txt"),
            }
        )
        return op_out
```

--------------------------------

### Import dflow Modules for Workflow Definition

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-slices.ipynb

Imports necessary classes from `dflow` and `dflow.python` for defining steps, workflows, and Python operations, including `OP`, `OPIO`, `Artifact`, `OPIOSign`, `PythonOPTemplate`, and `Slices`.

```python
from typing import List

from dflow import Step, Workflow
from dflow.python import OP, OPIO, Artifact, OPIOSign, PythonOPTemplate, Slices
```

--------------------------------

### Create and Submit `dflow` Recursive Workflow

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-recurse.ipynb

This final snippet creates a `dflow` Workflow instance named 'recurse', using the configured `steps` template as its core. Once the workflow is defined, the `submit()` method is called to initiate its execution. This action deploys and runs the recursive iteration process on the `dflow` platform.

```python
wf = Workflow("recurse", steps=steps)
wf.submit();
```

--------------------------------

### Instantiate Python OP Template from Class or Function

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This Python code snippet shows how to create an instance of `PythonOPTemplate` from a previously defined class-based or function-based OP. It requires specifying the OP definition (e.g., `SimpleExample`) and the container image where the OP will execute. The `pydflow` package is automatically uploaded to the container.

```python
from dflow.python import PythonOPTemplate

simple_example_templ = PythonOPTemplate(SimpleExample, image="python:3.8")
```

--------------------------------

### Execute dflow Workflow with Chained OPs (Python)

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-function.ipynb

This snippet demonstrates how to define and execute a dflow workflow by chaining the previously defined `write_file` and `duplicate` OPs. It initializes a workflow, passes outputs from `write_file` as inputs to `duplicate`, and submits the workflow for execution.

```python
with Workflow(name="python") as wf:
    out = write_file(msg="HelloWorld!")
    duplicate(in_num=out["length"], in_art=out["out_art"])
```

--------------------------------

### Define dflow Hello OP for File Writing

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-slices.ipynb

Defines the `Hello` dflow Operation (`OP`) responsible for writing 'Hello' multiple times to a specified file. It takes a `filename` as input and outputs the path to the generated artifact.

```python
class Hello(OP):
    def __init__(self):
        pass

    @classmethod
    def get_input_sign(cls):
        return OPIOSign({
            'filename': str
        })

    @classmethod
    def get_output_sign(cls):
        return OPIOSign({
            'out_art': Artifact(str)
        })

    @OP.exec_sign_check
    def execute(
            self,
            op_in: OPIO,
    ) -> OPIO:
        file_num = int(op_in["filename"].split('.')[0][1:])
        with open(op_in["filename"], "w") as f:
            f.write("Hello" * file_num)
        op_out = OPIO({
            'out_art': op_in["filename"]
        })
        return op_out
```

--------------------------------

### Implement dflow Workflow with Parallel Slices

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-slices.ipynb

Constructs a dflow `Workflow` that utilizes the `Hello` and `Check` OPs. It demonstrates parallel execution using `Slices` to write to multiple files concurrently, passing the outputs of `Hello` to `Check`.

```python
import sys
wf = Workflow("slices")
hello = Step("hello",
            PythonOPTemplate(Hello, image=f"python:{sys.version_info.major}.{sys.version_info.minor}",
                            slices=Slices("{{item}}",
                                        input_parameter=["filename"],
                                        output_artifact=["out_art"]
                                        )
                            ),
            parameters={"filename": [f"f{x}.txt" for x in range(3)]},
            with_param=range(3))
check = Step("check",
            PythonOPTemplate(Check, image=f"python:{sys.version_info.major}.{sys.version_info.minor}"),
            artifacts={"files": hello.outputs.artifacts["out_art"]},
            )
wf.add(hello)
wf.add(check)
wf.submit();
```

--------------------------------

### Import dflow Components for Conditional Workflows

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-conditional.ipynb

This snippet imports necessary classes and functions from the `dflow` library, including `Workflow`, `Step`, `Steps`, `OutputParameter`, `OutputArtifact`, `Outputs`, and `if_expression`, which are essential for defining and managing conditional workflows.

```python
import random

from dflow import (OutputArtifact, OutputParameter, Outputs, Step, Steps,
                   Workflow, if_expression)
from dflow.python import OP, OPIO, Artifact, OPIOSign, PythonOPTemplate
```

--------------------------------

### Import dflow and Python Packages

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-function.ipynb

This snippet imports necessary modules from the `sys`, `pathlib`, and `dflow` libraries. These imports are essential for defining and running dflow workflows and handling file paths and artifacts.

```python
import sys
from pathlib import Path

from dflow import Workflow
from dflow.python import OP, Artifact
```

--------------------------------

### Import `dflow` Components for Workflow Definition

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-recurse.ipynb

This snippet imports essential classes and functions from the `dflow` library. These components are necessary for defining input parameters, steps, step collections, Python operators, and the overall workflow structure. It sets up the environment for building a `dflow` pipeline.

```python
from dflow import (
    InputParameter,
    Inputs,
    Step,
    Steps,
    Workflow,
)
from dflow.python import OP, OPIO, OPIOSign, PythonOPTemplate
```

--------------------------------

### Define Function-based Python OP Template with dflow

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This Python code demonstrates a more concise way to define an OP using a function (`SimpleExample`) decorated with `@OP.function`. Input and output structures are declared using type annotations directly in the function signature and return type hint. The function body defines the execution logic, similar to the class-based approach, handling artifact copying and parameter passing.

```python
from dflow.python import OP, Artifact
from pathlib import Path
import shutil

@OP.function
def SimpleExample(
		msg: str,
		foo: Artifact(Path),
) -> {"msg": str, "bar": Artifact(Path)}:
    shutil.copy(foo, "bar.txt")
    out_msg = msg
    return {"msg": out_msg, "bar": Path("bar.txt")}
```

--------------------------------

### Upload Artifacts in dflow

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

Dflow offers tools for uploading files to the artifact repository and downloading files from it. Users can upload a file/directory, a list of files/directories, or a dictionary of files/directories to obtain an artifact object, which can then be used as an argument for a step. Global artifact repository settings can be configured via `dflow.s3_config`.

```Python
artifact = upload_artifact([path1, path2])
step = Step(
    artifacts={"foo": artifact}
)
```

--------------------------------

### APIDOC: dflow Python Duplicate OP Class

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-python.ipynb

API documentation for the `Duplicate` dflow OP, designed to duplicate content from an input file and multiply an input number. It details the class structure, input parameters/artifacts, and output artifacts/parameters.

```APIDOC
class Duplicate(OP):
    __init__():
        No parameters.

    get_input_sign():
        Returns: OPIOSign
        Parameters:
            in_art (Artifact(Path)): The input file artifact whose content will be duplicated.
            in_num (int): The input integer number to be multiplied.

    get_output_sign():
        Returns: OPIOSign
        Parameters:
            out_art (Artifact(Path)): The path to the new file containing the duplicated content.
            out_num (int): The input number multiplied by two.

    execute(op_in: OPIO) -> OPIO:
        Description: Reads content from 'in_art', writes it duplicated to 'bar.txt', and multiplies 'in_num' by 2.
        Parameters:
            op_in (OPIO): Input object containing 'in_art' and 'in_num'.
        Returns: OPIO
            out_art (Path): Path object for 'bar.txt'.
            out_num (int): Integer representing 'in_num' multiplied by 2.
```

--------------------------------

### Python dflow OP: Duplicate File Content and Multiply Number

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-python.ipynb

This `Duplicate` OP class reads content from an input artifact file, duplicates it twice, and writes it to a new file named 'bar.txt'. It also takes an input integer and multiplies it by two, returning both the new file artifact and the multiplied number.

```python
class Duplicate(OP):
    def __init__(self):
        pass

    @classmethod
    def get_input_sign(cls):
        return OPIOSign({
            "in_art": Artifact(Path),
            "in_num": int,
        })

    @classmethod
    def get_output_sign(cls):
        return OPIOSign({
            "out_art": Artifact(Path),
            "out_num": int,
        })

    @OP.exec_sign_check
    def execute(
            self,
            op_in: OPIO,
    ) -> OPIO:
        with open(op_in["in_art"], "r") as f:
            content = f.read()
        with open("bar.txt", "w") as f:
            f.write(content * 2)
        
        op_out = OPIO({
            "out_art": Path("bar.txt"),
            "out_num": op_in["in_num"] * 2,
        })
        return op_out
```

--------------------------------

### APIDOC: dflow Python WriteFile OP Class

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-python.ipynb

API documentation for the `WriteFile` dflow OP, designed to write a string to a file and return its length. It details the class structure, input parameters, and output artifacts/parameters.

```APIDOC
class WriteFile(OP):
    __init__():
        No parameters.

    get_input_sign():
        Returns: OPIOSign
        Parameters:
            msg (str): The input message string to be written to the file.

    get_output_sign():
        Returns: OPIOSign
        Parameters:
            out_art (Artifact(Path)): The path to the file containing the input message.
            length (int): The length of the input message string.

    execute(op_in: OPIO) -> OPIO:
        Description: Writes the 'msg' to 'msg.txt' and returns the file path and message length.
        Parameters:
            op_in (OPIO): Input object containing 'msg'.
        Returns: OPIO
            out_art (Path): Path object for 'msg.txt'.
            length (int): Integer representing the length of the input message.
```

--------------------------------

### Import Dflow Components for Workflow Definition

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-reuse.ipynb

Imports necessary classes from 'dflow' and 'dflow.python' to define workflow steps, inputs, and operators. These imports are foundational for constructing Dflow workflows.

```python
from dflow import InputParameter, Inputs, Step, Steps, Workflow
from dflow.python import OP, OPIO, OPIOSign, PythonOPTemplate
```

--------------------------------

### Define a Step in dflow

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

The `Step` component is central for establishing data flow rules. It's created by instantiating an OP template, which requires specifying values for all input parameters and sources for all input artifacts. Inputs can be static at submission time or dynamically from outputs of another step.

```Python
from dflow import Step

simple_example_step = Step(
    name="step0",
    template=simple_example_templ,
    parameters={"msg": "HelloWorld!"},
    artifacts={"inp_art": foo},
)
```

--------------------------------

### dflow Check OP API Reference

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-slices.ipynb

API documentation for the `Check` dflow Operation, detailing its input artifacts (`files`) and lack of explicit output.

```APIDOC
class Check(OP):
    # Input:
    #   files: Artifact(List[str]) (list of file paths to check)
    # Output:
    #   None
    # Methods:
    #   get_input_sign(): Returns OPIOSign for 'files'
    #   get_output_sign(): Returns empty OPIOSign
    #   execute(op_in: OPIO) -> OPIO: Prints the input 'files'
```

--------------------------------

### Initialize Dflow Steps for Iterative Workflow

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-reuse.ipynb

Initializes a 'Steps' object named "iter" with initial input parameters 'iter' set to 0 and 'limit' set to 3. This sets up the container for the recursive workflow logic.

```python
steps = Steps(
    name="iter",
    inputs=Inputs(
        parameters={"iter": InputParameter(value=0), "limit": InputParameter(value=3)}
    ),
)
```

--------------------------------

### Initialize `Steps` Template for Iterative Workflow

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-recurse.ipynb

This snippet initializes a `Steps` template named 'iter', which acts as a container for a sequence of steps and can be used to construct more complex workflows, including recursive loops. It defines initial input parameters 'iter' (defaulting to 0) and 'limit' (defaulting to 3), which control the loop's behavior. `Steps` is a powerful `OPTemplate` subclass for building structured workflows.

```python
steps = Steps(
    name="iter",
    inputs=Inputs(
        parameters={"iter": InputParameter(value=0), "limit": InputParameter(value=3)}
    ),
)
```

--------------------------------

### Create and Submit a dflow Workflow

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

The `Workflow` component connects steps together to build a data flow. Steps can be added sequentially for serial execution or as a list for concurrent execution. Workflows are submitted using the `submit()` method to initiate their execution.

```Python
from dflow import Workflow

wf = Workflow(name="hello-world")
wf.add(simple_example_step)
```

```Python
wf.submit()
```

--------------------------------

### dflow OP: Duplicate File Content and Multiply Number (Python)

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-function.ipynb

This dflow OP function, `duplicate`, reads content from an input file artifact, duplicates it, and writes the duplicated content to a new file named 'bar.txt'. It also takes an integer input and multiplies it by two. It takes a file artifact `in_art` and an integer `in_num` as inputs, and outputs a file artifact `out_art` and an integer `out_num`.

```python
@OP.function(image=f"python:{sys.version_info.major}.{sys.version_info.minor}")
def duplicate(in_art: Artifact(Path), in_num: int) -> {"out_art": Artifact(Path), "out_num": int}:
    with open(in_art, "r") as f:
        content = f.read()
    with open("bar.txt", "w") as f:
        f.write(content * 2)

    return {
        "out_art": Path("bar.txt"),
        "out_num": in_num * 2
    }
```

--------------------------------

### Initialize dflow Steps with Conditional Outputs

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-conditional.ipynb

This snippet initializes a `Steps` object named 'conditional-steps' with defined output parameters and artifacts that will be conditionally set. It then creates a `Step` named 'random' using the `Random` Python OP, which is subsequently added to the `Steps` to generate the initial conditional values.

```python
steps = Steps(
    name="conditional-steps", 
    outputs=Outputs(
        parameters={"msg": OutputParameter()},
        artifacts={"res": OutputArtifact()}
    ),
)

import sys
random_step = Step(
    name="random",
    template=PythonOPTemplate(Random, image=f"python:{sys.version_info.major}.{sys.version_info.minor}")
)

steps.add(random_step)
```

--------------------------------

### Configure Dflow Debug Mode for Local Workflow Execution

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This section explains how to activate Dflow's debug mode, allowing workflows to run locally without requiring a Kubernetes environment. It highlights the importance of configuring local dependencies for OPs and describes the default working directory structure generated for workflows and steps in debug mode, including input/output artifacts and parameters.

```python
from dflow import config
config["mode"] = "debug"
```

--------------------------------

### Submit Initial Dflow Recursive Workflow

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-reuse.ipynb

Adds the 'addition' and 'loop' steps to the 'steps' object and creates a 'Workflow' named "recurse". This workflow is then submitted, executing the recursive addition process until 'iter' reaches the initial 'limit' of 3.

```python
steps.add(addition)
steps.add(loop)

wf = Workflow("recurse", steps=steps)
wf.submit();
```

--------------------------------

### Define dflow Check OP for File Verification

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-slices.ipynb

Defines the `Check` dflow Operation (`OP`) that verifies the files written by the `Hello` OP. It takes a list of file paths as an artifact input and prints them to the console.

```python
class Check(OP):
    def __init__(self):
        pass

    @classmethod
    def get_input_sign(cls):
        return OPIOSign({
            'files': Artifact(List[str])
        })

    @classmethod
    def get_output_sign(cls):
        return OPIOSign()

    @OP.exec_sign_check
    def execute(
            self,
            op_in: OPIO,
    ) -> OPIO:
        print(op_in["files"])
        return OPIO()
```

--------------------------------

### Submit dflow Workflow with Conditional Logic

Source: https://github.com/dptech-corp/dflow/blob/master/tutorials/dflow-conditional.ipynb

This snippet creates a `Workflow` named 'conditional' using the previously defined `Steps` object, which incorporates the `if_expression` logic. It then submits the workflow for execution, demonstrating the final step to run a dflow workflow that leverages conditional output assignments.

```python
wf = Workflow(name="conditional", steps=steps)
wf.submit();
```

--------------------------------

### APIDOC: dflow.python.OP Class and Related Types

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

This section provides API documentation for the `dflow.python.OP` class and related components used in defining dflow operations. It outlines the structure for class-based and function-based OPs, including methods for declaring input/output signs, the execution method, and relevant decorators. It also details the types for parameters and artifacts.

```APIDOC
Class: dflow.python.OP
  Description: Base class for defining dflow operations.

  Methods:
    __init__(self)
      Description: Constructor for the OP class.

    @classmethod
    get_input_sign(cls) -> dflow.python.OPIOSign
      Description: Declares the input structure of an OP.
      Returns: An OPIOSign object (dictionary mapping parameter/artifact name to type).
      Artifact Types: Artifact(Path), Artifact(List[Path]), Artifact(Dict[str, Path]), Artifact(dflow.python.NestedDict[Path])

    @classmethod
    get_output_sign(cls) -> dflow.python.OPIOSign
      Description: Declares the output structure of an OP.
      Returns: An OPIOSign object (dictionary mapping parameter/artifact name to type).
      Artifact Types: Artifact(Path), Artifact(List[Path]), Artifact(Dict[str, Path]), Artifact(dflow.python.NestedDict[Path])

    @dflow.python.OP.exec_sign_check
    execute(self, op_in: dflow.python.OPIO) -> dflow.python.OPIO
      Description: Defines the execution logic of the OP.
      Parameters:
        op_in: dflow.python.OPIO - Input parameters and artifacts.
      Returns: dflow.python.OPIO - Output parameters and artifacts.

  Decorators:
    @dflow.python.OP.function
      Description: Decorator to define a function-based OP. Input/output are declared via type annotations.

Class: dflow.python.PythonOPTemplate
  Description: Creates an OP template from a Python class or function.

  Constructor:
    __init__(self, op_class_or_func, image: str, **kwargs)
      Parameters:
        op_class_or_func: class | function - The Python class or function defining the OP.
        image: str - The container image where the OP script runs.
        **kwargs: any - Optional arguments for the template.
```

--------------------------------

### dflow Workflow Management API

Source: https://github.com/dptech-corp/dflow/blob/master/README.md

APIs for managing dflow workflows after submission or retrieval. Includes methods for querying status, steps, and terminating workflows. Provides details on how to access step outputs like parameters and artifacts.

```APIDOC
{
  "class": "Workflow",
  "methods": [
    {
      "name": "id",
      "type": "property",
      "description": "Workflow ID in Argo.",
      "return_type": "string"
    },
    {
      "name": "query_status()",
      "description": "Queries the real-time status of the workflow.",
      "return_type": "string",
      "return_values": ["Pending", "Running", "Succeeded", "..."]
    },
    {
      "name": "query_step(name=None, key=None, phase=None, id=None, type=None)",
      "description": "Queries steps by various criteria.",
      "parameters": [
        {"name": "name", "type": "string", "description": "Name of the step (supports regex)."},
        {"name": "key", "type": "string", "description": "Key of the step."},
        {"name": "phase", "type": "string", "description": "Phase of the step (e.g., 'Pending', 'Running', 'Succeeded')."},
        {"name": "id", "type": "string", "description": "ID of the step."},
        {"name": "type", "type": "string", "description": "Type of the step."}
      ],
      "return_type": "list",
      "return_description": "A list of Argo step objects.",
      "nested_properties": [
        {"name": "step.phase", "type": "string", "description": "Phase of a step (e.g., 'Pending', 'Running', 'Succeeded')."},
        {"name": "step.outputs.parameters", "type": "dictionary", "description": "A dictionary mapping parameter names to parameter objects. Get the value by `step.outputs.artifacts[\"foo\"].value`."},
        {"name": "step.outputs.artifacts", "type": "dictionary", "description": "A dictionary mapping artifact names to artifact objects. Download an artifact by `download_artifact(step.outputs.artifacts[\"bar\"])`."}
      ]
    },
    {
      "name": "terminate()",
      "description": "Terminates the workflow.",
      "return_type": "None"
    }
  ]
}
```