Creating Your Own Backend

This guide walks you through creating a custom scheduler backend for AtmosPyre's LoggerScheduler.

What You'll Build

A scheduler backend that integrates a scheduling library with AtmosPyre's logging system.

Prerequisites

• Understanding of the Backend Interface
• Familiarity with your chosen scheduling library
• Python 3.10+

Implement the Following Interfaces

Classes

YourBackendTag

Bases: SchedulerDispatchTag

Tag class for [YourBackend] backend type dispatch.

This class inherits from SchedulerDispatchTag and serves as a type marker for multipledispatch, allowing functions to be dispatched to the [YourBackend] backend without requiring specific types for all arguments.

Source code in atmospyre/scheduler/_template/backend_template.py

class YourBackendTag(SchedulerDispatchTag):
    """Tag class for [YourBackend] backend type dispatch.

    This class inherits from SchedulerDispatchTag and serves as a type marker
    for multipledispatch, allowing functions to be dispatched to the [YourBackend]
    backend without requiring specific types for all arguments.

    Examples
    --------
    >>> from atmospyre.scheduler import LoggerScheduler
    >>> from atmospyre.scheduler.yourbackend.yourbackend_backend import YourBackendTag
    >>> scheduler = LoggerScheduler(scheduler_dispatch_tag=YourBackendTag())
    >>> print(scheduler)
    LoggerScheduler(backend=yourbackend, loggers=0)
    """

    pass

Functions

_create_scheduler(tag)

Create and return a [YourBackend] scheduler instance.

Parameters:

Name	Type	Description	Default
tag	YourBackendTag	Tag for dispatch (not used in implementation).	required

Returns:

Type	Description
[YourSchedulerType]	A new scheduler instance from the [your-library] library.

Source code in atmospyre/scheduler/_template/backend_template.py

@dispatch(YourBackendTag, namespace=scheduler_dispatch_namespace)
def _create_scheduler(tag: YourBackendTag):
    """Create and return a [YourBackend] scheduler instance.

    Parameters
    ----------
    tag : YourBackendTag
        Tag for dispatch (not used in implementation).

    Returns
    -------
    [YourSchedulerType]
        A new scheduler instance from the [your-library] library.

    Examples
    --------
    >>> tag = YourBackendTag()
    >>> sched = _create_scheduler(tag)
    >>> print(type(sched).__name__)
    [YourSchedulerType]
    """

_add_logger(tag, scheduler_instance, logger, interval_seconds)

Add a logger to the [YourBackend]-based scheduler.

Parameters:

Name	Type	Description	Default
tag	YourBackendTag	Tag for dispatch (not used in implementation).	required
scheduler_instance	LoggerScheduler	The scheduler instance to add the logger to.	required
logger	SensorLogger	The sensor logger to schedule.	required
interval_seconds	int	Interval in seconds between logger executions.	required

Raises:

Type	Description
ValueError	If the logger is already registered.

Source code in atmospyre/scheduler/_template/backend_template.py

@dispatch(YourBackendTag, object, object, int, namespace=scheduler_dispatch_namespace)
def _add_logger(
    tag: YourBackendTag, scheduler_instance, logger: SensorLogger, interval_seconds: int
):
    """Add a logger to the [YourBackend]-based scheduler.

    Parameters
    ----------
    tag : YourBackendTag
        Tag for dispatch (not used in implementation).
    scheduler_instance : LoggerScheduler
        The scheduler instance to add the logger to.
    logger : SensorLogger
        The sensor logger to schedule.
    interval_seconds : int
        Interval in seconds between logger executions.

    Raises
    ------
    ValueError
        If the logger is already registered.

    Examples
    --------
    >>> from atmospyre.scheduler import LoggerScheduler
    >>> from atmospyre.loggers import SensorLogger
    >>> scheduler = LoggerScheduler(scheduler_dispatch_tag=YourBackendTag())
    >>> logger = SensorLogger(sensor=my_sensor, tags=[...], interval_seconds=60, output_path=".")
    >>> scheduler.add_logger(logger)
    Added logger <SensorLogger> with interval 60s
    """

_remove_logger(tag, scheduler_instance, logger)

Remove a logger from the [YourBackend]-based scheduler.

Parameters:

Name	Type	Description	Default
tag	YourBackendTag	Tag for dispatch (not used in implementation).	required
scheduler_instance	LoggerScheduler	The scheduler instance to remove the logger from.	required
logger	SensorLogger	The logger instance to remove.	required

Returns:

Type	Description
bool	True if logger was found and removed, False otherwise.

Source code in atmospyre/scheduler/_template/backend_template.py

@dispatch(YourBackendTag, object, object, namespace=scheduler_dispatch_namespace)
def _remove_logger(
    tag: YourBackendTag, scheduler_instance, logger: SensorLogger
) -> bool:
    """Remove a logger from the [YourBackend]-based scheduler.

    Parameters
    ----------
    tag : YourBackendTag
        Tag for dispatch (not used in implementation).
    scheduler_instance : LoggerScheduler
        The scheduler instance to remove the logger from.
    logger : SensorLogger
        The logger instance to remove.

    Returns
    -------
    bool
        True if logger was found and removed, False otherwise.

    Examples
    --------
    >>> scheduler.add_logger(logger1)
    >>> scheduler.remove_logger(logger1)
    Removed logger <SensorLogger>
    True
    >>> scheduler.remove_logger(logger1)
    False
    """

