symfonic.agent.scheduler

scheduler

InProcessScheduler -- lightweight asyncio-based task scheduler.

Runs tasks in the current event loop with no persistence and no external dependencies. For production use, replace with a real scheduler (Celery, APScheduler, etc.).

InProcessScheduler

InProcessScheduler()

Lightweight asyncio-based task scheduler.

Runs tasks in the current event loop. No persistence, no external dependencies. For production, replace with a real scheduler (Celery, APScheduler, etc.).

Source code in src/symfonic/agent/scheduler.py

def __init__(self) -> None:
    self._tasks: dict[str, asyncio.Task[Any]] = {}
    self._running = True

cancel async

cancel(name: str) -> bool

Cancel a scheduled task by name.

Returns True if the task existed and was cancelled.

Source code in src/symfonic/agent/scheduler.py

async def cancel(self, name: str) -> bool:
    """Cancel a scheduled task by name.

    Returns True if the task existed and was cancelled.
    """
    return await self._cancel_existing(name)

list_tasks async

list_tasks() -> list[dict[str, Any]]

List all scheduled tasks with status.

Source code in src/symfonic/agent/scheduler.py

async def list_tasks(self) -> list[dict[str, Any]]:
    """List all scheduled tasks with status."""
    return [
        {
            "name": name,
            "done": task.done(),
            "cancelled": task.cancelled(),
        }
        for name, task in self._tasks.items()
    ]

schedule_interval async

schedule_interval(
    name: str,
    coro_factory: Callable[[], Awaitable[Any]],
    interval_seconds: float,
) -> None

Schedule a coroutine to run repeatedly at a fixed interval.

Parameters:

Name	Type	Description	Default
name	str	Unique task identifier.	required
coro_factory	Callable[[], Awaitable[Any]]	Zero-arg callable that returns a fresh awaitable on each invocation.	required
interval_seconds	float	Seconds between successive executions.	required

Source code in src/symfonic/agent/scheduler.py

async def schedule_interval(
    self,
    name: str,
    coro_factory: Callable[[], Awaitable[Any]],
    interval_seconds: float,
) -> None:
    """Schedule a coroutine to run repeatedly at a fixed interval.

    Args:
        name: Unique task identifier.
        coro_factory: Zero-arg callable that returns a fresh awaitable
            on each invocation.
        interval_seconds: Seconds between successive executions.
    """
    if not self._running:
        logger.warning(
            "Scheduler is shut down; ignoring schedule_interval(%s)", name,
        )
        return
    await self._cancel_existing(name)

    async def _loop() -> None:
        try:
            while True:
                await asyncio.sleep(interval_seconds)
                try:
                    await coro_factory()
                except Exception:
                    logger.warning(
                        "Interval task %s iteration failed", name, exc_info=True,
                    )
        except asyncio.CancelledError:
            logger.debug("Interval task %s cancelled", name)
        finally:
            self._tasks.pop(name, None)

    loop = asyncio.get_running_loop()
    self._tasks[name] = loop.create_task(_loop())

schedule_once async

schedule_once(
    name: str,
    coro: Awaitable[Any],
    *,
    delay_seconds: float = 0,
) -> None

Schedule a coroutine to run once after an optional delay.

Parameters:

Name	Type	Description	Default
name	str	Unique task identifier. Cancels any existing task with the same name before scheduling.	required
coro	Awaitable[Any]	Awaitable to execute.	required
delay_seconds	float	Seconds to wait before execution (default 0).	0

Source code in src/symfonic/agent/scheduler.py

async def schedule_once(
    self,
    name: str,
    coro: Awaitable[Any],
    *,
    delay_seconds: float = 0,
) -> None:
    """Schedule a coroutine to run once after an optional delay.

    Args:
        name: Unique task identifier.  Cancels any existing task
            with the same name before scheduling.
        coro: Awaitable to execute.
        delay_seconds: Seconds to wait before execution (default 0).
    """
    if not self._running:
        logger.warning("Scheduler is shut down; ignoring schedule_once(%s)", name)
        return
    await self._cancel_existing(name)

    async def _wrapper() -> None:
        try:
            if delay_seconds > 0:
                await asyncio.sleep(delay_seconds)
            await coro
        except asyncio.CancelledError:
            logger.debug("Task %s cancelled", name)
        except Exception:
            logger.warning("Task %s failed", name, exc_info=True)
        finally:
            self._tasks.pop(name, None)

    loop = asyncio.get_running_loop()
    self._tasks[name] = loop.create_task(_wrapper())

shutdown async

shutdown() -> None

Cancel all tasks and mark the scheduler as stopped.

Source code in src/symfonic/agent/scheduler.py

async def shutdown(self) -> None:
    """Cancel all tasks and mark the scheduler as stopped."""
    self._running = False
    tasks = list(self._tasks.values())
    for task in tasks:
        task.cancel()
    if tasks:
        await asyncio.gather(*tasks, return_exceptions=True)
    self._tasks.clear()