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 16:59:59 +08:00
parent d49ca32878
commit 5f852140a9
1 changed files with 28 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

56
dagflow/dagflow.py
View File

@ -24,33 +24,33 @@ aeval = Interpreter()
# 2. 在你的 node_transfer 或条件判断逻辑中 # 2. 在你的 node_transfer 或条件判断逻辑中
def safe_eval_condition(cond_str, ctx_dict): def safe_eval_condition(cond_str, ctx_dict):
""" """
安全地评估条件表达式 安全地评估条件表达式
Args: Args:
cond_str: 表达式字符串，例如 "ctx['age'] > 18 and ctx['score'] >= 60" cond_str: 表达式字符串，例如 "ctx['age'] > 18 and ctx['score'] >= 60"
ctx_dict: 上下文数据字典 ctx_dict: 上下文数据字典
Returns: None 错误或的确是None Returns: None 错误或的确是None
其他: 表达式结果 其他: 表达式结果
""" """
try: try:
# 将 ctx 注入到 asteval 的符号表中 # 将 ctx 注入到 asteval 的符号表中
aeval.symtable['ctx'] = ctx_dict aeval.symtable['ctx'] = ctx_dict
# 执行表达式 # 执行表达式
# mode='eval' 表示只允许单个表达式，不允许语句（如赋值、import） # mode='eval' 表示只允许单个表达式，不允许语句（如赋值、import）
result = aeval(cond_str, mode='eval') result = aeval(cond_str, mode='eval')
# 检查是否有语法错误或运行时错误 # 检查是否有语法错误或运行时错误
if aeval.error: if aeval.error:
print(f"asteval Error: {aeval.error}") print(f"asteval Error: {aeval.error}")
return False return False
return bool(result) return bool(result)
except Exception as e: except Exception as e:
# asteval 通常很安全，但以防万一 # asteval 通常很安全，但以防万一
print(f"Eval Exception: {e}") print(f"Eval Exception: {e}")
return False return False
# --------------------------------------------------------------------- # ---------------------------------------------------------------------
# Table definitions # Table definitions
@ -527,11 +527,11 @@ where instance_id=${instance_id}$
async def task_error_callback(self, ne_id, errmsg): async def task_error_callback(self, ne_id, errmsg):
env = Server() env = Server()
async with get_sor_context(env, 'dagflow') as sor: async with get_sor_context(env, 'dagflow') as sor:
recs = await sor.R('node_execution', {'id': ne_id}) recs = await sor.R('node_execution', {'id': ne_id})
if not recs: if not recs:
debug(f'{ne_id=} 没找到node_execution记录') debug(f'{ne_id=} 没找到node_execution记录')
return return
ne = recs[0] ne = recs[0]
ne.status = 'failed', ne.status = 'failed',
ne.output = errmsg ne.output = errmsg
await sor.U('node_execution', ne) await sor.U('node_execution', ne)