orionis.console.tasks.schedule.Schedule

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

class documentation

class Schedule(ISchedule): (source)

Constructor: Schedule(reactor, exception_handler, stores)

View In Hierarchy

Undocumented

Method	`__addConfiguredJobStore`	Add the database or redis jobstore matching the configured driver.
Method	`__createManagedTask`	Create and manage an asyncio task with automatic cleanup.
Async Method	`__gracefulShutdown`	Execute graceful shutdown of the scheduler without blocking.
Async Method	`__handleListenerException`	Handle exceptions raised in event listeners.
Method	`__init__`	Initialize the Schedule instance.
Async Method	`__loadAvailableCommands`	Retrieve and cache available command signatures from the reactor.
Method	`__suppressApschedulerLogging`	Disable APScheduler logging to prevent log message interference.
Async Method	`__validateAndLoadFluentTasks`	Validate and load fluent tasks from the internal registry.
Method	`_dispatchSchedulerEventListener`	Call the registered global event listener for a scheduler event.
Method	`_dispatchTaskEventListener`	Dispatch registered task event listener for a specific job event.
Async Method	`_reactorCall`	Execute a reactor command asynchronously.
Async Method	`boot`	Boot the scheduler and register all configured tasks.
Method	`command`	Add a command for fluent configuration.
Async Method	`info`	Retrieve information about all loaded fluent tasks.
Method	`isPaused`	Determine if the scheduler is currently paused.
Method	`isRunning`	Determine if the scheduler is currently running.
Method	`isStopped`	Determine if the scheduler is currently stopped.
Method	`on`	Register a listener for a scheduler event.
Method	`pause`	Pause the scheduler if it is running.
Method	`pauseTask`	Pause a running task by its signature.
Method	`removeAllTasks`	Remove all tasks from the scheduler.
Method	`removeTask`	Remove a task from the scheduler by its signature.
Method	`resume`	Resume the scheduler if it is paused.
Method	`resumeTask`	Resume a paused task by its signature.
Method	`shutdown`	Shut down the task scheduler safely without waiting for running tasks.
Method	`state`	Return the current scheduler state as a string.
Async Method	`wait`	Wait for the scheduler shutdown to complete.
Constant	`_SCHEDULER_NOT_STARTED_ERROR`	Undocumented
Instance Variable	`__available_command_signatures`	Undocumented
Instance Variable	`__config`	Undocumented
Instance Variable	`__exception_handler`	Undocumented
Instance Variable	`__fluent_tasks`	Undocumented
Instance Variable	`__job_store`	Undocumented
Instance Variable	`__paused_tasks`	Undocumented
Instance Variable	`__pending_listener_tasks`	Undocumented
Instance Variable	`__reactor`	Undocumented
Instance Variable	`__removed_tasks`	Undocumented
Instance Variable	`__running_tasks`	Undocumented
Instance Variable	`__scheduler`	Undocumented
Instance Variable	`__scheduler_listeners`	Undocumented
Instance Variable	`__shutdown_event`	Undocumented
Instance Variable	`__state`	Undocumented
Instance Variable	`__stores`	Undocumented
Instance Variable	`__tasks`	Undocumented
Instance Variable	`__tasks_listeners`	Undocumented
Instance Variable	`__wait_to_shutdown`	Undocumented

def __addConfiguredJobStore(self): (source) ¶

Add the database or redis jobstore matching the configured driver.

stores.database/stores.redis default to None when not declared in config, so this validates explicitly instead of letting an AttributeError leak out of the config object when the selected driver's dedicated section was never configured.

Returns
`None`	This method does not return a value. It registers the corresponding jobstore on the running scheduler, if applicable.

Raises
`RuntimeError`	If the selected store is "database" or "redis" but its dedicated configuration section is missing.

def __createManagedTask(self, coroutine: Coroutine): (source) ¶

Create and manage an asyncio task with automatic cleanup.

Parameters
coroutine:`Coroutine`	The coroutine to be executed as an async task.

Returns
`None`	This method does not return a value. It manages the lifecycle of the created asyncio task.

async def __gracefulShutdown(self): (source) ¶

Execute graceful shutdown of the scheduler without blocking.

This method runs the scheduler shutdown in a thread executor to avoid blocking the event loop, then sets the shutdown event to notify any waiting tasks that shutdown is complete.

Returns
`None`	This method does not return a value.

async def __handleListenerException(self, exception: Exception): (source) ¶

Handle exceptions raised in event listeners.

