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

Compare commits

base: yumoqing:c1bca0a8e0ad84400863c681f7588f6d118b1898
Branches Tags
yumoqing:main
...
compare: yumoqing:c59e2dbaae68cecf79b3f6885b6d86798aa2a173
Branches Tags
yumoqing:main

2 Commits
c1bca0a8e0 ... c59e2dbaae

Author SHA1 Message Date
yumoqing
c59e2dbaae bugfix 2026-03-17 13:26:31 +08:00
yumoqing
9613b02ffe bugfix 2026-03-17 13:22:30 +08:00
1 changed files with 3 additions and 1 deletions
Show Stats Expand all files Collapse all files

4
dagflow/dagflow.py
View File

@ -312,6 +312,7 @@ where backid=${backid}$
}) })
break; break;
elif status in ['done', 'cancelled', 'failed']: elif status in ['done', 'cancelled', 'failed']:
debug(f'节点完成，检查后续节点')
await self.node_transfer(sor, dsl, inst, ne, node, flow_def) await self.node_transfer(sor, dsl, inst, ne, node, flow_def)
await sor.U('node_execution', { await sor.U('node_execution', {
'id': ne.id, 'id': ne.id,
@ -335,7 +336,7 @@ where backid=${backid}$
async def check_instance_completed(self, sor, inst): async def check_instance_completed(self, sor, inst):
sql = "select * from node_execution where instance_id=${instance_id} and type='end'" sql = "select * from node_execution where instance_id=${instance_id} and type='end'"
recs = await sor.sqlExe(sql, {'instance_id', inst.id}) recs = await sor.sqlExe(sql, {'instance_id': inst.id})
if recs: if recs:
if recs[0].status == 'completed': if recs[0].status == 'completed':
await sor.U('flow_instance', { await sor.U('flow_instance', {
@ -346,6 +347,7 @@ where backid=${backid}$
async def node_transfer(self, sor, dsl, inst, ne, node, flow_def): async def node_transfer(self, sor, dsl, inst, ne, node, flow_def):
nnodes = [] nnodes = []
ctx = json.loads(inst.ctx) ctx = json.loads(inst.ctx)
debug(f'{dsl=}, {inst=}, {ne=}, {node=}')
for edge in dsl.get('edges', []): for edge in dsl.get('edges', []):
if edge['from'] != ne.node_id: if edge['from'] != ne.node_id:
continue continue