Argo Workflows SDK

导航

已验证详情

这些详情已由PyPI 验证
维护者
Avatar for goern from gravatar.com goern

未验证详情

这些详情尚未由PyPI 验证
项目链接
元数据
  • 许可证: Apache软件许可协议（Apache Software License）
  • 作者: Marek Cermak
分类

项目描述

argo-python-sdk   Release

License   CI  

Argo Workflows的Python SDK

如果您是Argo的新手，我们建议查看纯YAML中的示例。这种语言是描述性的，Argo的示例提供了详尽的说明。

对于经验更丰富的受众，此SDK允许您使用Python编程定义Argo Workflows，然后将其转换为Argo YAML规范。

SDK使用Argo Python客户端存储库中定义的Argo模型。结合两种方法，我们获得了对Argo Workflows的完全底层控制。

入门

Hello World

此示例演示了最基本的功能。通过使用@Workflow类和单个模板（使用@template装饰器）定义一个Workflow。

工作流的入口点定义为entrypoint类属性。

Argo YAMLArgo Python

# @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"]


from argo.workflows.sdk import Workflow
from argo.workflows.sdk import template

from argo.workflows.sdk.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

DAG：任务

本示例演示了通过依赖关系形成菱形结构的任务定义。任务使用@task装饰器定义，并且它们必须返回一个有效的模板。

入口点会自动为Workflow的顶层任务创建为main。

Argo YAMLArgo Python

# @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}}"]


from argo.workflows.sdk import Workflow

from argo.workflows.sdk.tasks import *
from argo.workflows.sdk.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

工件

工件可以像参数一样以三种形式传递：参数、输入和输出，其中参数是默认形式（简单使用@artifact或@parameter）。

即：inputs.artifact(...)

工件和参数是逐个传递的，这意味着对于多个工件（参数），应调用

@inputs.artifact(name="artifact", ...)
@inputs.parameter(name="parameter_a", ...)
@inputs.parameter(...)
def foo(self, artifact: V1alpha1Artifact, prameter_b: V1alpha1Parameter, ...): pass

完整示例

Argo YAMLArgo Python

# @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"]


from argo.workflows.sdk import Workflow

from argo.workflows.sdk.tasks import *
from argo.workflows.sdk.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


更进一步：闭包和作用域

这部分内容相当有趣。到目前为止，我们只是触及了Python实现提供的优势。

如果我们想使用原生Python代码并将其作为工作流中的一个步骤执行，我们有哪些选择？

选项A)是重复使用现有的思维模式，将代码放入字符串中，将其作为V1ScriptTemplate模型的源传递，并用template装饰器包装。以下代码块展示了这一点

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

结果如下

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'

不错，但还没有充分发挥潜力。既然我们已经在写Python，为什么要将代码放入字符串中呢？这就是我们引入闭包的地方。

闭包

闭包的逻辑相当简单。只需将您要执行的功能包装在@closure装饰器的容器中。然后，闭包会处理其余的操作，并返回一个模板（就像@template装饰器一样）。

我们唯一需要关注的是提供一个图像，该图像已安装必要的Python依赖项，并且存在于集群中。

未来计划消除这一步骤，但当前这是不可避免的。

按照之前的示例

class ScriptsPython(Workflow):

    ...

    @closure(
      image="python:alpine3.6"
    )
    def gen_random_int() -> V1alpha1ScriptTemplate:
          import random

          i = random.randint(1, 100)
          print(i)

闭包实现了V1alpha1ScriptTemplate，这意味着您可以传递诸如resources、env等...

此外，确保您导入了正在使用的任何库，上下文不会被保留 --- 闭包表现得像一个staticmethod，并且是从模块作用域中被沙箱化的。

作用域

现在，如果我们有一个函数（或整个脚本）相当大，将其包装在单个Python函数中并不是非常Pythonic，而且很繁琐。这就是我们可以使用作用域的地方。

例如，如果我们想在运行gen_random_int函数之前初始化日志。

    ...

    @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)

注意我们做了三个更改

    @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"):

给定作用域中的每个函数都通过作用域名称进行命名空间化，并注入到闭包中。

即，生成的YAML文件看起来像这样

...
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)

编译过程还将所有导入移到前面，并删除重复项，以便更方便地查看，这样您在查看生成的YAML时就不会感到不适。


有关更多示例，请参阅示例文件夹。


作者

@AICoE，Red Hat

项目详情

已验证详情

这些详情已由PyPI 验证
维护者
Avatar for goern from gravatar.com goern

未验证详情

这些详情尚未由PyPI 验证
项目链接
元数据
  • 许可证: Apache软件许可协议（Apache Software License）
  • 作者: Marek Cermak
分类

版本历史 发布通知 | RSS源

此版本

0.1.0.dev0 预发布

下载文件

下载适用于您平台上的文件。如果您不确定选择哪个，请了解更多关于 安装包 的信息。

源代码发行版

argo-workflows-sdk-0.1.0.dev0.tar.gz (22.5 kB 查看哈希值)

上传时间 源代码

构建发行版

argo_workflows_sdk-0.1.0.dev0-py3-none-any.whl (21.1 kB 查看哈希值)

上传时间 Python 3

支持者