Parameters
exception:`Exception`	The exception raised by the listener.

Returns
`None`	This method does not return any value.

def __init__(self, reactor: IReactor, exception_handler: ICatch, stores: IScheduleStore): (source) ¶

Initialize the Schedule instance.

Parameters
reactor:`IReactor`	Reactor instance for command execution.
exception_handler:`ICatch`	Exception handler for managing errors.
stores:`IScheduleStore`	Schedule store for managing job stores.

Returns
`None`	This constructor does not return a value.

async def __loadAvailableCommands(self): (source) ¶

Retrieve and cache available command signatures from the reactor.

Queries the reactor for all available jobs/commands and stores their signatures in the internal set for quick lookup.

Returns
`None`	This method does not return any value. It updates the internal set of available command signatures.

def __suppressApschedulerLogging(self): (source) ¶

Disable APScheduler logging to prevent log message interference.

Set APScheduler component loggers to CRITICAL level and disable message propagation to prevent internal scheduler logs from appearing in the application's logging output.

Returns
`None`	Modifies APScheduler logging configuration in place.

async def __validateAndLoadFluentTasks(self): (source) ¶

Validate and load fluent tasks from the internal registry.

Ensures that all fluent tasks have signatures available in the reactor. Populates the internal tasks dictionary with validated task entities.

Returns
`None`	This method updates the internal tasks dictionary in place.

def _dispatchSchedulerEventListener(self, event: APSchedulerEvent): (source) ¶

Call the registered global event listener for a scheduler event.

Parameters
event:`APSchedulerEvent`	The scheduler event that occurred.

Returns
`None`	This method does not return a value. It triggers the appropriate listener for the event.

def _dispatchTaskEventListener(self, event: APJobEvent): (source) ¶

Dispatch registered task event listener for a specific job event.

Parameters
event:`APJobEvent`	The job event that occurred.

Returns
`None`	This method does not return a value. It triggers the appropriate listener for the event.

async def _reactorCall(self, signature: str, args: list[str] | None = None) -> int: (source) ¶

Execute a reactor command asynchronously.

Notes

This method wraps the reactor call for task execution.

Parameters
signature:`str`	The command signature to execute.
args:`list of str, optional`	Arguments to pass to the command. Defaults to an empty list.

Returns
`int`	The result of the reactor call, typically an exit code or status.

async def boot(self): (source) ¶

Boot the scheduler and register all configured tasks.

Load available command signatures, validate and load fluent tasks, initialize the scheduler, register event listeners, and add all jobs to the scheduler.

Returns
`None`	This method does not return a value but initializes the scheduler.

def command(self, signature: str, args: list[str] | None = None, purpose: str | None = None) -> ITask: (source) ¶

Add a command for fluent configuration.

Parameters
signature:`str`	Unique signature of the command to schedule.
args:`list[str] \| None, optional`	Arguments for the command. Defaults to None.
purpose:`str \| None, optional`	Description of the command's purpose.

Returns
`ITask`	Task instance for further configuration.

Raises
`RuntimeError`	If the scheduler has already been started.
`TypeError`	If the signature is not a non-empty string or arguments are invalid.

async def info(self) -> list[dict]: (source) ¶

Retrieve information about all loaded fluent tasks.

Returns
`list of dict`	A list of dictionaries, each containing details about a loaded task.

def isPaused(self) -> bool: (source) ¶

Determine if the scheduler is currently paused.

Returns
`bool`	True if the scheduler state is "PAUSED", otherwise False.

def isRunning(self) -> bool: (source) ¶

Determine if the scheduler is currently running.

Returns
`bool`	True if the scheduler state is "RUNNING", otherwise False.

def isStopped(self) -> bool: (source) ¶

Determine if the scheduler is currently stopped.

Returns
`bool`	True if the scheduler state is "STOPPED", otherwise False.

def on(self, event: SchedulerEvent, listener: Callable) -> Self: (source) ¶

Parameters
event:`SchedulerEvent`	The scheduler event to listen for.
listener:`Callable`	The callable to invoke when the event occurs.

Returns
`Self`	The Schedule instance for method chaining.

Raises
`RuntimeError`	If the scheduler has already been booted.
`TypeError`	If event is not a SchedulerEvent or listener is not callable.

def pause(self) -> bool: (source) ¶

Pause the scheduler if it is running.

Parameters
self:`Schedule`	The Schedule instance.

Returns
`bool`	True if the scheduler was successfully paused.

Raises
`RuntimeError`	If the scheduler is not started or not running.

