Argo YAML | Argo Python |
---|---|
```yaml # @file: hello-world.yaml apiVersion: argoproj.io/v1alpha1 kind: Workflow metadata: name: hello-world generateName: hello-world- spec: entrypoint: whalesay templates: - name: whalesay container: name: whalesay image: docker/whalesay:latest command: [cowsay] args: ["hello world"] ``` |
```python from argo.workflows.dsl import Workflow from argo.workflows.dsl import template from argo.workflows.dsl.templates import V1Container class HelloWorld(Workflow): entrypoint = "whalesay" @template def whalesay(self) -> V1Container: container = V1Container( image="docker/whalesay:latest", name="whalesay", command=["cowsay"], args=["hello world"] ) return container ``` |
Argo YAML | Argo Python |
---|---|
```yaml # @file: dag-diamond.yaml # The following workflow executes a diamond workflow # # A # / \ # B C # \ / # D apiVersion: argoproj.io/v1alpha1 kind: Workflow metadata: name: dag-diamond generateName: dag-diamond- spec: entrypoint: main templates: - name: main dag: tasks: - name: A template: echo arguments: parameters: [{name: message, value: A}] - name: B dependencies: [A] template: echo arguments: parameters: [{name: message, value: B}] - name: C dependencies: [A] template: echo arguments: parameters: [{name: message, value: C}] - name: D dependencies: [B, C] template: echo arguments: parameters: [{name: message, value: D}] # @task: [A, B, C, D] - name: echo inputs: parameters: - name: message container: name: echo image: alpine:3.7 command: [echo, "{{inputs.parameters.message}}"] ``` |
```python from argo.workflows.dsl import Workflow from argo.workflows.dsl.tasks import * from argo.workflows.dsl.templates import * 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 ``` |
Argo YAML | Argo Python |
---|---|
```yaml # @file: artifacts.yaml apiVersion: argoproj.io/v1alpha1 kind: Workflow metadata: name: artifact-passing generateName: artifact-passing- spec: entrypoint: main templates: - name: main dag: tasks: - name: generate-artifact template: whalesay - name: consume-artifact template: print-message arguments: artifacts: # bind message to the hello-art artifact # generated by the generate-artifact step - name: message from: "{{tasks.generate-artifact.outputs.artifacts.hello-art}}" - name: whalesay container: name: "whalesay" image: docker/whalesay:latest command: [sh, -c] args: ["cowsay hello world | tee /tmp/hello_world.txt"] outputs: artifacts: # generate hello-art artifact from /tmp/hello_world.txt # artifacts can be directories as well as files - name: hello-art path: /tmp/hello_world.txt - name: print-message inputs: artifacts: # unpack the message input artifact # and put it at /tmp/message - name: message path: /tmp/message container: name: "print-message" image: alpine:latest command: [sh, -c] args: ["cat", "/tmp/message"] ``` |
```python from argo.workflows.dsl import Workflow from argo.workflows.dsl.tasks import * from argo.workflows.dsl.templates import * class ArtifactPassing(Workflow): @task def generate_artifact(self) -> V1alpha1Template: return self.whalesay() @task @artifact( name="message", _from="{{tasks.generate-artifact.outputs.artifacts.hello-art}}" ) def consume_artifact(self, message: V1alpha1Artifact) -> V1alpha1Template: return self.print_message(message=message) @template @outputs.artifact(name="hello-art", path="/tmp/hello_world.txt") def whalesay(self) -> V1Container: container = V1Container( name="whalesay", image="docker/whalesay:latest", command=["sh", "-c"], args=["cowsay hello world | tee /tmp/hello_world.txt"] ) return container @template @inputs.artifact(name="message", path="/tmp/message") def print_message(self, message: V1alpha1Artifact) -> V1Container: container = V1Container( name="print-message", image="alpine:latest", command=["sh", "-c"], args=["cat", "/tmp/message"], ) return container ``` |
closure
and scope
import textwrap
class ScriptsPython(Workflow):
...
@template
def gen_random_int(self) -> V1alpha1ScriptTemplate:
source = textwrap.dedent("""\
import random
i = random.randint(1, 100)
print(i)
""")
template = V1alpha1ScriptTemplate(
image="python:alpine3.6",
name="gen-random-int",
command=["python"],
source=source
)
return template
Which results in:
api_version: argoproj.io/v1alpha1
kind: Workflow
metadata:
generate_name: scripts-python-
name: scripts-python
spec:
entrypoint: main
...
templates:
- name: gen-random-int
script:
command:
- python
image: python:alpine3.6
name: gen-random-int
source: 'import random\ni = random.randint(1, 100)\nprint(i)\n'
class ScriptsPython(Workflow):
...
@closure(
image="python:alpine3.6"
)
def gen_random_int() -> V1alpha1ScriptTemplate:
import random
i = random.randint(1, 100)
print(i)
...
@closure(
scope="main",
image="python:alpine3.6"
)
def gen_random_int(main) -> V1alpha1ScriptTemplate:
import random
main.init_logging()
i = random.randint(1, 100)
print(i)
@scope(name="main")
def init_logging(level="DEBUG"):
import logging
logging_level = getattr(logging, level, "INFO")
logging.getLogger("__main__").setLevel(logging_level)
Notice the 3 changes that we've made:
@closure(
scope="main", # <--- provide the closure a scope
image="python:alpine3.6"
)
def gen_random_int(main): # <--- use the scope name
@scope(name="main") # <--- add function to a scope
def init_logging(level="DEBUG"):
...
spec:
...
templates:
- name: gen-random-int
script:
command:
- python
image: python:alpine3.6
name: gen-random-int
source: |-
import logging
import random
class main:
"""Scoped objects injected from scope 'main'."""
@staticmethod
def init_logging(level="DEBUG"):
logging_level = getattr(logging, level, "INFO")
logging.getLogger("__main__").setLevel(logging_level)
main.init_logging()
i = random.randint(1, 100)
print(i)
The compilation also takes all imports to the front and remove duplicates for convenience and more natural look so that you don't feel like poking your eyes when you look at the resulting YAML.
For more examples see the examples folder.
Authors: