基于单个等幂python类和依赖项的作业编排框架。
dojobber的Python项目详细描述
dojobber
=
dojobber是一个python任务编排框架，它基于编写小的单任务幂等类（jobs），定义
相互依赖，并让python以"正确的顺序"完成运行它们的所有工作。工作图。它将运行没有未满足依赖关系的
作业，在链上运行
，直到它到达根目录或由于
作业失败而无法继续运行。
每个作业都有一个目的，并且必须是等幂的，
即，如果执行一次或执行多个
，它将产生相同的结果因此，您可以多次运行您的python脚本
，并且随着以前失败的作业的成功，它将越来越接近完成。
我的朋友们过来看电影-这是"tests/dojobber-example.py"脚本的结果。图片：：https://raw.githubusercontent.com/extrahop/dojobber/master/example.png
：alt:dojobber示例图
：宽度：90%
：align:center
de
您可以直接进入dojobber框架。
dojobber在概念上基于一个名为
masher的google程序，该程序是为自动化服务和数据中心
备份而构建的，但不与它共享代码。
job structure
===
少女。举个例子：
_kwargs）：
pass
每个作业都有一个deps属性，``check``方法和``run``方法。
deps
---
deps定义它所依赖的其他作业。这用于生成内部图。
check
----
``check``执行，如果没有引发异常，则认为
已通过。如果通过，则作业通过，下一个作业将
运行。其目的是验证我们是否处于
此工作所需的状态。例如，如果作业要创建用户，这可能会在/etc/passwd中查找该用户。
run
--
`run``在"check"失败时执行。它的工作是做些事情来实现我们的目标。Dojobber不在乎它是否返回任何内容、抛出
异常或退出—所有这些都将被忽略。
``再次检查``方法。希望
"run"方法完成了必要的工作，因此"check"将验证
现在一切正常。如果是这样（即"check"不会引发异常），则
我们认为此作业成功，并且不会阻止任何依赖作业
运行。
这些是通过
set-args方法设置的：
dojob.set-args（'arg1'，'arg2'，foo='foo'，bar='bar'，…）
因此，最好在"check"和"run"方法中同时接受：
def check（self，*args，**kwargs）：
…
def run（self，*args，**kwargs）：
…
myparser.myparser.add参数（'--movie'movie'dest'movie'movie'help'movie to watch.'）
…
args=myparser.parse args（）
dojob.set args（**args参数参数参数参数参数参数参数参数参数参数设置参数参数参数参数参数参数参数参数参数参数设置参数参数参数参数参数参数参数参数参数参数
n然后在` ` ` ` ` ` ` ` ` ` ` ` ` ` ` ` ` ` ` ` ` `运行``运行``中使用参数参数参数参数参数参数参数参数参数参数时，可以使用它们的名称：
'：
raise error（"真的吗？"）
本地作业存储
----
本地存储允许您在作业的"check"和"run"方法之间共享信息。例如，``check`
可能会进行代价高昂的查找或初始化，然后，``run``可能会用来加快工作速度。
有关如何在作业之间共享信息，请参见全局作业存储。
方法）
def run（self，*dummy_args，**kwargs）：
子进程.调用（command+[self.storage['sql_username']]）
global job storage
----
global storage允许您在
作业之间共享信息。当然，您需要确保任何需要全局存储的
作业都定义为
依赖于设置全局存储的作业。
若要使用全局作业存储，只需使用
"check"和/或"run"方法中的"self.global\u storage"字典即可。
这取决于
避免命名冲突。
示例：：
对于打开的X（'/proc/cpuinfo'）.readlines（）
如果X中有"供应商ID"）
…..
清除
----
作业可以有清除方法。checknrun完成后，
将执行运行的每个作业（即"run"已执行）的清除方法
。它们是按后进先出顺序运行的，因此清除"展开"所有内容。
/>
动态创建作业
--------
如果需要基于命令行选项创建新作业，这将非常有用。动态创建许多小的单用途作业
比创建一个动态确定其需要执行和检查的大型单片作业要好得多。
``和``name``:
基本作业
类sendInvite（作业）：
email=none
name=none
def check（self，*args，**kwargs）：
r=requests.get（
'https://api.example.com/invited/'+self.email）
断言（r.status\u code==200）
def run（self，*args，**kwargs）：
requests.post（
'https://api.example.com/invite/'+self.email）
基于此类的新作业，您可以创建它们，并将它们
放到现有作业的"deps"中，以便它们显示在图形中：
class invitefriends（dummyjob）：
"将动态依赖于其他作业的作业。""
deps=[]
def in邀请朋友（人）：
"为这些人添加邀请工作。
people是一个字典列表，其中包含关键电子邮件和姓名。
"
"对于人中的人：
job=type（'in vite{}）。format（person['name']），
（sendinvite，），{}）
job.email=person['email']
job.name=person['name']
invitefriends.deps.append（job）
def main（）：
_朋友们（[
{name'：'wendell bagg'，'email'：'bagg@example.com'}，
{name'：'lawyer cat'，'email'：'lawyercat@example.com'}
]）
成功。你可能会想写一些类似这样的东西：
…
retry=5
而retry:
dojob.checknrun（）
如果dojob.success（）：
break
print（'trying again…'）
retry-=1
实际上，这是在浪费计算周期。上面的代码会导致我们不必要地检查已经成功的节点，从而减慢速度。
相反，您可以使用两个类属性来配置retry
参数。`` tries``指定在我们放弃之前，作业可以
erun的次数，``retry_delay``指定重试之间的最短时间量。
重试对于由于临时条件（可能远程服务器
是una）而导致``run`
中的操作失败的情况非常有用。短暂可用），或者在"run"中触发的活动需要时间才能完成（可能API调用会立即返回，但后台填充需要30秒）。
有一个更健壮的
作业图。
这意味着
在运行**之间，"存储"不可用，但是"全局存储"不可用。这样做是为了在
作业执行之间尽可能保持原始状态。
tries定义了在放弃之前允许作业执行的尝试次数（检查/运行/重新检查周期）
。它必须是>；=1。
如果未指定，则尝试默认值为3，可以通过"default_tries=``参数在"configure（）"中更改
例如：
class foo（job）：
tries=10
…
class bar（job）：
deps=（foo，）
…#no tries属性
…
.
重试延迟
----
重试延迟定义在发出带有永久故障的
之前，在此**作业的
次尝试（检查/运行/重新检查周期）之间等待的最短时间。它以秒为单位，may
是任何非负数值，包括0和小数
秒，如0.02。
重试延迟=10.5一个很长但非常精确的值…
…
class bar（job）：
deps=（foo，）
…#无重试延迟属性
…
dojob=dojobber.dojobber（）
dojob.configure（foo，默认重试延迟=0.5）
在上述情况下，除非自上次foo尝试以来至少经过10.5秒
，否则foo将永远不会启动，而bar只需要
0.5 se条件已通过，因为它没有指定"重试延迟"，配置中的"默认重试延迟"为0.5。
检查/运行/复查。一旦每个可访问的作业都被命中，我们将"重新开始"失败的作业。
（除非图形是高度线性且没有未阻塞的作业。）
sitoncouch.重新检查：失败"沙发上没有空间。"
爆米花碗。重新检查：失败"洗碗机循环尚未完成。"
披萨。重新检查：失败"giordano还未到达。"
turnontv.重新检查：失败"遥控电池没电。"
sitoncouch.重新检查：失败"沙发上没有空间。"
pop玉米碗。重新检查：不合格"洗碗机循环尚未完成。"
比萨饼。重新检查：不合格"乔达诺还未到达。"
Turnontv。重新检查：不合格"远程电池已耗尽。"
SitonCouch。重新检查：不合格"沙发上没有空间。"
爆米花碗。重新检查：不合格"洗碗机循环尚未完成。"
爆米花碗。重新检查：失败"洗碗机循环尚未完成。"
爆米花碗。重新检查：失败"洗碗机循环尚未完成。"
爆米花。重新检查：失败"仍然爆米花…"
爆米花。重新检查：失败"仍然爆米花…"
请注意，最初我们有几项工作在
蒸馏时失败。ct分支，这些可以以循环方式重试。只有当我们最终严格依赖于爆米花碗和爆米花时，我们才会看到单个作业在没有其他作业时间的情况下被重试。
可以有可选的清理：
类create user（job）：
"创建用户帐户。"
def check（self，*\u，**kwargs）：
"验证用户是否存在"
import pwd
pwd.getpwnam（kwargs['username']）
def run（self，*2;，**kwargs）：
"根据命令行用户名/gecos参数创建用户"
"导入子进程
子进程。调用（[
"sudo"，'/usr/sbin/adduser'，
"shell"，'/bin/tcsh'，
"gecos"，kwargs['gecos']，
kwargs['username']）
它只用于
有一个作业用于对依赖的或动态创建的作业进行分组。
因此dummyjob看起来很简单：
类占位符（dummyjob）：
deps=（dependency1，dependency2，…）
runonlyjob
——
a``runonlyjob``没有检查，只有一个``run```每次都会运行。
如果``run``引发异常，则认为作业失败。
er已执行。
因此，示例"run"可能如下所示：：
class removedously（runonlyjob）：
deps=（useracceptedtheconsequences，）
def run（…）：
os.system（'rm-rf/'）
一般情况下，请避免"runonlyjob s"-最好是了解是否需要进行更改。
bber
调试，这在编写dojobber代码时非常有用。
运行时调试
----
要增加作业成功和失败的详细程度，请将"verbose"或"debug"关键字参数传递给"configure"：
dojob=dojobber.dojobber（）
dojob.configure（rootjob，verbose=true，….）
或
dojob.configure（rootjob，debug=true，….）
emote.run:pass
findTvremote.reveck:pass
turnonTV.check:fail
turnonTV.run:pass
…
在编写dojobber代码时，您可能希望启用开发人员调试功能。当dojobber通过传递'dojobber_loglevel`
关键字参数来初始化时，这将启用
import logging
dojob=dojobber（dojobber_loglevel=logging.debug）
dojobber的默认值是仅显示'critical'错误。
可接受的级别是在日志模块。
这有助于在编写代码时发现问题，
例如将不可iterable作为"deps"变量传递，
监视从
类创建的作业图等。
源目录是
功能齐全的测试套件，其中有大量的注释散布在各处。
你可以在他的网站上找到他的
`presentation<；https://www.ifokr.org/bri/presentations/lfnw-2018-dojobber/>；``u
，以及
`presentation video<；https://www.youtube.com/watch？v=bjtlw6ndouw>；``可以在YouTube上找到。
欢迎加入QQ群-->： 979659372
