审批流
旨在运维平台上实现一个简单通用(线性)的审批流
审批流节点
节点类型
节点有以下四种类型:
| 节点类型 | 说明 | 数量 |
|---|---|---|
| 开始 | 固定的第一个节点 | 有且仅有 1 个 |
| 审批 | 多人或批 | 不限 |
| 动作 | 内部的回调函数 | 不限 |
| 结束 | 固定的最后一个节点 | 有且仅有 1 个 |
- 开始与结束为固定节点
- 审批为或批,需要且批时使用多个审批节点
- 动作为回调,一般是在审批完成之后要执行的功能函数
示例
以运维平台上提供的 SQL上线 功能为例:
- 由审批流节点排列组合成特定的审批流步骤
SQL上线审批流总共5个步骤节点- 除了开始结束,有2个审批、1个动作
有哪些属性
| 属性 | 类型 | 含义 | 示例 |
|---|---|---|---|
| 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()飞书自定义审批卡片

