python - Job Scheduler - 用于编写作业定义的 YAML？

Question

在我们遗留的作业调度软件(建立在 crontab 之上)中，我们使用 apache 配置格式 ( parser ) 来编写作业定义，我们使用 perl config general 来解析配置文件。该软件是高度定制的，具有诸如在检查是否满足该命令的依赖性后在我的工作中运行命令、在命令失败时重新安排工作、支持自定义通知等功能。

我们现在计划用 python 重写这个软件，并考虑使用 YAML 之类的选项而不是 apache 配置来编写作业定义。 YAML 是否适合编写此类动态配置？

作业定义示例(每天凌晨 2 点运行此作业，检查是否是星期二而不是印度的假期，如果是，请预订我的航类并发送通知):

// python function to check if it is tuesday
checkIfTuesdayAndNotHoliday()

<job> 
    calendar: indian

        <dependency: arbitrary_python_code: checkIfTuesdayAndNotHoliday()>
        <command>  
            check availability of flight
        </command>

        <success: notify: email: agrawall/>
        <failure: notify: email: ops>
        <command>
            some command to book my flight
        </command>
</job>

<crontab> 0 2 * * * </crontab>

我很难理解应该使用什么格式来定义作业(YAML、Apache Config、XML、JSON 等)。请注意，此作业定义将在我的 python 脚本中转换为作业对象。

我们当前使用的 perl 中的 Apache 配置解析器 https://metacpan.org/source/TLINDEN/Config-General-2.63/General.pm#L769

我们计划使用 python 中的 Apache 配置解析器 https://github.com/etingof/apacheconfig

Answer 1

基于 Python 的配置文件至少以 distutils 的形式存在' setup.py在 Python 1.6 中(即 2000 年之前)。使用这种格式的主要缺点是很难以编程方式更新配置中的值。即使你只是想做一些额外的实用程序来分析这些文件，你甚至必须特别注意你可以在不执行代码的情况下导入这样的配置文件，而且还没有通过导入引入各种依赖项。这可以通过使用 if __name__ == '__main__': 来实现或仅将配置信息作为文件中的数据结构更容易。

因此，如果更新文件永远不会成为问题，那么您可以使用基于 Python 的数据结构，这些结构非常可读。

XML 和 JSON 不是适合手动编辑的格式。 XML 有很多 <和 >无需特殊工具即可轻松打字。 JSON 有很多双引号，这让阅读变得困难，但它也有各种各样的问题，因为 JSON 不允许在数组和对象中使用尾随逗号，导致人们编写如下对象:

{ 
    "a": 1
  , "b": 2
}

这可以防止您删除最后一行并忘记删除分隔键/值对的逗号，但 IMO 可读性不同。

另一方面，YAML 可以编写得非常可读，但在编辑文件时必须考虑一些规则。在我的回答中 here我展示了一些可以包含在 YAML 文件中的基本规则，编辑人员在编辑时需要考虑这些规则。 YAML 可以被 Python 以外的其他语言读取(使用基于 Python 的配置文件很难做到这一点)。

您可以使用 YAML 标签(以及与这些标签关联的适当 Python 对象)，因此您不必依赖于从某些键值对中解释键来理解值解释的内容:

- !Job
  calendar: !Calendar indian
  dependency: !Arbitrary_python_code checkIfTuesdayAndNotHoliday()
  command: !CommandTester
     exec: !Exec check availability of flight
     success: !Commands
       - !Notify 
          email: agrawall
       - !Exec some command to book my flight
     failure: !Commands
       - !Notify 
           email: ops

(底部是与这些标签关联的类的部分示例实现)

当您使用 ruamel.yaml 时，即使不丢失注释、键顺序、标签，YAML 也可以以编程方式更新。 (免责声明:我是该软件包的作者)。

---

我一直在通过阅读我的通用 setup.py 的配置参数来参数化我的 Python 包(我管理了 100 多个包，其中一些在 PyPI 上，其他仅针对特定客户)。来自每个包裹的 __init__.py文件。我已经尝试插入 Python 的 JSON 子集，但最终开发了 PON (Python 对象表示法)可以很容易地被 setup.py 解析无需导入 __init__.py在 AST 上带有小(100 行)扩展名的文件 literal_eval包含在 Python 标准库中。

PON 可以在没有任何库的情况下使用(因为它是 Python 数据结构的子集，包括字典、列表、集合、元组和基本类型，如整数、 float 、 bool 值、字符串、日期、日期时间。因为它基于AST 评估器，您可以在配置文件中进行计算(secs_per_day = 24 * 60 * 60)和其他评估。

PON 自述文件还更详细地描述了该格式相对于 YAML、JSON、INI、XML 的优点(和缺点)。

使用配置数据不需要 PON 包，只有当您想对 PON 数据进行编程往返(加载-编辑-转储)时才需要。

---

import sys
from ruamel.yaml import YAML, yaml_object

yaml = YAML()

@yaml_object(yaml)
class CommandTester:
    yaml_tag = u'!CommandTester'

    def __init__(self, exec=None, success=None, failure=None):
        self.exec = exec
        self.success = success
        self.failure = failure

    def __call__(self):
        if self.exec():
            self.success()
        else:
            self.failure()

@yaml_object(yaml)
class Commands:
    """a list of commands"""
    yaml_tag = u'!Commands'

    def __init__(self, commands):
        self._commands = commands  # list of commands to execute

    @classmethod
    def from_yaml(cls, constructor, node):
        for m in yaml.constructor.construct_yaml_seq(node):
            pass
        return cls(m)

    @classmethod
    def to_yaml(cls, representer, node):
        return representer.represent_sequence(cls.yaml_tag, node._commands)

    def __call__(self, verbose=0, stop_on_error=False):
        res = True
        for cmd in self._cmd:
            try:
                res = subprocess.check_output(cmd)
            except Exception as e:
                res = False
                if stop_on_error:
                    break
            return res

@yaml_object(yaml)
class Command(Commands):
    """a single command"""
    yaml_tag = u'!Exec'

    def __init__(self, command):
        Commands.__init__(self, [command])

    @classmethod
    def from_yaml(cls, constructor, node):
        return cls(node.value)

    @classmethod
    def to_yaml(cls, representer, node):
        return representer.represent_scalar(cls.yaml_tag, node._commands[0])


@yaml_object(yaml)
class Notifier:
    yaml_tag = u'!Notify'

with open("job.yaml") as fp:
    job = yaml.load(fp)

yaml.dump(job, sys.stdout)