yumoqing/dagflow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bugfix

Browse Source
This commit is contained in:
yumoqing 2026-03-17 15:43:35 +08:00
parent c59e2dbaae
commit c12553990b
1 changed files with 61 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

122
dagflow/dagflow.py
View File

@ -285,46 +285,58 @@ where backid=${backid}$
node['id'] = node_id
ntype = node['type']
status = ne.status
if status == 'pending':
if ntype == 'human':
if ne.assignee is None:
await self.auto_assign(sor, inst, ne, node)
elif ntype == 'task':
await self.try_start_running(sor, inst, ne, node)
elif ntype == 'subflow':
await self.try_start_subflow(sor, inst, ne, node)
elif ntype == 'start':
if ntype == 'start':
await sor.U('node_execution', {
'id': ne.id,
'status': 'completed'
})
await self.node_transfer(sor, dsl, inst, ne, node, flow_def)
elif ntype == 'end':
flg = await self.is_ok_to_step_next(sor, node, edges, inst)
if flg:
await sor.U('node_execution', {
'id': ne.id,
'status': 'completed'
})
await sor.U('flow_instance', {
'id': inst.id,
'status': 'completed'
})
break;
else:
if status == 'pending':
if ntype == 'human':
if ne.assignee is None:
await self.auto_assign(sor, inst, ne, node)
elif ntype == 'task':
flg= self.is_ok_to_step_next(sor, node, edges, inst)
if flg:
await self.try_start_running(sor, inst, ne, node)
elif ntype == 'subflow':
flg= self.is_ok_to_step_next(sor, node, edges, inst)
if flg:
await self.try_start_subflow(sor, inst, ne, node)
elif status in ['done', 'cancelled']:
debug(f'节点完成，检查后续节点')
await self.node_transfer(sor, dsl, inst, ne, node, flow_def)
await sor.U('node_execution', {
'id': ne.id,
'status': 'completed'
})
elif ntype == 'end':
flg = await self.is_ok_to_step_next(sor, node, edges, inst)
if flg:
await sor.U('node_execution', {
'id': ne.id,
'status': 'completed'
})
await sor.U('flow_instance', {
'id': inst.id,
'status': 'completed'
})
break;
elif status in ['done', 'cancelled', 'failed']:
debug(f'节点完成，检查后续节点')
await self.node_transfer(sor, dsl, inst, ne, node, flow_def)
await sor.U('node_execution', {
'id': ne.id,
'status': 'completed'
})
await self.check_instance_completed(sor, inst)
elif status == 'failed':
debug(f'节点执行失败，需要运维人员参与')
async def run_auto_task(self, ne):
env = ServerEnv()
users = get_org_users(ne.orgid)
if not users:
debug(f'{orgid=} 没有用户')
return
# ------------------------------
# Human task APIs
# ------------------------------
async def auto_assign(self, sor, inst, ne, node):
recs = await env.get_org_users(sor, inst.org_id, ne.role)
if recs:
@ -334,15 +346,11 @@ where backid=${backid}$
'assignee': recs[i].id
})
async def check_instance_completed(self, sor, inst):
sql = "select * from node_execution where instance_id=${instance_id} and type='end'"
recs = await sor.sqlExe(sql, {'instance_id': inst.id})
if recs:
if recs[0].status == 'completed':
await sor.U('flow_instance', {
'id': inst.id,
'status': 'completed'
})
async def instance_completed(self, sor, inst):
await sor.U('flow_instance', {
'id': inst.id,
'status': 'completed'
})
async def node_transfer(self, sor, dsl, inst, ne, node, flow_def):
nnodes = []
@ -455,7 +463,7 @@ where node_id in ${backnodes}$
})
if join == 'all':
for r in recs:
if r.status in ['pending', 'running']:
if r.status != 'completed':
return False
return True
if join == 'xor':
@ -484,28 +492,20 @@ where instance_id=${instance_id}$
return True
return False
async def start_running(self, sor, node, edges, inst):
flg= self.is_ok_to_step_next(sor, node, edges, inst)
if flg:
await sor.R('node_execution', {
'id': node['id'],
'status': 'running',
'running_time': timestampstr()
async def start_running(self, sor, ne, edges, inst):
await sor.R('node_execution', {
'id': ne['id'],
'status': 'running',
'running_time': timestampstr()
})
try:
out = await self.run_auto_task(ne)
except Exception as e:
await sor.U('node_execution', {
'id': ne['id'],
'status': 'failed',
'stopping_at': timestampstr()
})
try:
out = await self.run_auto_task(node)
await sor.U('node_execution', {
'id': node['id'],
'status': 'done',
'stopping_at': timestampstr(),
'output_ctx': json.dumps(out, ensure_ascii=False)
})
except Exception as e:
await sor.U('node_execution', {
'id': node['id'],
'status': 'failed',
'stopping_at': timestampstr()
})
async def list_human_tasks(self, org_id, user_roles):
env = ServerEnv()