def pauseTask(self, signature: str) -> bool: (source) ¶

Pause a running task by its signature.

Notes

This method pauses a running task in the scheduler.

Parameters
signature:`str`	Unique identifier of the task to pause.

Returns
`bool`	True if the task was successfully paused, otherwise raises an exception.

Raises
`RuntimeError`	If the scheduler has not been started or the task is not running.
`ValueError`	If the specified task does not exist.

def removeAllTasks(self) -> bool: (source) ¶

Remove all tasks from the scheduler.

Notes

This method removes all tasks from the scheduler and updates internal state.

Returns
`bool`	True if all tasks were successfully removed, otherwise raises an exception.

Raises
`RuntimeError`	If the scheduler has not been started or an error occurs during removal.

def removeTask(self, signature: str) -> bool: (source) ¶

Remove a task from the scheduler by its signature.

Notes

This method removes a task from the scheduler and updates internal state.

Parameters
signature:`str`	Unique identifier of the task to remove.

Returns
`bool`	True if the task was successfully removed, otherwise raises an exception.

Raises
`RuntimeError`	If the scheduler has not been started or the task is not found.
`ValueError`	If the specified task does not exist.

def resume(self) -> bool: (source) ¶

Resume the scheduler if it is paused.

Parameters
self:`Schedule`	The Schedule instance.

Returns
`bool`	True if the scheduler was successfully resumed.

Raises
`RuntimeError`	If the scheduler is not started or not paused.

def resumeTask(self, signature: str) -> bool: (source) ¶

Resume a paused task by its signature.

Notes

This method resumes a paused task in the scheduler.

Parameters
signature:`str`	Unique identifier of the task to resume.

Returns
`bool`	True if the task was successfully resumed, otherwise raises an exception.

Raises
`RuntimeError`	If the scheduler has not been started or the task is not paused.
`ValueError`	If the specified task does not exist.

def shutdown(self, wait: int | None = None): (source) ¶

Shut down the task scheduler safely without waiting for running tasks.

This method terminates the scheduler execution safely. It does not wait for currently executing tasks to complete, but prevents new tasks from starting and cleans up scheduler resources. Ideal for console environments where the process stops immediately after calling this method.

Parameters
wait:`int \| None, optional`	Time in seconds to wait before completing shutdown. Defaults to None, which keeps the previously configured grace period instead of collapsing it to zero.

Returns
`None`	This method does not return a value. It initiates graceful shutdown.

Raises
`TypeError`	If `wait` is neither None nor a non-negative integer (bool is explicitly rejected even though it is technically an int subtype).

def state(self) -> str: (source) ¶

Return the current scheduler state as a string.

Returns
`str`	The current state of the scheduler, e.g., "RUNNING", "PAUSED", or "STOPPED".

async def wait(self): (source) ¶

Wait for the scheduler shutdown to complete.

This method blocks until the shutdown process initiated by the shutdown() method has finished. It provides a way to synchronize with the graceful shutdown process.

Returns
`None`	This method does not return a value but blocks until shutdown completes.

_SCHEDULER_NOT_STARTED_ERROR: str = (source) ¶

Undocumented

Value

'The Orionis task scheduler has not been started.'

__available_command_signatures: set[str] = (source) ¶

Undocumented

__config: ConfigScheduler = (source) ¶

Undocumented

__exception_handler: ICatch = (source) ¶

Undocumented

__fluent_tasks: dict[str, ITask] = (source) ¶

Undocumented

__job_store: str = (source) ¶

Undocumented

__paused_tasks: set[str] = (source) ¶

Undocumented

__pending_listener_tasks: set[asyncio.Task] = (source) ¶

Undocumented

__reactor: IReactor = (source) ¶

Undocumented

__removed_tasks: set[str] = (source) ¶

Undocumented

__running_tasks: set[str] = (source) ¶

Undocumented

__scheduler: AsyncIOScheduler | None = (source) ¶

Undocumented

__scheduler_listeners: dict[SchedulerEvent, Callable] = (source) ¶

Undocumented

__shutdown_event = (source) ¶

Undocumented

__state: ScheduleStates = (source) ¶

Undocumented

__stores: IScheduleStore = (source) ¶

Undocumented

__tasks: dict[str, TaskEntity] = (source) ¶

Undocumented

__tasks_listeners: dict[str, dict[TaskEvent, Callable]] = (source) ¶

Undocumented

__wait_to_shutdown: float = (source) ¶

Undocumented