INNER CODE UNIT · Python

_add_to_scheduler

bunkeriot/BunkerM · backend/app/agent-api/main.py:355

def _add_to_scheduler(job: dict):
    try:
        trigger = CronTrigger.from_crontab(job["cron"])
        scheduler.add_job(
            _publish_and_record, trigger=trigger, id=job["id"],
            args=[job["topic"], job["payload"], job.get("qos", 0), job.get("retain", False), job["id"]],
            replace_existing=True,
        )
    except Exception as e:
        logger.warning(f"Could not schedule job {job['id']}: {e}")


@app.get("/schedules")
async def list_schedules(x_api_key: str = Header(...)):
    _auth(x_api_key)
    with _lock:
        data = _load_agents()
    return {"jobs": data["jobs"]}

View source record →