submit failed, prompt "argo.workflows.client.rest.ApiException: (400)" about argo-python-dsl HOT 6 OPEN

I have the same error using Minikube.

from argo-python-dsl.

This small hack fixed it for me:

# Current hack to make it work: should be fixed soon
manifest = wk.to_dict()
manifest["apiVersion"] = "argoproj.io/v1alpha1"
manifest["kind"] = "Workflow"

from argo-python-dsl.

This small hack fixed it for me:

# Current hack to make it work: should be fixed soon
manifest = wk.to_dict()
manifest["apiVersion"] = "argoproj.io/v1alpha1"
manifest["kind"] = "Workflow"

could you show me full code? i can't get how to submit by python dsl

from argo-python-dsl.

Here it is:

import secrets

from argo.workflows.client import V1alpha1Api
from argo.workflows.config import load_kube_config

from argo.workflows.dsl import Workflow
from argo.workflows.dsl.tasks import task, dependencies
from argo.workflows.dsl.templates import parameter
from argo.workflows.dsl.templates import inputs
from argo.workflows.dsl.templates import template
from argo.workflows.dsl.templates import V1alpha1Parameter
from argo.workflows.dsl.templates import V1alpha1Template
from argo.workflows.dsl.templates import V1Container


# Define your workflow
class DagDiamond(Workflow):

    @task
    @parameter(name="message", value="A")
    def A(self, message: V1alpha1Parameter) -> V1alpha1Template:
        return self.echo(message=message)

    @task
    @parameter(name="message", value="B")
    @dependencies(["A"])
    def B(self, message: V1alpha1Parameter) -> V1alpha1Template:
        return self.echo(message=message)

    @task
    @parameter(name="message", value="C")
    @dependencies(["A"])
    def C(self, message: V1alpha1Parameter) -> V1alpha1Template:
        return self.echo(message=message)

    @task
    @parameter(name="message", value="D")
    @dependencies(["B", "C"])
    def D(self, message: V1alpha1Parameter) -> V1alpha1Template:
        return self.echo(message=message)

    @template
    @inputs.parameter(name="message")
    def echo(self, message: V1alpha1Parameter) -> V1Container:
        container = V1Container(
            image="alpine:3.7",
            name="echo",
            command=["echo", "{{inputs.parameters.message}}"],
        )

        return container
    
# Create a workflow
wk = DagDiamond()

# Load the API to communicate with the cluster
load_kube_config()
v1alpha1 = V1alpha1Api()

# Current hack to make it work: should be fixed soon
manifest = wk.to_dict()
manifest["apiVersion"] = "argoproj.io/v1alpha1"
manifest["kind"] = "Workflow"
manifest["metadata"]["name"] += secrets.token_hex(5)
manifest["spec"]["serviceAccountName"] = "argo"

# Submit the workflow to the cluster
sub_wk = v1alpha1.create_namespaced_workflow(namespace="argo", body=manifest)

# List workflows (executed or not)
wfs = v1alpha1.list_namespaced_workflows(namespace="argo")
print(f"{len(wfs.items)} workflows on the cluster.")

from argo-python-dsl.

Here it is:

import secrets

from argo.workflows.client import V1alpha1Api
from argo.workflows.config import load_kube_config

from argo.workflows.dsl import Workflow
from argo.workflows.dsl.tasks import task, dependencies
from argo.workflows.dsl.templates import parameter
from argo.workflows.dsl.templates import inputs
from argo.workflows.dsl.templates import template
from argo.workflows.dsl.templates import V1alpha1Parameter
from argo.workflows.dsl.templates import V1alpha1Template
from argo.workflows.dsl.templates import V1Container


# Define your workflow
class DagDiamond(Workflow):

    @task
    @parameter(name="message", value="A")
    def A(self, message: V1alpha1Parameter) -> V1alpha1Template:
        return self.echo(message=message)

    @task
    @parameter(name="message", value="B")
    @dependencies(["A"])
    def B(self, message: V1alpha1Parameter) -> V1alpha1Template:
        return self.echo(message=message)

    @task
    @parameter(name="message", value="C")
    @dependencies(["A"])
    def C(self, message: V1alpha1Parameter) -> V1alpha1Template:
        return self.echo(message=message)

    @task
    @parameter(name="message", value="D")
    @dependencies(["B", "C"])
    def D(self, message: V1alpha1Parameter) -> V1alpha1Template:
        return self.echo(message=message)

    @template
    @inputs.parameter(name="message")
    def echo(self, message: V1alpha1Parameter) -> V1Container:
        container = V1Container(
            image="alpine:3.7",
            name="echo",
            command=["echo", "{{inputs.parameters.message}}"],
        )

        return container
    
# Create a workflow
wk = DagDiamond()

# Load the API to communicate with the cluster
load_kube_config()
v1alpha1 = V1alpha1Api()

# Current hack to make it work: should be fixed soon
manifest = wk.to_dict()
manifest["apiVersion"] = "argoproj.io/v1alpha1"
manifest["kind"] = "Workflow"
manifest["metadata"]["name"] += secrets.token_hex(5)
manifest["spec"]["serviceAccountName"] = "argo"

# Submit the workflow to the cluster
sub_wk = v1alpha1.create_namespaced_workflow(namespace="argo", body=manifest)

# List workflows (executed or not)
wfs = v1alpha1.list_namespaced_workflows(namespace="argo")
print(f"{len(wfs.items)} workflows on the cluster.")

thanks very much, i got sucess also

from argo-python-dsl.

should probably be solved here https://github.com/argoproj-labs/argo-python-dsl

Also pypi argo-python-dsl now points to the community maintained release

from argo-python-dsl.