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-02-13 15:57:15 +08:00
parent de796b7d7e
commit 03a356008d
1 changed files with 13 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
dagflow/dagflow.py
View File

@ -11,6 +11,8 @@ Features:
- RBAC-based task assignment - RBAC-based task assignment
- Query APIs for UI - Query APIs for UI
""" """
from appPublic.log import debug
from sqlor.dbpools import get_sor_context
# --------------------------------------------------------------------- # ---------------------------------------------------------------------
# Table definitions # Table definitions
@ -104,15 +106,20 @@ class FlowEngine:
def __init__(self): def __init__(self):
self.aio = [] self.aio = []
async def bgtask(self, app): async def bgtask(self):
env = ServerEnv() env = ServerEnv()
self.aio = [] self.aio = []
cnt = 0
while True:
async with (env, 'workflow') as sor: async with (env, 'workflow') as sor:
self.aio = await sor.R('flow_instance', {'status': 'running'}) self.aio = await sor.R('flow_instance', {'status': 'running'})
while True: if cnt >= 60:
debug(f'{self.aio=}')
cnt = 0
for r in self.aio: for r in self.aio:
self.step(r.org_id, r.id) self.step(r.org_id, r.id)
await asyncio.sleep(0.5) await asyncio.sleep(2)
cnt += 1
async def create_definition(self, org_id, name, version, dsl_text): async def create_definition(self, org_id, name, version, dsl_text):
env = ServerEnv() env = ServerEnv()
@ -129,7 +136,7 @@ class FlowEngine:
async def create_instance(self, org_id, flow_def_id, ctx=None): async def create_instance(self, org_id, flow_def_id, ctx=None):
env = ServerEnv() env = ServerEnv()
async with (env, 'workflow') as sor: async with get_sor_context(env, 'dagflow') as sor:
iid = env.uuid() iid = env.uuid()
await sor.C('flow_instance', { await sor.C('flow_instance', {
'id': iid, 'id': iid,
@ -147,7 +154,7 @@ class FlowEngine:
async def step(self, org_id, instance_id): async def step(self, org_id, instance_id):
env = ServerEnv() env = ServerEnv()
async with get_sor_context(env, 'workflow') as sor: async with get_sor_context(env, 'dagflow') as sor:
insts = await sor.R('flow_instance', { insts = await sor.R('flow_instance', {
'id': instance_id, 'id': instance_id,
'org_id': org_id 'org_id': org_id