审批流
旨在运维平台上实现一个简单通用(线性)的审批流
审批流节点
节点类型
节点有以下四种类型:
节点类型 | 说明 | 数量 |
---|---|---|
开始 | 固定的第一个节点 | 有且仅有 1 个 |
审批 | 多人或批 | 不限 |
动作 | 内部的回调函数 | 不限 |
结束 | 固定的最后一个节点 | 有且仅有 1 个 |
- 开始与结束为固定节点
- 审批为或批,需要且批时使用多个审批节点
- 动作为回调,一般是在审批完成之后要执行的功能函数
示例
以运维平台上提供的 SQL上线
功能为例:
- 由审批流节点排列组合成特定的审批流步骤
SQL上线
审批流总共5个步骤节点- 除了开始结束,有2个审批同部门审批DBA审批、1个动作SQL执行
有哪些属性
属性 | 类型 | 含义 | 示例 |
---|---|---|---|
name | string | 步骤节点名称 | DBA审批 |
approval_users | list | 审批人列表 | ["zhangsan", "lisi"] |
action | string | 动作,仅在动作节点配置 | sql_execute |
is_agree | bool | 是否审批通过 | True |
note | string | 批注 | 同意 |
create_ts | float | 到达步骤的时间戳 | 1704971172.1719062 |
update_ts | float | 完成步骤的时间戳 | 1704971548.1112325 |
used_time | int | 此步骤耗时 | 375 |
used_time_desc | string | 易读的耗时 | 6分15秒 |
- 动作存储的为内部函数名,通过 importlib 库来关联使用,同时需要约定好所有回调函数均定义在同一个功能模块中,下述代码中函数位于
api.tasks
功能模块中:pythonaction = 'sql_execute' function = getattr(importlib.import_module('api.tasks'), action)
- 耗时也可不用存储,转而通过开始结束时间进行计算得出
如何配置
还是以 SQL上线
为例,其在数据库中的配置如下:
json
[
{
"name": "发起申请",
"approval_users": []
},
{
"name": "同部门审批",
"approval_users": [
"@perm[env.sql.approval]"
]
},
{
"name": "DBA审批",
"approval_users": [
"@duty_op"
]
},
{
"name": "SQL执行",
"approval_users": [],
"action": "sql_execute"
},
{
"name": "结束",
"approval_users": []
}
]
审批流的扭转
代码
使用一个递归函数来进行审批流的扭转,伪代码如下:
python
def approval_run(step, is_agree, approval_username=None):
"""审批流的扭转"""
if is_agree is True:
# 更新当前步骤为同意,并进入下一步
approval_run(step+1, None)
elif is_agree is False:
# 审批流完成,更新状态为驳回
return
elif is_agree is None:
# 1. 判断当前步骤是否为最后一步,是则完成并退出
if step == total_step:
# 审批流完成
return
# 2. 判断此步骤是否配置有审批人
if not current_step_has_approval_users:
# 如果无需审批,则自动同意
approval_run(step, True, None)
else:
# 2.1 判断申请人是否在此步的审批人当中
if username in current_approval_users:
# 申请人在这一步的审批人当中,自动同意
approval_run(step, True, username)
# 2.2 判断上一步的审批人是否在此步的审批人当中
elif last_step_approval_user in current_approval_users:
# 上一步的审批人在这一步的审批人当中,自动同意
approval_run(step, True, last_step_approval_user)
else:
# 创建一个审批通知给审批人
approval_create()