_clear(tag, scheduler_instance)

Remove all loggers and clear the schedule.

Parameters:

Name	Type	Description	Default
tag	YourBackendTag	Tag for dispatch (not used in implementation).	required
scheduler_instance	LoggerScheduler	The scheduler instance to clear.	required

Source code in atmospyre/scheduler/_template/backend_template.py

@dispatch(YourBackendTag, object, namespace=scheduler_dispatch_namespace)
def _clear(tag: YourBackendTag, scheduler_instance):
    """Remove all loggers and clear the schedule.

    Parameters
    ----------
    tag : YourBackendTag
        Tag for dispatch (not used in implementation).
    scheduler_instance : LoggerScheduler
        The scheduler instance to clear.

    Examples
    --------
    >>> scheduler.add_logger(logger1)
    >>> scheduler.add_logger(logger2)
    >>> scheduler.clear()
    Cleared all loggers from scheduler
    >>> print(len(scheduler.loggers))
    0
    """

_get_status(tag, scheduler_instance)

Get current scheduler status.

Parameters:

Name	Type	Description	Default
tag	YourBackendTag	Tag for dispatch (not used in implementation).	required
scheduler_instance	LoggerScheduler	The scheduler instance to get status from.	required

Returns:

Type	Description
dict	Dictionary containing backend name, logger counts, and details. Keys include: - 'backend' (str): Name of the backend ('yourbackend') - 'total_loggers' (int): Number of registered loggers - 'total_jobs' (int): Number of scheduled jobs - 'logger_details' (list): List of dicts with logger information

Source code in atmospyre/scheduler/_template/backend_template.py

@dispatch(YourBackendTag, object, namespace=scheduler_dispatch_namespace)
def _get_status(tag: YourBackendTag, scheduler_instance) -> Dict:
    """Get current scheduler status.

    Parameters
    ----------
    tag : YourBackendTag
        Tag for dispatch (not used in implementation).
    scheduler_instance : LoggerScheduler
        The scheduler instance to get status from.

    Returns
    -------
    dict
        Dictionary containing backend name, logger counts, and details.
        Keys include:
        - 'backend' (str): Name of the backend ('yourbackend')
        - 'total_loggers' (int): Number of registered loggers
        - 'total_jobs' (int): Number of scheduled jobs
        - 'logger_details' (list): List of dicts with logger information

    Examples
    --------
    >>> scheduler.add_logger(logger1)
    >>> status = scheduler.get_status()
    >>> print(f"Backend: {status['backend']}")
    Backend: yourbackend
    >>> print(f"Total loggers: {status['total_loggers']}")
    Total loggers: 1
    >>> print(status['logger_details'][0]['interval'])
    60
    """

_run_pending(tag, scheduler_instance)

Run all pending jobs manually.

Parameters:

Name	Type	Description	Default
tag	YourBackendTag	Tag for dispatch (not used in implementation).	required
scheduler_instance	LoggerScheduler	The scheduler instance to run pending jobs on.	required

Source code in atmospyre/scheduler/_template/backend_template.py

@dispatch(YourBackendTag, object, namespace=scheduler_dispatch_namespace)
def _run_pending(tag: YourBackendTag, scheduler_instance):
    """Run all pending jobs manually.

    Parameters
    ----------
    tag : YourBackendTag
        Tag for dispatch (not used in implementation).
    scheduler_instance : LoggerScheduler
        The scheduler instance to run pending jobs on.

    Examples
    --------
    >>> import time
    >>> scheduler = LoggerScheduler(scheduler_dispatch_tag=YourBackendTag())
    >>> scheduler.add_logger(logger1)
    >>> while True:
    ...     scheduler.run_pending()
    ...     time.sleep(1)
    """

_run(tag, scheduler_instance)

Start the scheduler in a blocking loop.

This method blocks indefinitely, running scheduled jobs at the configured check interval. Press Ctrl+C to stop.

Parameters:

Name	Type	Description	Default
tag	YourBackendTag	Tag for dispatch (not used in implementation).	required
scheduler_instance	LoggerScheduler	The scheduler instance to run.	required

Source code in atmospyre/scheduler/_template/backend_template.py

@dispatch(YourBackendTag, object, namespace=scheduler_dispatch_namespace)
def _run(tag: YourBackendTag, scheduler_instance):
    """Start the scheduler in a blocking loop.

    This method blocks indefinitely, running scheduled jobs at the configured
    check interval. Press Ctrl+C to stop.

    Parameters
    ----------
    tag : YourBackendTag
        Tag for dispatch (not used in implementation).
    scheduler_instance : LoggerScheduler
        The scheduler instance to run.

    Examples
    --------
    >>> scheduler = LoggerScheduler(scheduler_dispatch_tag=YourBackendTag())
    >>> scheduler.add_logger(logger1)
    >>> scheduler.add_logger(logger2)
    >>> scheduler.run()  # Blocks until Ctrl+C
    Starting scheduler with 2 logger(s)
    Press Ctrl+C to stop
    ^C
    Scheduler stopped by user
    """