INNER CODE UNIT · Python

_run_pending_tasks

evolution-foundation/evo-nexus · dashboard/backend/app.py:1080

    def _run_pending_tasks():
        """Check for pending scheduled tasks and execute them."""
        from datetime import datetime as _dt, timezone as _tz
        from models import ScheduledTask

        try:
            now = _dt.now(_tz.utc)
            pending = ScheduledTask.query.filter(
                ScheduledTask.status == "pending",
                ScheduledTask.scheduled_at <= now,
            ).all()

            for task in pending:
                log_path = WORKSPACE / "ADWs" / "logs" / "scheduler.log"
                with open(log_path, "a") as log:
                    log.write(f"  [{_dt.now().strftime('%H:%M')}] Running scheduled task #{task.id}: {task.name}\n")

                t = threading.Thread(target=_execute_task_with_context, args=(task.id,), daemon=True)

View source record →