Lilota module

Lilota

High-level interface for Lilota task scheduling and worker management.

The Lilota class coordinates a scheduler instance and a pool of worker processes that execute user-defined task scripts.

A scheduler manages task persistence, distribution, and node heartbeats, while worker processes execute tasks defined inside the provided script.

Workers are started as separate Python processes that run the provided script file as a module entry point.

Source code in lilota/core.py

class Lilota:
    """High-level interface for Lilota task scheduling and worker management.

    The Lilota class coordinates a scheduler instance and a pool of worker
    processes that execute user-defined task scripts.

    A scheduler manages task persistence, distribution, and node heartbeats,
    while worker processes execute tasks defined inside the provided script.

    Workers are started as separate Python processes that run the provided
    script file as a module entry point.
    """

    @classmethod
    def scheduler(cls, db_url: str = DEFAULT_DB_URL, **kwargs) -> LilotaScheduler:
        """Create a LilotaScheduler instance.

        Args:
          db_url (str): Database connection URL.
          **kwargs: Additional keyword arguments passed to the constructor.

        Returns:
          LilotaScheduler: Configured LilotaScheduler instance.
        """
        lilota = cls(
            mode=LilotaMode.SCHEDULER,
            db_url=db_url,
            script_path=None,
            number_of_workers=0,
            **kwargs,
        )
        return lilota._scheduler

    @classmethod
    def workers(
        cls,
        script_path: str,
        number_of_workers: int,
        db_url: str = DEFAULT_DB_URL,
        **kwargs,
    ):
        """Create a Lilota instance running only worker processes.

        Args:
          db_url (str): Database connection URL.
          script_path (str): Path to the worker script.
          number_of_workers (int): Number of worker processes to spawn.
          **kwargs: Additional keyword arguments passed to the constructor.

        Returns:
          Lilota: Configured instance in worker-only mode.
        """
        return cls(
            mode=LilotaMode.WORKERS,
            db_url=db_url,
            script_path=script_path,
            number_of_workers=number_of_workers,
            **kwargs,
        )

    def __init__(
        self,
        db_url: str = DEFAULT_DB_URL,
        *,
        mode: LilotaMode = LilotaMode.ALL,
        script_path: str = None,
        number_of_workers: int = cpu_count(),
        scheduler_heartbeat_interval: float = 5,
        scheduler_timeout_sec: int = 20,
        process_manager_interval: float = 1.0,
        stop_worker_timeout: int = 60,
        logging_level=logging.INFO,
    ):
        """Create a new Lilota runtime instance.

        Args:
          db_url (str):
            Database connection URL used by the scheduler.

          mode (LilotaMode, optional):
            Execution mode controlling whether the scheduler, workers,
            or both are started. Defaults to ALL.

          script_path (str):
            Path to a Python script that defines and registers Lilota tasks.
            Each worker process executes this script as its entry point.

          number_of_workers (int, optional):
            Number of worker processes to spawn. Defaults to the number of CPU cores and cannot exceed this value * 5.

          scheduler_heartbeat_interval (float, optional):
            Interval in seconds between scheduler node heartbeats.
            Defaults to 5 seconds.

          scheduler_timeout_sec (int, optional):
            Time in seconds after which a node is considered inactive
            if no heartbeat is received. Defaults to 20 seconds.

          stop_worker_timeout (int, optional):
            Maximum time in seconds to wait for worker processes to
            exit gracefully before they are forcefully killed.
            Defaults to 60 seconds.

          logging_level (int, optional):
            Logging level used by the scheduler. Defaults to logging.INFO.
        """
        if mode in (LilotaMode.WORKERS, LilotaMode.ALL):
            if number_of_workers <= 0:
                raise ValueError("number_of_workers must be > 0 in WORKER or ALL mode")

            number_of_cpus = cpu_count()
            max_workers = number_of_cpus * 5
            if number_of_workers > max_workers:
                raise ValueError(
                    f"number_of_workers cannot be greater than {max_workers}"
                )

        self._db_url = db_url
        self._mode = mode
        self._script_path = script_path
        self._number_of_workers = number_of_workers
        self._process_manager_interval = process_manager_interval
        self._stop_worker_timeout = stop_worker_timeout
        self._process_manager_heartbeat: Heartbeat = None
        self._process_manager = ProcessManager(on_fatal_error=self._handle_fatal_error)
        self._error = None
        self._error_event = threading.Event()
        atexit.register(self._process_manager.stop_all)

        if mode in (LilotaMode.SCHEDULER, LilotaMode.ALL):
            self._scheduler = LilotaScheduler(
                db_url=db_url,
                node_heartbeat_interval=scheduler_heartbeat_interval,
                node_timeout_sec=scheduler_timeout_sec,
                logging_level=logging_level,
            )

        self._node_store = NodeStore(self._db_url, None)
        self._task_store = TaskStore(self._db_url, None)
        self._is_started = False

    @property
    def error(self):
        """Return the last fatal error encountered by the system.

        Returns:
          Any: The error object or message captured from a worker process,
          or None if no error has occurred.
        """
        return self._error

    def has_failed(self) -> bool:
        """Check whether a fatal error has occurred.

        Returns:
          bool: True if a fatal error has been detected, otherwise False.
        """
        return self._error_event.is_set()

    def wait_for_failure(self, timeout=None):
        """Block until a fatal error occurs or a timeout is reached.

        Args:
          timeout (float | None, optional): Maximum time in seconds to wait.
            If None, waits indefinitely.

        Returns:
          bool: True if a failure occurred before the timeout, False otherwise.
        """
        return self._error_event.wait(timeout)

    def _handle_fatal_error(self, error):
        self._error = error
        self._error_event.set()

    def start(self):
        """Start the configured Lilota components.

        Depending on the selected mode, this method:
          - Starts the scheduler (SCHEDULER, ALL)
          - Spawns worker processes (WORKER, ALL)
          - Starts process monitoring for workers (WORKER, ALL)

        After startup, the instance is ready to schedule and/or execute tasks.
        """
        if self._is_started:
            return

        if self._mode in (LilotaMode.SCHEDULER, LilotaMode.ALL):
            self._start_scheduler()

        if self._mode in (LilotaMode.WORKERS, LilotaMode.ALL):
            if not self._script_path:
                raise ValueError("script_path must be provided in WORKER or ALL mode")
            self._start_workers()
            self._start_process_manager()

        self._is_started = True

    def _start_scheduler(self):
        self._scheduler.start()

    def _start_workers(self):
        # Start scripts
        for _ in range(self._number_of_workers):
            self._process_manager.start(self._script_path)

    def _start_process_manager(self):
        # Start the process manager
        process_manager_task = ProcessManagerTask(
            interval=self._process_manager_interval,
            process_manager=self._process_manager,
        )
        self._process_manager_heartbeat = Heartbeat(
            "process_manager", process_manager_task, None
        )
        self._process_manager_heartbeat.start()

    def stop(self):
        """Stop all running Lilota components.

        This includes:
          - Stopping the scheduler (if running)
          - Stopping process monitoring
          - Gracefully terminating worker processes

        Worker processes are first asked to terminate gracefully. If a worker
        does not exit within ``stop_worker_timeout`` seconds, it is forcefully killed.
        """
        if not self._is_started:
            return

        if self._mode in (LilotaMode.SCHEDULER, LilotaMode.ALL):
            self._stop_scheduler()

        if self._mode in (LilotaMode.WORKERS, LilotaMode.ALL):
            self._stop_process_manager()
            self._stop_workers()

        self._is_started = False

    def _stop_scheduler(self):
        self._scheduler.stop()

    def _stop_workers(self):
        self._process_manager.stop_all(self._stop_worker_timeout)

    def _stop_process_manager(self):
        if self._process_manager_heartbeat:
            self._process_manager_heartbeat.stop_and_join(
                timeout=self._stop_worker_timeout
            )
            self._process_manager_heartbeat = None

    def schedule(self, name: str, input: Any = None) -> int:
        """Schedule a task for execution.

        Args:
          name (str): Name of the registered task.
          input (Any, optional): Input payload for the task. Defaults to None.

        Returns:
          int: Identifier of the scheduled task.
        """
        return self._scheduler.schedule(name, input)

    def get_all_nodes(self) -> list[Node]:
        """Retrieve all registered scheduler nodes.

        Returns:
            list[Node]: A list containing all nodes currently stored in the
            node store.
        """
        return self._node_store.get_all_nodes()

    def get_task_by_id(self, id: UUID):
        """Retrieve a task by its unique identifier.

        Args:
          id (UUID): Unique task identifier.

        Returns:
          Any: Task object associated with the given ID.
        """
        return self._task_store.get_task_by_id(id)

error property

Return the last fatal error encountered by the system.

Returns:

Name	Type	Description
Any		The error object or message captured from a worker process,
		or None if no error has occurred.

__init__(db_url=DEFAULT_DB_URL, *, mode=LilotaMode.ALL, script_path=None, number_of_workers=cpu_count(), scheduler_heartbeat_interval=5, scheduler_timeout_sec=20, process_manager_interval=1.0, stop_worker_timeout=60, logging_level=logging.INFO)

Create a new Lilota runtime instance.

Parameters:

Name	Type	Description	Default
db_url	str	Database connection URL used by the scheduler.	DEFAULT_DB_URL
mode	LilotaMode	Execution mode controlling whether the scheduler, workers, or both are started. Defaults to ALL.	ALL
script_path	str	Path to a Python script that defines and registers Lilota tasks. Each worker process executes this script as its entry point.	None
number_of_workers	int	Number of worker processes to spawn. Defaults to the number of CPU cores and cannot exceed this value * 5.	cpu_count()
scheduler_heartbeat_interval	float	Interval in seconds between scheduler node heartbeats. Defaults to 5 seconds.	5
scheduler_timeout_sec	int	Time in seconds after which a node is considered inactive if no heartbeat is received. Defaults to 20 seconds.	20
stop_worker_timeout	int	Maximum time in seconds to wait for worker processes to exit gracefully before they are forcefully killed. Defaults to 60 seconds.	60
logging_level	int	Logging level used by the scheduler. Defaults to logging.INFO.	INFO

Source code in lilota/core.py

def __init__(
    self,
    db_url: str = DEFAULT_DB_URL,
    *,
    mode: LilotaMode = LilotaMode.ALL,
    script_path: str = None,
    number_of_workers: int = cpu_count(),
    scheduler_heartbeat_interval: float = 5,
    scheduler_timeout_sec: int = 20,
    process_manager_interval: float = 1.0,
    stop_worker_timeout: int = 60,
    logging_level=logging.INFO,
):
    """Create a new Lilota runtime instance.

    Args:
      db_url (str):
        Database connection URL used by the scheduler.

      mode (LilotaMode, optional):
        Execution mode controlling whether the scheduler, workers,
        or both are started. Defaults to ALL.

      script_path (str):
        Path to a Python script that defines and registers Lilota tasks.
        Each worker process executes this script as its entry point.

      number_of_workers (int, optional):
        Number of worker processes to spawn. Defaults to the number of CPU cores and cannot exceed this value * 5.

      scheduler_heartbeat_interval (float, optional):
        Interval in seconds between scheduler node heartbeats.
        Defaults to 5 seconds.

      scheduler_timeout_sec (int, optional):
        Time in seconds after which a node is considered inactive
        if no heartbeat is received. Defaults to 20 seconds.

      stop_worker_timeout (int, optional):
        Maximum time in seconds to wait for worker processes to
        exit gracefully before they are forcefully killed.
        Defaults to 60 seconds.

      logging_level (int, optional):
        Logging level used by the scheduler. Defaults to logging.INFO.
    """
    if mode in (LilotaMode.WORKERS, LilotaMode.ALL):
        if number_of_workers <= 0:
            raise ValueError("number_of_workers must be > 0 in WORKER or ALL mode")

        number_of_cpus = cpu_count()
        max_workers = number_of_cpus * 5
        if number_of_workers > max_workers:
            raise ValueError(
                f"number_of_workers cannot be greater than {max_workers}"
            )

    self._db_url = db_url
    self._mode = mode
    self._script_path = script_path
    self._number_of_workers = number_of_workers
    self._process_manager_interval = process_manager_interval
    self._stop_worker_timeout = stop_worker_timeout
    self._process_manager_heartbeat: Heartbeat = None
    self._process_manager = ProcessManager(on_fatal_error=self._handle_fatal_error)
    self._error = None
    self._error_event = threading.Event()
    atexit.register(self._process_manager.stop_all)

    if mode in (LilotaMode.SCHEDULER, LilotaMode.ALL):
        self._scheduler = LilotaScheduler(
            db_url=db_url,
            node_heartbeat_interval=scheduler_heartbeat_interval,
            node_timeout_sec=scheduler_timeout_sec,
            logging_level=logging_level,
        )

    self._node_store = NodeStore(self._db_url, None)
    self._task_store = TaskStore(self._db_url, None)
    self._is_started = False

get_all_nodes()

Retrieve all registered scheduler nodes.

Returns:

Type	Description
list[Node]	list[Node]: A list containing all nodes currently stored in the
list[Node]	node store.

Source code in lilota/core.py

def get_all_nodes(self) -> list[Node]:
    """Retrieve all registered scheduler nodes.

    Returns:
        list[Node]: A list containing all nodes currently stored in the
        node store.
    """
    return self._node_store.get_all_nodes()

get_task_by_id(id)

Retrieve a task by its unique identifier.

Parameters:

Name	Type	Description	Default
id	UUID	Unique task identifier.	required

Returns:

Name	Type	Description
Any		Task object associated with the given ID.

Source code in lilota/core.py

def get_task_by_id(self, id: UUID):
    """Retrieve a task by its unique identifier.

    Args:
      id (UUID): Unique task identifier.

    Returns:
      Any: Task object associated with the given ID.
    """
    return self._task_store.get_task_by_id(id)

has_failed()

Check whether a fatal error has occurred.

Returns:

Name	Type	Description
bool	bool	True if a fatal error has been detected, otherwise False.

Source code in lilota/core.py

def has_failed(self) -> bool:
    """Check whether a fatal error has occurred.

    Returns:
      bool: True if a fatal error has been detected, otherwise False.
    """
    return self._error_event.is_set()

schedule(name, input=None)

Schedule a task for execution.

Parameters:

Name	Type	Description	Default
name	str	Name of the registered task.	required
input	Any	Input payload for the task. Defaults to None.	None

Returns:

Name	Type	Description
int	int	Identifier of the scheduled task.

Source code in lilota/core.py

def schedule(self, name: str, input: Any = None) -> int:
    """Schedule a task for execution.

    Args:
      name (str): Name of the registered task.
      input (Any, optional): Input payload for the task. Defaults to None.

    Returns:
      int: Identifier of the scheduled task.
    """
    return self._scheduler.schedule(name, input)

scheduler(db_url=DEFAULT_DB_URL, **kwargs) classmethod

Create a LilotaScheduler instance.

Parameters:

Name	Type	Description	Default
db_url	str	Database connection URL.	DEFAULT_DB_URL
**kwargs		Additional keyword arguments passed to the constructor.	{}

Returns:

Name	Type	Description
LilotaScheduler	LilotaScheduler	Configured LilotaScheduler instance.

Source code in lilota/core.py

@classmethod
def scheduler(cls, db_url: str = DEFAULT_DB_URL, **kwargs) -> LilotaScheduler:
    """Create a LilotaScheduler instance.

    Args:
      db_url (str): Database connection URL.
      **kwargs: Additional keyword arguments passed to the constructor.

    Returns:
      LilotaScheduler: Configured LilotaScheduler instance.
    """
    lilota = cls(
        mode=LilotaMode.SCHEDULER,
        db_url=db_url,
        script_path=None,
        number_of_workers=0,
        **kwargs,
    )
    return lilota._scheduler

start()

Start the configured Lilota components.

Depending on the selected mode, this method: - Starts the scheduler (SCHEDULER, ALL) - Spawns worker processes (WORKER, ALL) - Starts process monitoring for workers (WORKER, ALL)

After startup, the instance is ready to schedule and/or execute tasks.

Source code in lilota/core.py

def start(self):
    """Start the configured Lilota components.

    Depending on the selected mode, this method:
      - Starts the scheduler (SCHEDULER, ALL)
      - Spawns worker processes (WORKER, ALL)
      - Starts process monitoring for workers (WORKER, ALL)

    After startup, the instance is ready to schedule and/or execute tasks.
    """
    if self._is_started:
        return

    if self._mode in (LilotaMode.SCHEDULER, LilotaMode.ALL):
        self._start_scheduler()

    if self._mode in (LilotaMode.WORKERS, LilotaMode.ALL):
        if not self._script_path:
            raise ValueError("script_path must be provided in WORKER or ALL mode")
        self._start_workers()
        self._start_process_manager()

    self._is_started = True

stop()

Stop all running Lilota components.

This includes

• Stopping the scheduler (if running)
• Stopping process monitoring
• Gracefully terminating worker processes

Worker processes are first asked to terminate gracefully. If a worker does not exit within stop_worker_timeout seconds, it is forcefully killed.

Source code in lilota/core.py

def stop(self):
    """Stop all running Lilota components.

    This includes:
      - Stopping the scheduler (if running)
      - Stopping process monitoring
      - Gracefully terminating worker processes

    Worker processes are first asked to terminate gracefully. If a worker
    does not exit within ``stop_worker_timeout`` seconds, it is forcefully killed.
    """
    if not self._is_started:
        return

    if self._mode in (LilotaMode.SCHEDULER, LilotaMode.ALL):
        self._stop_scheduler()

    if self._mode in (LilotaMode.WORKERS, LilotaMode.ALL):
        self._stop_process_manager()
        self._stop_workers()

    self._is_started = False

wait_for_failure(timeout=None)

Block until a fatal error occurs or a timeout is reached.

Parameters:

Name	Type	Description	Default
timeout	float | None	Maximum time in seconds to wait. If None, waits indefinitely.	None

Returns:

Name	Type	Description
bool		True if a failure occurred before the timeout, False otherwise.

Source code in lilota/core.py

def wait_for_failure(self, timeout=None):
    """Block until a fatal error occurs or a timeout is reached.

    Args:
      timeout (float | None, optional): Maximum time in seconds to wait.
        If None, waits indefinitely.

    Returns:
      bool: True if a failure occurred before the timeout, False otherwise.
    """
    return self._error_event.wait(timeout)

workers(script_path, number_of_workers, db_url=DEFAULT_DB_URL, **kwargs) classmethod

Create a Lilota instance running only worker processes.

Parameters:

Name	Type	Description	Default
db_url	str	Database connection URL.	DEFAULT_DB_URL
script_path	str	Path to the worker script.	required
number_of_workers	int	Number of worker processes to spawn.	required
**kwargs		Additional keyword arguments passed to the constructor.	{}

Returns:

Name	Type	Description
Lilota		Configured instance in worker-only mode.

Source code in lilota/core.py

@classmethod
def workers(
    cls,
    script_path: str,
    number_of_workers: int,
    db_url: str = DEFAULT_DB_URL,
    **kwargs,
):
    """Create a Lilota instance running only worker processes.

    Args:
      db_url (str): Database connection URL.
      script_path (str): Path to the worker script.
      number_of_workers (int): Number of worker processes to spawn.
      **kwargs: Additional keyword arguments passed to the constructor.

    Returns:
      Lilota: Configured instance in worker-only mode.
    """
    return cls(
        mode=LilotaMode.WORKERS,
        db_url=db_url,
        script_path=script_path,
        number_of_workers=number_of_workers,
        **kwargs,
    )

LilotaMode

Bases: str, Enum

Execution mode for the Lilota runtime.

Attributes:

Name	Type	Description
SCHEDULER		Run only the scheduler component.
WORKERS		Run only worker processes.
ALL		Run both scheduler and workers.

Source code in lilota/core.py

class LilotaMode(str, Enum):
    """Execution mode for the Lilota runtime.

    Attributes:
      SCHEDULER: Run only the scheduler component.
      WORKERS: Run only worker processes.
      ALL: Run both scheduler and workers.
    """

    SCHEDULER = "scheduler"
    WORKERS = "workers"
    ALL = "all"

ProcessManager

Source code in lilota/core.py

class ProcessManager:
    def __init__(self, on_fatal_error=None):
        """Initializes the ProcessManager.

        Args:
          on_fatal_error (Callable[[str], None] | None): Optional callback that is
            invoked when a managed process exits with an error. The callback
            receives the stderr output of the failed process.
        """
        self._on_fatal_error = on_fatal_error
        self._processes: List[ManagedProcess] = []

    def start(self, script_path: str) -> subprocess.Popen:
        """Starts a new subprocess running the given Python script.

        The process is launched using the current Python interpreter and added
        to the internal list of managed processes.

        Args:
          script_path (str): Path to the Python script to execute.

        Returns:
          subprocess.Popen: The Popen object representing the started process.
        """
        proc = subprocess.Popen(
            [sys.executable, script_path],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            preexec_fn=os.setsid if os.name != "nt" else None,
            creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0,
        )

        self._processes.append(ManagedProcess(proc, script_path))
        return proc

    def monitor(self):
        """Monitors managed processes for termination and handles restarts.

        Iterates over all managed processes and checks whether they have exited.
        If a process exits with a non-zero return code and produces stderr output,
        all processes are stopped and the fatal error callback is invoked.

        If a process exits successfully, it is automatically restarted using
        the original script path.
        """
        for mp in list(self._processes):
            proc = mp.proc
            returncode = proc.poll()

            if returncode is not None:  # process exited
                # This cannot be debugged. Use print statements because if the worker is killed also the debugger stops.
                stdout, stderr = proc.communicate()

                # Error handling in case of an exception
                # = 0: Process exited successfully
                # > 0:Process exited with an error
                # < 0: Process was terminated by a signal (Unix only)
                if returncode != 0 and stderr:
                    self._handle_error(stderr)
                    return

                # Restart process
                self._processes.remove(mp)
                self.start(mp.script_path)

    def _handle_error(self, stderr: str):
        # Stop all processes
        self.stop_all()

        # Signal to caller
        if self._on_fatal_error:
            self._on_fatal_error(stderr)

    def stop_all(self, timeout: float = 5.0):
        """Stops all managed processes gracefully, then forcefully if needed.

        Attempts to terminate all running processes. If a process does not exit
        within the specified timeout, it is forcefully killed. After stopping,
        the internal process list is cleared.

        Args:
          timeout (float): Maximum time in seconds to wait for each process to
            terminate gracefully before forcefully killing it. Defaults to 5.0.
        """
        for mp in self._processes:
            proc = mp.proc
            if proc.poll() is None:
                self._terminate(proc)

        for mp in self._processes:
            proc = mp.proc
            try:
                proc.wait(timeout=timeout)
            except subprocess.TimeoutExpired:
                self._kill(proc)

        self._processes.clear()

    def _terminate(self, proc):
        if os.name == "nt":
            proc.send_signal(signal.CTRL_BREAK_EVENT)
        else:
            os.killpg(os.getpgid(proc.pid), signal.SIGTERM)

    def _kill(self, proc):
        if os.name == "nt":
            proc.kill()
        else:
            os.killpg(os.getpgid(proc.pid), signal.SIGKILL)

__init__(on_fatal_error=None)

Initializes the ProcessManager.

Parameters:

Name	Type	Description	Default
on_fatal_error	Callable[[str], None] | None	Optional callback that is invoked when a managed process exits with an error. The callback receives the stderr output of the failed process.	None

Source code in lilota/core.py

def __init__(self, on_fatal_error=None):
    """Initializes the ProcessManager.

    Args:
      on_fatal_error (Callable[[str], None] | None): Optional callback that is
        invoked when a managed process exits with an error. The callback
        receives the stderr output of the failed process.
    """
    self._on_fatal_error = on_fatal_error
    self._processes: List[ManagedProcess] = []

monitor()

Monitors managed processes for termination and handles restarts.

Iterates over all managed processes and checks whether they have exited. If a process exits with a non-zero return code and produces stderr output, all processes are stopped and the fatal error callback is invoked.

If a process exits successfully, it is automatically restarted using the original script path.

Source code in lilota/core.py

def monitor(self):
    """Monitors managed processes for termination and handles restarts.

    Iterates over all managed processes and checks whether they have exited.
    If a process exits with a non-zero return code and produces stderr output,
    all processes are stopped and the fatal error callback is invoked.

    If a process exits successfully, it is automatically restarted using
    the original script path.
    """
    for mp in list(self._processes):
        proc = mp.proc
        returncode = proc.poll()

        if returncode is not None:  # process exited
            # This cannot be debugged. Use print statements because if the worker is killed also the debugger stops.
            stdout, stderr = proc.communicate()

            # Error handling in case of an exception
            # = 0: Process exited successfully
            # > 0:Process exited with an error
            # < 0: Process was terminated by a signal (Unix only)
            if returncode != 0 and stderr:
                self._handle_error(stderr)
                return

            # Restart process
            self._processes.remove(mp)
            self.start(mp.script_path)

start(script_path)

Starts a new subprocess running the given Python script.

The process is launched using the current Python interpreter and added to the internal list of managed processes.

Parameters:

Name	Type	Description	Default
script_path	str	Path to the Python script to execute.	required

Returns:

Type	Description
Popen	subprocess.Popen: The Popen object representing the started process.

Source code in lilota/core.py

def start(self, script_path: str) -> subprocess.Popen:
    """Starts a new subprocess running the given Python script.

    The process is launched using the current Python interpreter and added
    to the internal list of managed processes.

    Args:
      script_path (str): Path to the Python script to execute.

    Returns:
      subprocess.Popen: The Popen object representing the started process.
    """
    proc = subprocess.Popen(
        [sys.executable, script_path],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        preexec_fn=os.setsid if os.name != "nt" else None,
        creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if os.name == "nt" else 0,
    )

    self._processes.append(ManagedProcess(proc, script_path))
    return proc

stop_all(timeout=5.0)

Stops all managed processes gracefully, then forcefully if needed.

Attempts to terminate all running processes. If a process does not exit within the specified timeout, it is forcefully killed. After stopping, the internal process list is cleared.

Parameters:

Name	Type	Description	Default
timeout	float	Maximum time in seconds to wait for each process to terminate gracefully before forcefully killing it. Defaults to 5.0.	5.0

Source code in lilota/core.py

def stop_all(self, timeout: float = 5.0):
    """Stops all managed processes gracefully, then forcefully if needed.

    Attempts to terminate all running processes. If a process does not exit
    within the specified timeout, it is forcefully killed. After stopping,
    the internal process list is cleared.

    Args:
      timeout (float): Maximum time in seconds to wait for each process to
        terminate gracefully before forcefully killing it. Defaults to 5.0.
    """
    for mp in self._processes:
        proc = mp.proc
        if proc.poll() is None:
            self._terminate(proc)

    for mp in self._processes:
        proc = mp.proc
        try:
            proc.wait(timeout=timeout)
        except subprocess.TimeoutExpired:
            self._kill(proc)

    self._processes.clear()

LilotaNode

Bases: ABC

Abstract base class for all Lilota nodes.

A node represents a running component in the Lilota system, such as a scheduler or a worker. This class handles common functionality including:

• database initialization and migrations
• node lifecycle management
• node status updates
• heartbeat management
• access to node and task stores

Source code in lilota/node.py

class LilotaNode(ABC):
    """Abstract base class for all Lilota nodes.

    A node represents a running component in the Lilota system, such as a
    scheduler or a worker. This class handles common functionality including:

    - database initialization and migrations
    - node lifecycle management
    - node status updates
    - heartbeat management
    - access to node and task stores
    """

    def __init__(
        self,
        *,
        db_url: str,
        node_type: NodeType,
        node_heartbeat_interval: float,
        node_timeout_sec: int,
        logger_name: str,
        logging_level,
    ):
        """Initialize a Lilota node.

        Args:
          db_url (str): Database connection URL.
          node_type (NodeType): Type of node (`scheduler` or `worker`).
          node_heartbeat_interval (float): Interval in seconds between node
            heartbeat updates.
          node_timeout_sec (int): Time in seconds after which a node is
            considered inactive if no heartbeat is received.
          logger_name (str): Name of the logger used by the node.
          logging_level (int): Logging level used for the node logger.
        """

        self._db_url = db_url
        self._node_type = node_type
        self._node_heartbeat_interval = node_heartbeat_interval
        self._node_heartbeat_join_timeout_in_sec = 60
        self._node_timeout_sec = node_timeout_sec
        self._node_id = None
        self._node_store = None
        self._task_store = None
        self._heartbeat: Heartbeat = None
        self._logging_level = logging_level
        self._is_started = False

        # Upgrade the database
        upgrade_db(self._db_url)

        # Setup logging
        self._logger = configure_logging(self._db_url, logger_name, logging_level)

    def start(self):
        """Start the node.

        Initializes the node in the database, sets its status to ``RUNNING``,
        and triggers the node-specific startup logic implemented in
        ``_on_started()``.

        Raises:
          Exception: If the node is already started.
        """

        # Check if the node is already started
        if self._is_started:
            raise Exception("The node is already started")

        if not self._node_id:
            # Create stores
            self._node_store = NodeStore(self._db_url, self._logger)
            self._task_store = TaskStore(self._db_url, self._logger)

            # Create node with status IDLE
            self._node_id = self._node_store.create_node(
                self._node_type, NodeStatus.RUNNING
            )

            # Create a context logger
            self._logger = create_context_logger(self._logger, node_id=self._node_id)
        else:
            # Change status to RUNNING
            self._node_store.update_node_status(self._node_id, NodeStatus.RUNNING)

        # Set the node as started
        self._is_started = True

        # On started
        self._on_started()

    def stop(self):
        """Stop the node.

        Updates the node status to ``STOPPING``, executes subclass-specific
        shutdown logic via ``_on_stop()``, and then sets the node status to
        ``IDLE``.

        Raises:
          Exception: If the node was not started.
        """

        # Check if the node was started
        if not self._is_started:
            raise Exception("The node cannot be stopped because it was not started")

        # Stop node heartbeat
        self._stop_node_heartbeat()

        # Change status to IDLE
        self._node_store.update_node_status(self._node_id, NodeStatus.IDLE)

        # Log Node stopped message
        self._logger.debug("Node stopped")

        # Set the node as not started
        self._is_started = False

    def get_nodes(self):
        """Return all nodes currently registered in the system.

        Returns:
          list: A list of node records.
        """
        return self._node_store.get_all_nodes()

    def get_node(self):
        """Return the current node record.

        Returns:
          Any | None: The node record if the node has been created,
          otherwise ``None``.
        """
        return self._node_store.get_node_by_id(self._node_id) if self._node_id else None

    def get_task_by_id(self, id: UUID):
        """Retrieve a task by its identifier.

        Args:
          id (UUID): Unique identifier of the task.

        Returns:
          Any: Task record associated with the given ID.
        """
        return self._task_store.get_task_by_id(id)

    def delete_task_by_id(self, id: UUID):
        """Delete a task from the system.

        Args:
          id (UUID): Unique identifier of the task.

        Returns:
          bool: True if the task was deleted, otherwise False.
        """
        self._logger.debug(f"Delete task with id {id}")
        success = self._task_store.delete_task_by_id(id)
        if success:
            self._logger.debug("Task deleted!")
        else:
            self._logger.debug("Task not deleted!")
        return success

    def _stop_node_heartbeat(self):
        # Stop heartbeat thread
        if self._heartbeat:
            self._heartbeat.stop_and_join(
                timeout=self._node_heartbeat_join_timeout_in_sec
            )
            self._heartbeat = None

    @abstractmethod
    def _on_started(self):
        """Hook executed after the node has successfully started.

        Subclasses should implement this method to start any required
        background threads, heartbeats, or task loops.
        """
        pass

__init__(*, db_url, node_type, node_heartbeat_interval, node_timeout_sec, logger_name, logging_level)

Initialize a Lilota node.

Parameters:

Name	Type	Description	Default
db_url	str	Database connection URL.	required
node_type	NodeType	Type of node (scheduler or worker).	required
node_heartbeat_interval	float	Interval in seconds between node heartbeat updates.	required
node_timeout_sec	int	Time in seconds after which a node is considered inactive if no heartbeat is received.	required
logger_name	str	Name of the logger used by the node.	required
logging_level	int	Logging level used for the node logger.	required

Source code in lilota/node.py

def __init__(
    self,
    *,
    db_url: str,
    node_type: NodeType,
    node_heartbeat_interval: float,
    node_timeout_sec: int,
    logger_name: str,
    logging_level,
):
    """Initialize a Lilota node.

    Args:
      db_url (str): Database connection URL.
      node_type (NodeType): Type of node (`scheduler` or `worker`).
      node_heartbeat_interval (float): Interval in seconds between node
        heartbeat updates.
      node_timeout_sec (int): Time in seconds after which a node is
        considered inactive if no heartbeat is received.
      logger_name (str): Name of the logger used by the node.
      logging_level (int): Logging level used for the node logger.
    """

    self._db_url = db_url
    self._node_type = node_type
    self._node_heartbeat_interval = node_heartbeat_interval
    self._node_heartbeat_join_timeout_in_sec = 60
    self._node_timeout_sec = node_timeout_sec
    self._node_id = None
    self._node_store = None
    self._task_store = None
    self._heartbeat: Heartbeat = None
    self._logging_level = logging_level
    self._is_started = False

    # Upgrade the database
    upgrade_db(self._db_url)

    # Setup logging
    self._logger = configure_logging(self._db_url, logger_name, logging_level)

delete_task_by_id(id)

Delete a task from the system.

Parameters:

Name	Type	Description	Default
id	UUID	Unique identifier of the task.	required

Returns:

Name	Type	Description
bool		True if the task was deleted, otherwise False.

Source code in lilota/node.py

def delete_task_by_id(self, id: UUID):
    """Delete a task from the system.

    Args:
      id (UUID): Unique identifier of the task.

    Returns:
      bool: True if the task was deleted, otherwise False.
    """
    self._logger.debug(f"Delete task with id {id}")
    success = self._task_store.delete_task_by_id(id)
    if success:
        self._logger.debug("Task deleted!")
    else:
        self._logger.debug("Task not deleted!")
    return success

get_node()

Return the current node record.

Returns:

Type	Description
	Any | None: The node record if the node has been created,
	otherwise None.

Source code in lilota/node.py

def get_node(self):
    """Return the current node record.

    Returns:
      Any | None: The node record if the node has been created,
      otherwise ``None``.
    """
    return self._node_store.get_node_by_id(self._node_id) if self._node_id else None

get_nodes()

Return all nodes currently registered in the system.

Returns:

Name	Type	Description
list		A list of node records.

Source code in lilota/node.py

def get_nodes(self):
    """Return all nodes currently registered in the system.

    Returns:
      list: A list of node records.
    """
    return self._node_store.get_all_nodes()

get_task_by_id(id)

Retrieve a task by its identifier.

Parameters:

Name	Type	Description	Default
id	UUID	Unique identifier of the task.	required

Returns:

Name	Type	Description
Any		Task record associated with the given ID.

Source code in lilota/node.py

def get_task_by_id(self, id: UUID):
    """Retrieve a task by its identifier.

    Args:
      id (UUID): Unique identifier of the task.

    Returns:
      Any: Task record associated with the given ID.
    """
    return self._task_store.get_task_by_id(id)

start()

Start the node.

Initializes the node in the database, sets its status to RUNNING, and triggers the node-specific startup logic implemented in _on_started().

Raises:

Type	Description
Exception	If the node is already started.

Source code in lilota/node.py

def start(self):
    """Start the node.

    Initializes the node in the database, sets its status to ``RUNNING``,
    and triggers the node-specific startup logic implemented in
    ``_on_started()``.

    Raises:
      Exception: If the node is already started.
    """

    # Check if the node is already started
    if self._is_started:
        raise Exception("The node is already started")

    if not self._node_id:
        # Create stores
        self._node_store = NodeStore(self._db_url, self._logger)
        self._task_store = TaskStore(self._db_url, self._logger)

        # Create node with status IDLE
        self._node_id = self._node_store.create_node(
            self._node_type, NodeStatus.RUNNING
        )

        # Create a context logger
        self._logger = create_context_logger(self._logger, node_id=self._node_id)
    else:
        # Change status to RUNNING
        self._node_store.update_node_status(self._node_id, NodeStatus.RUNNING)

    # Set the node as started
    self._is_started = True

    # On started
    self._on_started()

stop()

Stop the node.

Updates the node status to STOPPING, executes subclass-specific shutdown logic via _on_stop(), and then sets the node status to IDLE.

Raises:

Type	Description
Exception	If the node was not started.

Source code in lilota/node.py

def stop(self):
    """Stop the node.

    Updates the node status to ``STOPPING``, executes subclass-specific
    shutdown logic via ``_on_stop()``, and then sets the node status to
    ``IDLE``.

    Raises:
      Exception: If the node was not started.
    """

    # Check if the node was started
    if not self._is_started:
        raise Exception("The node cannot be stopped because it was not started")

    # Stop node heartbeat
    self._stop_node_heartbeat()

    # Change status to IDLE
    self._node_store.update_node_status(self._node_id, NodeStatus.IDLE)

    # Log Node stopped message
    self._logger.debug("Node stopped")

    # Set the node as not started
    self._is_started = False

NodeHeartbeatTask

Bases: HeartbeatTask

Heartbeat task used by Lilota nodes.

This task periodically updates the last_seen_at timestamp of the associated node in the database to indicate that the node is still alive.

Source code in lilota/node.py

class NodeHeartbeatTask(HeartbeatTask):
    """Heartbeat task used by Lilota nodes.

    This task periodically updates the ``last_seen_at`` timestamp of the
    associated node in the database to indicate that the node is still alive.
    """

    def __init__(
        self,
        interval: float,
        jitter: float,
        node_id: str,
        node_store: NodeStore,
        logger: logging.Logger,
    ):
        """Initialize the heartbeat task.

        Args:
          interval (float): Interval in seconds between heartbeat executions.
          jitter (float): Jitter that is used for the heartbeat interval.
          node_id (str): Unique identifier of the node.
          node_store (SqlAlchemyNodeStore): Store used for node operations.
          logger (logging.Logger): Logger instance used for reporting errors.
        """
        super().__init__(interval, jitter)
        self._node_id = node_id
        self._node_store = node_store
        self._logger = logger

    def execute(self):
        """Execute the heartbeat update.

        Updates the ``last_seen_at`` field of the node in the database.
        Any errors are logged without interrupting the heartbeat loop.
        """
        try:
            self._node_store.update_node_last_seen_at(self._node_id)
        except Exception:
            self._logger.exception(
                f"Heartbeat update failed for node_id '{self._node_id}'"
            )

__init__(interval, jitter, node_id, node_store, logger)

Initialize the heartbeat task.

Parameters:

Name	Type	Description	Default
interval	float	Interval in seconds between heartbeat executions.	required
jitter	float	Jitter that is used for the heartbeat interval.	required
node_id	str	Unique identifier of the node.	required
node_store	SqlAlchemyNodeStore	Store used for node operations.	required
logger	Logger	Logger instance used for reporting errors.	required

Source code in lilota/node.py

def __init__(
    self,
    interval: float,
    jitter: float,
    node_id: str,
    node_store: NodeStore,
    logger: logging.Logger,
):
    """Initialize the heartbeat task.

    Args:
      interval (float): Interval in seconds between heartbeat executions.
      jitter (float): Jitter that is used for the heartbeat interval.
      node_id (str): Unique identifier of the node.
      node_store (SqlAlchemyNodeStore): Store used for node operations.
      logger (logging.Logger): Logger instance used for reporting errors.
    """
    super().__init__(interval, jitter)
    self._node_id = node_id
    self._node_store = node_store
    self._logger = logger

execute()

Execute the heartbeat update.

Updates the last_seen_at field of the node in the database. Any errors are logged without interrupting the heartbeat loop.

Source code in lilota/node.py

def execute(self):
    """Execute the heartbeat update.

    Updates the ``last_seen_at`` field of the node in the database.
    Any errors are logged without interrupting the heartbeat loop.
    """
    try:
        self._node_store.update_node_last_seen_at(self._node_id)
    except Exception:
        self._logger.exception(
            f"Heartbeat update failed for node_id '{self._node_id}'"
        )

LilotaScheduler

Bases: LilotaNode

Scheduler node responsible for creating and managing tasks.

The scheduler registers itself as a scheduler node in the system and periodically sends heartbeats to indicate it is alive. Its main role is to create tasks and store them in the task store for workers to execute.

Source code in lilota/scheduler.py

class LilotaScheduler(LilotaNode):
    """Scheduler node responsible for creating and managing tasks.

    The scheduler registers itself as a scheduler node in the system and
    periodically sends heartbeats to indicate it is alive. Its main role
    is to create tasks and store them in the task store for workers to
    execute.
    """

    LOGGER_NAME = "lilota.scheduler"

    def __init__(
        self,
        db_url: str,
        node_heartbeat_interval: float = 5.0,
        node_timeout_sec: int = 20,
        logging_level=logging.INFO,
        **kwargs,
    ):
        """Initialize the scheduler node.

        Args:
          db_url (str): Database connection URL.
          node_heartbeat_interval (float, optional): Interval in seconds between
            node heartbeats. Defaults to 5.0.
          node_timeout_sec (int, optional): Time in seconds before a node is
            considered inactive. Defaults to 20.
          logging_level (int, optional): Logging level used by the scheduler.
            Defaults to logging.INFO.
          **kwargs: Additional keyword arguments passed to the parent
            ``LilotaNode`` initializer.
        """
        super().__init__(
            db_url=db_url,
            node_type=NodeType.SCHEDULER,
            node_heartbeat_interval=node_heartbeat_interval,
            node_timeout_sec=node_timeout_sec,
            logger_name=self.LOGGER_NAME,
            logging_level=logging_level,
            **kwargs,
        )

    def _on_started(self):
        # Start heartbeat thread
        heartbeat_task = NodeHeartbeatTask(
            self._node_heartbeat_interval,
            None,
            self._node_id,
            self._node_store,
            self._logger,
        )
        self._heartbeat = Heartbeat(
            f"scheduler_heartbeat_{self._node_id}", heartbeat_task, self._logger
        )
        self._heartbeat.start()

        # Log Node started message
        self._logger.debug("Node started")

    def _on_stop(self):
        # Stop heartbeat thread
        self._stop_node_heartbeat()

    def schedule(self, name: str, input: Any = None) -> int:
        """Create and store a new task.

        Args:
          name (str): Name of the registered task.
          input (Any, optional): Input payload for the task. Defaults to None.

        Returns:
          int: Identifier of the created task.
        """
        self._logger.debug(f"Create task (name: '{name}', input: {input})")
        return self._task_store.create_task(name, input)

    def has_unfinished_tasks(self):
        """Check whether there are unfinished tasks in the system.

        Returns:
          bool: True if unfinished tasks exist, otherwise False.
        """
        return self._task_store.has_unfinished_tasks()

__init__(db_url, node_heartbeat_interval=5.0, node_timeout_sec=20, logging_level=logging.INFO, **kwargs)

Initialize the scheduler node.

Parameters:

Name	Type	Description	Default
db_url	str	Database connection URL.	required
node_heartbeat_interval	float	Interval in seconds between node heartbeats. Defaults to 5.0.	5.0
node_timeout_sec	int	Time in seconds before a node is considered inactive. Defaults to 20.	20
logging_level	int	Logging level used by the scheduler. Defaults to logging.INFO.	INFO
**kwargs		Additional keyword arguments passed to the parent LilotaNode initializer.	{}

Source code in lilota/scheduler.py

def __init__(
    self,
    db_url: str,
    node_heartbeat_interval: float = 5.0,
    node_timeout_sec: int = 20,
    logging_level=logging.INFO,
    **kwargs,
):
    """Initialize the scheduler node.

    Args:
      db_url (str): Database connection URL.
      node_heartbeat_interval (float, optional): Interval in seconds between
        node heartbeats. Defaults to 5.0.
      node_timeout_sec (int, optional): Time in seconds before a node is
        considered inactive. Defaults to 20.
      logging_level (int, optional): Logging level used by the scheduler.
        Defaults to logging.INFO.
      **kwargs: Additional keyword arguments passed to the parent
        ``LilotaNode`` initializer.
    """
    super().__init__(
        db_url=db_url,
        node_type=NodeType.SCHEDULER,
        node_heartbeat_interval=node_heartbeat_interval,
        node_timeout_sec=node_timeout_sec,
        logger_name=self.LOGGER_NAME,
        logging_level=logging_level,
        **kwargs,
    )

has_unfinished_tasks()

Check whether there are unfinished tasks in the system.

Returns:

Name	Type	Description
bool		True if unfinished tasks exist, otherwise False.

Source code in lilota/scheduler.py

def has_unfinished_tasks(self):
    """Check whether there are unfinished tasks in the system.

    Returns:
      bool: True if unfinished tasks exist, otherwise False.
    """
    return self._task_store.has_unfinished_tasks()

schedule(name, input=None)

Create and store a new task.

Parameters:

Name	Type	Description	Default
name	str	Name of the registered task.	required
input	Any	Input payload for the task. Defaults to None.	None

Returns:

Name	Type	Description
int	int	Identifier of the created task.

Source code in lilota/scheduler.py

def schedule(self, name: str, input: Any = None) -> int:
    """Create and store a new task.

    Args:
      name (str): Name of the registered task.
      input (Any, optional): Input payload for the task. Defaults to None.

    Returns:
      int: Identifier of the created task.
    """
    self._logger.debug(f"Create task (name: '{name}', input: {input})")
    return self._task_store.create_task(name, input)

LilotaWorker

Bases: LilotaNode

Worker node responsible for executing scheduled tasks.

Workers poll the task store for pending tasks, execute registered functions, and update task status and progress. Each worker also sends periodic heartbeats and participates in leader election for cluster maintenance.

Source code in lilota/worker.py

class LilotaWorker(LilotaNode):
    """Worker node responsible for executing scheduled tasks.

    Workers poll the task store for pending tasks, execute registered
    functions, and update task status and progress. Each worker also
    sends periodic heartbeats and participates in leader election
    for cluster maintenance.
    """

    LOGGER_NAME = "lilota.worker"
    MAX_ATTEMPTS = 1

    def __init__(
        self,
        db_url: str = DEFAULT_DB_URL,
        node_heartbeat_interval: float = 5.0,
        node_heartbeat_interval_jitter: float = 0.2,
        node_timeout_sec: int = 20,
        task_heartbeat_interval: float = 0.1,
        max_task_heartbeat_interval: float = 5.0,
        logging_level=logging.INFO,
        **kwargs,
    ):
        """Initialize a worker node.

        Args:
          db_url (str): Database connection URL.
          node_heartbeat_interval (float, optional): Interval in seconds between
            node heartbeats. Defaults to 5.0.
          node_heartbeat_interval_jitter (float, optional): Jitter applied to the heartbeat interval
            to introduce small random variations in execution timing and avoid
            synchronized worker behavior.
          node_timeout_sec (int, optional): Time in seconds before a node is
            considered inactive. Defaults to 20.
          task_heartbeat_interval (float, optional): Initial interval in seconds
            between polling attempts for tasks. Defaults to 0.1.
          max_task_heartbeat_interval (float, optional): Maximum polling interval
            when no tasks are available. Defaults to 5.0.
          logging_level (int, optional): Logging level used by the worker.
          **kwargs: Additional keyword arguments passed to ``LilotaNode``.
        """

        super().__init__(
            db_url=db_url,
            node_type=NodeType.WORKER,
            node_heartbeat_interval=node_heartbeat_interval,
            node_timeout_sec=node_timeout_sec,
            logger_name=self.LOGGER_NAME,
            logging_level=logging_level,
            **kwargs,
        )

        if node_heartbeat_interval_jitter is not None:
            if (
                node_heartbeat_interval_jitter <= 0
                or node_heartbeat_interval_jitter >= 1
            ):
                raise ValueError(
                    "node_heartbeat_interval_jitter must be greater than 0 and smaller than 1"
                )

        self._node_heartbeat_interval_jitter = node_heartbeat_interval_jitter
        self._task_heartbeat_interval: float = task_heartbeat_interval
        self._max_task_heartbeat_interval: float = max_task_heartbeat_interval
        self._registry: dict[str, RegisteredTask] = {}

    def _analyze_task_signature(self, func):
        sig = inspect.signature(func)
        hints = get_type_hints(func)

        input_model = None
        input_param_name = None
        context_param_name = None

        params = list(sig.parameters.values())
        if len(params) > 2:
            raise TypeError(
                "Lilota tasks support:\n"
                "- no parameters\n"
                "- payload\n"
                "- payload + TaskContext"
            )

        for param in params:
            annotation = hints.get(param.name)

            if annotation is TaskContext:
                context_param_name = param.name
                continue

            if input_model is not None:
                raise TypeError("Only one input parameter is allowed")

            input_model = annotation
            input_param_name = param.name

        output_model = hints.get("return")

        return {
            "input_model": input_model,
            "input_param_name": input_param_name,
            "context_param_name": context_param_name,
            "output_model": output_model,
        }

    def _register(
        self,
        name: str,
        func: Callable,
        *,
        input_model: Optional[Type[Any]] = None,
        output_model: Optional[Type[Any]] = None,
        input_param_name: Optional[str] = None,
        context_param_name: Optional[str] = None,
        timeout: Optional[timedelta] = None,
        max_attempts: int = MAX_ATTEMPTS,
    ):
        if self._is_started:
            raise Exception(
                "It is not allowed to register functions after startup. Stop by using the stop() method."
            )

        if name in self._registry:
            raise RuntimeError(f"Task {name!r} is already registered")

        # Register the task
        task = RegisteredTask(
            func=func,
            input_model=input_model,
            output_model=output_model,
            input_param_name=input_param_name,
            context_param_name=context_param_name,
            timeout=timeout,
            max_attempts=max_attempts,
        )

        self._registry[name] = task

    def task(
        self,
        func=None,
        *,
        name: Optional[str] = None,
        timeout: timedelta = timedelta(minutes=5),
        max_attempts: int = MAX_ATTEMPTS,
    ):
        def decorator(fn):
            # Analyze the task signature
            inferred = self._analyze_task_signature(fn)

            # Task name
            task_name = name or fn.__name__

            # Register the task in the registry
            self._register(
                name=task_name,
                func=fn,
                input_model=inferred["input_model"],
                output_model=inferred["output_model"],
                input_param_name=inferred["input_param_name"],
                context_param_name=inferred["context_param_name"],
                timeout=timeout,
                max_attempts=max_attempts,
            )

            return fn

        # Supports:
        # @lilota.task
        if func is not None:
            return decorator(func)

        # Supports:
        # @lilota.task(...)
        return decorator

    def _on_started(self):
        # Create node leader store
        node_leader_store = NodeLeaderStore(
            self._db_url, self._logger, self._node_timeout_sec
        )

        # Start worker heartbeat thread
        heartbeat_task = WorkerHeartbeatTask(
            self._node_heartbeat_interval,
            self._node_heartbeat_interval_jitter,
            self._node_id,
            self._node_timeout_sec,
            self._node_store,
            node_leader_store,
            self._task_store,
            self._logger,
        )
        self._heartbeat = Heartbeat(
            f"scheduler_heartbeat_{self._node_id}", heartbeat_task, self._logger
        )
        self._heartbeat.start()

        # Log Node started message
        self._logger.debug("Node started")

        # Execute tasks
        self._execute_tasks()

    def _execute_tasks(self):
        interval: float = self._task_heartbeat_interval

        while True:
            # Get the next available task
            task = self._task_store.get_next_task(self._node_id)
            if task:
                # Initialize the variables
                task_id = task.id
                interval = 0.1

                # Configure logging
                logger: logging.Logger = create_context_logger(
                    self._logger, node_id=self._node_id, task_id=task_id
                )

                # Create the TaskProgress object
                task_progress = TaskProgress(task.id, self._task_store.set_progress)

                # Get the registered task
                registered_task = self._registry.get(task.name)
                if registered_task is None:
                    error_message = f"Task {task.name!r} not registered"
                    error = error_to_dict(error_message)
                    self._task_store.end_task_failure(task_id, error, task_progress)
                    logger.error(error_message)
                else:
                    # Execute the task
                    try:
                        # Set status to running
                        timeout = registered_task.timeout
                        max_attempts = registered_task.max_attempts
                        started_task = self._task_store.start_task(
                            task_id, max_attempts, timeout
                        )

                        # Run task
                        result = self._execute_task_with_watchdog(
                            started_task, registered_task, task_progress, logger
                        )

                        # Set status to completed
                        self._task_store.end_task_success(
                            task_id, result, task_progress
                        )
                    except Exception as ex:
                        self._logger.exception(f"Task execution failed (id: {task_id})")
                        self._task_store.end_task_failure(
                            task_id, exception_to_dict(ex), task_progress
                        )
                        raise
            else:
                # Increase the interval
                interval = min(interval * 2, self._max_task_heartbeat_interval)

            # Sleep
            time.sleep(interval)

    def _execute_task_with_watchdog(
        self,
        task: Task,
        registered_task: RegisteredTask,
        task_progress: TaskProgress,
        logger: logging.Logger,
    ):
        # Create the task context
        task_context: TaskContext = None
        if registered_task.context_param_name is not None:
            task_context: TaskContext = TaskContext(
                task_id=task.id,
                progress=task_progress,
                logger=logger,
            )

        if task.expires_at is None:
            return registered_task(task.input, task_context)

        timer = self._calculate_timer_and_set_watchdog(task, logger)
        try:
            return registered_task(task.input, task_context)
        finally:
            timer.cancel()

    def _calculate_timer_and_set_watchdog(
        self, task: Task, logger: logging.Logger
    ) -> threading.Timer:
        timeout: float = max(
            0, (task.expires_at - datetime.now(timezone.utc)).total_seconds()
        )

        def watchdog():
            # Log that process will be killed
            logger.error(
                f"The process will be stopped because the task '{task.name}' ({task.id}) has expired"
            )

            # Kill the worker process
            os._exit(1)

        timer = threading.Timer(timeout, watchdog)
        timer.start()
        return timer

    def _on_stop(self):
        # Stop worker heartbeat thread
        self._stop_node_heartbeat()

__init__(db_url=DEFAULT_DB_URL, node_heartbeat_interval=5.0, node_heartbeat_interval_jitter=0.2, node_timeout_sec=20, task_heartbeat_interval=0.1, max_task_heartbeat_interval=5.0, logging_level=logging.INFO, **kwargs)

Initialize a worker node.

Parameters:

Name	Type	Description	Default
db_url	str	Database connection URL.	DEFAULT_DB_URL
node_heartbeat_interval	float	Interval in seconds between node heartbeats. Defaults to 5.0.	5.0
node_heartbeat_interval_jitter	float	Jitter applied to the heartbeat interval to introduce small random variations in execution timing and avoid synchronized worker behavior.	0.2
node_timeout_sec	int	Time in seconds before a node is considered inactive. Defaults to 20.	20
task_heartbeat_interval	float	Initial interval in seconds between polling attempts for tasks. Defaults to 0.1.	0.1
max_task_heartbeat_interval	float	Maximum polling interval when no tasks are available. Defaults to 5.0.	5.0
logging_level	int	Logging level used by the worker.	INFO
**kwargs		Additional keyword arguments passed to LilotaNode.	{}

Source code in lilota/worker.py

def __init__(
    self,
    db_url: str = DEFAULT_DB_URL,
    node_heartbeat_interval: float = 5.0,
    node_heartbeat_interval_jitter: float = 0.2,
    node_timeout_sec: int = 20,
    task_heartbeat_interval: float = 0.1,
    max_task_heartbeat_interval: float = 5.0,
    logging_level=logging.INFO,
    **kwargs,
):
    """Initialize a worker node.

    Args:
      db_url (str): Database connection URL.
      node_heartbeat_interval (float, optional): Interval in seconds between
        node heartbeats. Defaults to 5.0.
      node_heartbeat_interval_jitter (float, optional): Jitter applied to the heartbeat interval
        to introduce small random variations in execution timing and avoid
        synchronized worker behavior.
      node_timeout_sec (int, optional): Time in seconds before a node is
        considered inactive. Defaults to 20.
      task_heartbeat_interval (float, optional): Initial interval in seconds
        between polling attempts for tasks. Defaults to 0.1.
      max_task_heartbeat_interval (float, optional): Maximum polling interval
        when no tasks are available. Defaults to 5.0.
      logging_level (int, optional): Logging level used by the worker.
      **kwargs: Additional keyword arguments passed to ``LilotaNode``.
    """

    super().__init__(
        db_url=db_url,
        node_type=NodeType.WORKER,
        node_heartbeat_interval=node_heartbeat_interval,
        node_timeout_sec=node_timeout_sec,
        logger_name=self.LOGGER_NAME,
        logging_level=logging_level,
        **kwargs,
    )

    if node_heartbeat_interval_jitter is not None:
        if (
            node_heartbeat_interval_jitter <= 0
            or node_heartbeat_interval_jitter >= 1
        ):
            raise ValueError(
                "node_heartbeat_interval_jitter must be greater than 0 and smaller than 1"
            )

    self._node_heartbeat_interval_jitter = node_heartbeat_interval_jitter
    self._task_heartbeat_interval: float = task_heartbeat_interval
    self._max_task_heartbeat_interval: float = max_task_heartbeat_interval
    self._registry: dict[str, RegisteredTask] = {}

WorkerHeartbeatTask

Bases: NodeHeartbeatTask

Heartbeat task used by worker nodes.

In addition to updating the node heartbeat, this task also performs leader election among workers. The elected leader periodically performs maintenance tasks such as cleaning up stale nodes.

Source code in lilota/worker.py

class WorkerHeartbeatTask(NodeHeartbeatTask):
    """Heartbeat task used by worker nodes.

    In addition to updating the node heartbeat, this task also performs
    leader election among workers. The elected leader periodically performs
    maintenance tasks such as cleaning up stale nodes.
    """

    def __init__(
        self,
        interval: float,
        jitter: float,
        node_id: str,
        node_timeout_sec: int,
        node_store: NodeStore,
        node_leader_store: NodeLeaderStore,
        task_store: TaskStore,
        logger: logging.Logger,
    ):
        """Initialize the worker heartbeat task.

        Args:
          interval (float): Interval in seconds between heartbeats.
          jitter (float): Jitter that is used for the heartbeat interval.
          node_id (str): Unique identifier of the worker node.
          node_timeout_sec (int): Timeout in seconds after which nodes are
            considered dead.
          node_store (SqlAlchemyNodeStore): Store used for node operations.
          node_leader_store (SqlAlchemyNodeLeaderStore): Store used for leader
            election and leadership renewal.
          logger (logging.Logger): Logger instance.
        """
        super().__init__(interval, jitter, node_id, node_store, logger)
        self._node_timeout_sec = node_timeout_sec
        self._node_leader_store = node_leader_store
        self._task_store = task_store
        self._is_leader = False

    def execute(self):
        """Execute the heartbeat logic.

        Updates the node's last-seen timestamp and attempts to perform leader
        election. If the node becomes leader, it will also trigger cleanup tasks.
        """

        # Update last_seen_at
        super().execute()

        # Try to set leader and the leader should trigger a cleanup
        self._try_set_leader_and_run_maintenance()

    def _try_set_leader_and_run_maintenance(self):
        # Try to renew leadership
        if self._is_leader:
            self._is_leader = self._node_leader_store.renew_leadership(self._node_id)

            if not self._is_leader:
                self._logger.debug(f"Leadership lost (node id: {self._node_id})")

        # Try to acquire leadership if not leader
        if not self._is_leader:
            self._is_leader = self._node_leader_store.try_acquire_leadership(
                self._node_id
            )

        # Leader-only work
        if self._is_leader:
            self._run_maintenance()

    def _run_maintenance(self) -> None:
        try:
            self._update_status_on_dead_nodes()
            self._update_status_on_expired_tasks()
            self._retry_tasks()
        except Exception:
            # Never let maintenance kill the heartbeat thread
            self._logger.exception("Node maintenance failed")

    def _update_status_on_dead_nodes(self):
        cutoff = datetime.now(timezone.utc) - timedelta(seconds=self._node_timeout_sec)
        cleaned = self._node_store.update_status_on_dead_nodes(cutoff, self._node_id)
        if cleaned > 0:
            self._logger.debug(f"Marked {cleaned} stale node(s) as DEAD")

    def _update_status_on_expired_tasks(self):
        self._task_store.expire_overdue_tasks()

    def _retry_tasks(self):
        self._task_store.retry_tasks()

__init__(interval, jitter, node_id, node_timeout_sec, node_store, node_leader_store, task_store, logger)

Initialize the worker heartbeat task.

Parameters:

Name	Type	Description	Default
interval	float	Interval in seconds between heartbeats.	required
jitter	float	Jitter that is used for the heartbeat interval.	required
node_id	str	Unique identifier of the worker node.	required
node_timeout_sec	int	Timeout in seconds after which nodes are considered dead.	required
node_store	SqlAlchemyNodeStore	Store used for node operations.	required
node_leader_store	SqlAlchemyNodeLeaderStore	Store used for leader election and leadership renewal.	required
logger	Logger	Logger instance.	required

Source code in lilota/worker.py

def __init__(
    self,
    interval: float,
    jitter: float,
    node_id: str,
    node_timeout_sec: int,
    node_store: NodeStore,
    node_leader_store: NodeLeaderStore,
    task_store: TaskStore,
    logger: logging.Logger,
):
    """Initialize the worker heartbeat task.

    Args:
      interval (float): Interval in seconds between heartbeats.
      jitter (float): Jitter that is used for the heartbeat interval.
      node_id (str): Unique identifier of the worker node.
      node_timeout_sec (int): Timeout in seconds after which nodes are
        considered dead.
      node_store (SqlAlchemyNodeStore): Store used for node operations.
      node_leader_store (SqlAlchemyNodeLeaderStore): Store used for leader
        election and leadership renewal.
      logger (logging.Logger): Logger instance.
    """
    super().__init__(interval, jitter, node_id, node_store, logger)
    self._node_timeout_sec = node_timeout_sec
    self._node_leader_store = node_leader_store
    self._task_store = task_store
    self._is_leader = False

execute()

Execute the heartbeat logic.

Updates the node's last-seen timestamp and attempts to perform leader election. If the node becomes leader, it will also trigger cleanup tasks.

Source code in lilota/worker.py

def execute(self):
    """Execute the heartbeat logic.

    Updates the node's last-seen timestamp and attempts to perform leader
    election. If the node becomes leader, it will also trigger cleanup tasks.
    """

    # Update last_seen_at
    super().execute()

    # Try to set leader and the leader should trigger a cleanup
    self._try_set_leader_and_run_maintenance()

LogStore

Database store for logging entries into Lilota.

Source code in lilota/stores.py

class LogStore:
    """Database store for logging entries into Lilota."""

    def __init__(self, db_url: str):
        """
        Args:
            db_url (str): Database connection URL.
        """
        self._db_url = db_url
        self._engine = None
        self._Session = None

    def _ensure_engine(self):
        if self._engine is None:
            self._engine = create_engine(self._db_url, pool_pre_ping=True)
            self._Session = sessionmaker(
                bind=self._engine,
                expire_on_commit=False,
            )

    def get_session(self):
        """Return a new SQLAlchemy session."""
        self._ensure_engine()
        return self._Session()

    def get_log_entries(self) -> list[LogEntry]:
        """Return all log entries."""
        with self.get_session() as session:
            return session.query(LogEntry).order_by(LogEntry.created_at).all()

    def get_log_entries_by_node_id(self, node_id: UUID) -> list[LogEntry]:
        """Return all log entries associated with a given node."""
        with self.get_session() as session:
            return (
                session.query(LogEntry)
                .filter(LogEntry.node_id == node_id)
                .order_by(LogEntry.created_at)
                .all()
            )

__init__(db_url)

Parameters:

Name	Type	Description	Default
db_url	str	Database connection URL.	required

Source code in lilota/stores.py

def __init__(self, db_url: str):
    """
    Args:
        db_url (str): Database connection URL.
    """
    self._db_url = db_url
    self._engine = None
    self._Session = None

get_log_entries()

Return all log entries.

Source code in lilota/stores.py

def get_log_entries(self) -> list[LogEntry]:
    """Return all log entries."""
    with self.get_session() as session:
        return session.query(LogEntry).order_by(LogEntry.created_at).all()

get_log_entries_by_node_id(node_id)

Return all log entries associated with a given node.

Source code in lilota/stores.py

def get_log_entries_by_node_id(self, node_id: UUID) -> list[LogEntry]:
    """Return all log entries associated with a given node."""
    with self.get_session() as session:
        return (
            session.query(LogEntry)
            .filter(LogEntry.node_id == node_id)
            .order_by(LogEntry.created_at)
            .all()
        )

get_session()

Return a new SQLAlchemy session.

Source code in lilota/stores.py

def get_session(self):
    """Return a new SQLAlchemy session."""
    self._ensure_engine()
    return self._Session()

NodeLeaderStore

Bases: StoreBase

Store managing leader election for worker nodes.

Source code in lilota/stores.py

class NodeLeaderStore(StoreBase):
    """Store managing leader election for worker nodes."""

    def __init__(self, db_url: str, logger: logging.Logger, node_timeout_sec: int):
        """
        Args:
            db_url (str): Database connection URL.
            logger (logging.Logger): Logger for store operations.
            node_timeout_sec (int): Leader lease timeout in seconds.
        """
        super().__init__(db_url, logger)
        self._node_timeout_sec: int = node_timeout_sec

    def try_acquire_leadership(self, node_id) -> bool:
        """Attempt to acquire leadership for the given node.

        Returns True if leadership is acquired, False otherwise.
        """
        now = datetime.now(timezone.utc)
        new_expiry = now + timedelta(seconds=self._node_timeout_sec)

        with self._get_session() as session:
            try:
                # Try to take over expired lease
                result = session.execute(
                    update(NodeLeader)
                    .where(
                        NodeLeader.id == 1,
                        NodeLeader.lease_expires_at < now,
                    )
                    .values(
                        node_id=node_id,
                        lease_expires_at=new_expiry,
                    )
                )

                if result.rowcount == 1:
                    session.commit()
                    if self._logger:
                        self._logger.debug(
                            f"Leadership acquired (new node id: {node_id})"
                        )
                    return True

                # Check if row exists
                exists = session.execute(
                    select(NodeLeader.id).where(NodeLeader.id == 1)
                ).first()

                if exists:
                    session.rollback()
                    return False

                # No row → try to create it
                session.add(
                    NodeLeader(
                        id=1,
                        node_id=node_id,
                        lease_expires_at=new_expiry,
                    )
                )
                session.commit()
                if self._logger:
                    self._logger.debug(
                        f"Leadership acquired first time (node id: {node_id})"
                    )
                return True
            except IntegrityError:
                # Someone else won the race to insert
                session.rollback()
                return False
            except Exception:
                session.rollback()
                if self._logger:
                    self._logger.exception("Unexpected error in leader election")
                raise

    def renew_leadership(self, node_id):
        """Renew leadership lease for the given node.

        Returns True if renewed successfully.
        """
        now = datetime.now(timezone.utc)
        new_expiry = now + timedelta(seconds=self._node_timeout_sec)

        with self._get_session() as session:
            try:
                result = session.execute(
                    update(NodeLeader)
                    .where(
                        NodeLeader.id == 1,
                        NodeLeader.node_id == node_id,
                        NodeLeader.lease_expires_at >= now,
                    )
                    .values(lease_expires_at=new_expiry)
                )

                session.commit()
                renewed = result.rowcount == 1
                if renewed and self._logger:
                    self._logger.debug(f"Leadership renewed (node id: {node_id})")
                return renewed
            except Exception as ex:
                session.rollback()
                if self._logger:
                    self._logger.exception(f"Renew leadership failed: {ex}")
                raise

__init__(db_url, logger, node_timeout_sec)

Parameters:

Name	Type	Description	Default
db_url	str	Database connection URL.	required
logger	Logger	Logger for store operations.	required
node_timeout_sec	int	Leader lease timeout in seconds.	required

Source code in lilota/stores.py

def __init__(self, db_url: str, logger: logging.Logger, node_timeout_sec: int):
    """
    Args:
        db_url (str): Database connection URL.
        logger (logging.Logger): Logger for store operations.
        node_timeout_sec (int): Leader lease timeout in seconds.
    """
    super().__init__(db_url, logger)
    self._node_timeout_sec: int = node_timeout_sec

renew_leadership(node_id)

Renew leadership lease for the given node.

Returns True if renewed successfully.

Source code in lilota/stores.py

def renew_leadership(self, node_id):
    """Renew leadership lease for the given node.

    Returns True if renewed successfully.
    """
    now = datetime.now(timezone.utc)
    new_expiry = now + timedelta(seconds=self._node_timeout_sec)

    with self._get_session() as session:
        try:
            result = session.execute(
                update(NodeLeader)
                .where(
                    NodeLeader.id == 1,
                    NodeLeader.node_id == node_id,
                    NodeLeader.lease_expires_at >= now,
                )
                .values(lease_expires_at=new_expiry)
            )

            session.commit()
            renewed = result.rowcount == 1
            if renewed and self._logger:
                self._logger.debug(f"Leadership renewed (node id: {node_id})")
            return renewed
        except Exception as ex:
            session.rollback()
            if self._logger:
                self._logger.exception(f"Renew leadership failed: {ex}")
            raise

try_acquire_leadership(node_id)

Attempt to acquire leadership for the given node.

Returns True if leadership is acquired, False otherwise.

Source code in lilota/stores.py

def try_acquire_leadership(self, node_id) -> bool:
    """Attempt to acquire leadership for the given node.

    Returns True if leadership is acquired, False otherwise.
    """
    now = datetime.now(timezone.utc)
    new_expiry = now + timedelta(seconds=self._node_timeout_sec)

    with self._get_session() as session:
        try:
            # Try to take over expired lease
            result = session.execute(
                update(NodeLeader)
                .where(
                    NodeLeader.id == 1,
                    NodeLeader.lease_expires_at < now,
                )
                .values(
                    node_id=node_id,
                    lease_expires_at=new_expiry,
                )
            )

            if result.rowcount == 1:
                session.commit()
                if self._logger:
                    self._logger.debug(
                        f"Leadership acquired (new node id: {node_id})"
                    )
                return True

            # Check if row exists
            exists = session.execute(
                select(NodeLeader.id).where(NodeLeader.id == 1)
            ).first()

            if exists:
                session.rollback()
                return False

            # No row → try to create it
            session.add(
                NodeLeader(
                    id=1,
                    node_id=node_id,
                    lease_expires_at=new_expiry,
                )
            )
            session.commit()
            if self._logger:
                self._logger.debug(
                    f"Leadership acquired first time (node id: {node_id})"
                )
            return True
        except IntegrityError:
            # Someone else won the race to insert
            session.rollback()
            return False
        except Exception:
            session.rollback()
            if self._logger:
                self._logger.exception("Unexpected error in leader election")
            raise

NodeStore

Bases: StoreBase

Database store for managing Lilota nodes.

Source code in lilota/stores.py

class NodeStore(StoreBase):
    """Database store for managing Lilota nodes."""

    def __init__(self, db_url: str, logger: logging.Logger):
        """
        Args:
            db_url (str): Database connection URL.
            logger (logging.Logger): Logger for store operations.
        """
        super().__init__(db_url, logger)

    def create_node(self, type: NodeType, status: NodeStatus = NodeStatus.IDLE) -> UUID:
        """Create a new node record in the database.

        Args:
            type (NodeType): Type of the node (scheduler or worker).
            status (NodeStatus): Initial lifecycle status.

        Returns:
            UUID: The unique identifier of the created node.
        """
        node = Node(type=type, status=status)

        with self._get_session() as session:
            with session.begin():
                session.add(node)

        return node.id

    def get_all_nodes(self):
        """Return all nodes in the database."""
        with self._get_session() as session:
            return session.query(Node).all()

    def get_node_by_id(self, id: UUID):
        """Return a node by its UUID.

        Args:
            id (UUID): Node identifier.

        Returns:
            Node | None: Node object if found, else None.
        """
        with self._get_session() as session:
            node = session.get(Node, id)
            if node is None:
                return None
            return node

    def update_node_status(self, id: UUID, status: NodeStatus):
        """Update the status of a node.

        Args:
            id (UUID): Node identifier.
            status (NodeStatus): New status.
        """
        with self._get_session() as session:
            with session.begin():
                stmt = update(Node).where(Node.id == id).values(status=status)
                session.execute(stmt)

    def update_status_on_dead_nodes(self, cutoff: datetime, exclude_node_id: UUID):
        """Mark nodes as DEAD if their last_seen_at is older than cutoff.

        Args:
            cutoff (datetime): Time threshold to consider a node dead.
            exclude_node_id (UUID): Node to exclude from the update.

        Returns:
            int: Number of nodes marked as DEAD.
        """
        with self._get_session() as session:
            with session.begin():
                result = session.execute(
                    update(Node)
                    .where(Node.last_seen_at < cutoff)
                    .where(Node.id != exclude_node_id)
                    .where(Node.status != NodeStatus.DEAD)
                    .values(status=NodeStatus.DEAD)
                )
                return result.rowcount

    def update_node_last_seen_at(self, id: UUID):
        """Update the heartbeat timestamp of a node.

        Args:
            id (UUID): Node identifier.
        """
        with self._get_session() as session:
            with session.begin():
                session.execute(
                    update(Node)
                    .where(Node.id == id)
                    .values(last_seen_at=datetime.now(timezone.utc))
                )

__init__(db_url, logger)

Parameters:

Name	Type	Description	Default
db_url	str	Database connection URL.	required
logger	Logger	Logger for store operations.	required

Source code in lilota/stores.py

def __init__(self, db_url: str, logger: logging.Logger):
    """
    Args:
        db_url (str): Database connection URL.
        logger (logging.Logger): Logger for store operations.
    """
    super().__init__(db_url, logger)

create_node(type, status=NodeStatus.IDLE)

Create a new node record in the database.

Parameters:

Name	Type	Description	Default
type	NodeType	Type of the node (scheduler or worker).	required
status	NodeStatus	Initial lifecycle status.	IDLE

Returns:

Name	Type	Description
UUID	UUID	The unique identifier of the created node.

Source code in lilota/stores.py

def create_node(self, type: NodeType, status: NodeStatus = NodeStatus.IDLE) -> UUID:
    """Create a new node record in the database.

    Args:
        type (NodeType): Type of the node (scheduler or worker).
        status (NodeStatus): Initial lifecycle status.

    Returns:
        UUID: The unique identifier of the created node.
    """
    node = Node(type=type, status=status)

    with self._get_session() as session:
        with session.begin():
            session.add(node)

    return node.id

get_all_nodes()

Return all nodes in the database.

Source code in lilota/stores.py

def get_all_nodes(self):
    """Return all nodes in the database."""
    with self._get_session() as session:
        return session.query(Node).all()

get_node_by_id(id)

Return a node by its UUID.

Parameters:

Name	Type	Description	Default
id	UUID	Node identifier.	required

Returns:

Type	Description
	Node | None: Node object if found, else None.

Source code in lilota/stores.py

def get_node_by_id(self, id: UUID):
    """Return a node by its UUID.

    Args:
        id (UUID): Node identifier.

    Returns:
        Node | None: Node object if found, else None.
    """
    with self._get_session() as session:
        node = session.get(Node, id)
        if node is None:
            return None
        return node

update_node_last_seen_at(id)

Update the heartbeat timestamp of a node.

Parameters:

Name	Type	Description	Default
id	UUID	Node identifier.	required

Source code in lilota/stores.py

def update_node_last_seen_at(self, id: UUID):
    """Update the heartbeat timestamp of a node.

    Args:
        id (UUID): Node identifier.
    """
    with self._get_session() as session:
        with session.begin():
            session.execute(
                update(Node)
                .where(Node.id == id)
                .values(last_seen_at=datetime.now(timezone.utc))
            )

update_node_status(id, status)

Update the status of a node.

Parameters:

Name	Type	Description	Default
id	UUID	Node identifier.	required
status	NodeStatus	New status.	required

Source code in lilota/stores.py

def update_node_status(self, id: UUID, status: NodeStatus):
    """Update the status of a node.

    Args:
        id (UUID): Node identifier.
        status (NodeStatus): New status.
    """
    with self._get_session() as session:
        with session.begin():
            stmt = update(Node).where(Node.id == id).values(status=status)
            session.execute(stmt)

update_status_on_dead_nodes(cutoff, exclude_node_id)

Mark nodes as DEAD if their last_seen_at is older than cutoff.

Parameters:

Name	Type	Description	Default
cutoff	datetime	Time threshold to consider a node dead.	required
exclude_node_id	UUID	Node to exclude from the update.	required

Returns:

Name	Type	Description
int		Number of nodes marked as DEAD.

Source code in lilota/stores.py

def update_status_on_dead_nodes(self, cutoff: datetime, exclude_node_id: UUID):
    """Mark nodes as DEAD if their last_seen_at is older than cutoff.

    Args:
        cutoff (datetime): Time threshold to consider a node dead.
        exclude_node_id (UUID): Node to exclude from the update.

    Returns:
        int: Number of nodes marked as DEAD.
    """
    with self._get_session() as session:
        with session.begin():
            result = session.execute(
                update(Node)
                .where(Node.last_seen_at < cutoff)
                .where(Node.id != exclude_node_id)
                .where(Node.status != NodeStatus.DEAD)
                .values(status=NodeStatus.DEAD)
            )
            return result.rowcount

StoreBase

Bases: ABC

Abstract base class for all database stores.

Provides common initialization and session management.

Source code in lilota/stores.py

class StoreBase(ABC):
    """Abstract base class for all database stores.

    Provides common initialization and session management.
    """

    def __init__(self, db_url: str, logger: logging.Logger):
        """
        Args:
            db_url (str): Database connection URL.
            logger (logging.Logger): Logger for store operations.
        """
        self._db_url = db_url
        self._logger = logger
        self._engine = None
        self._Session = None

    def _ensure_engine(self):
        if self._engine is None:
            self._engine = get_engine(self._db_url)
            self._Session = sessionmaker(bind=self._engine, expire_on_commit=False)

    def _get_session(self):
        self._ensure_engine()
        return self._Session()

__init__(db_url, logger)

Parameters:

Name	Type	Description	Default
db_url	str	Database connection URL.	required
logger	Logger	Logger for store operations.	required

Source code in lilota/stores.py

def __init__(self, db_url: str, logger: logging.Logger):
    """
    Args:
        db_url (str): Database connection URL.
        logger (logging.Logger): Logger for store operations.
    """
    self._db_url = db_url
    self._logger = logger
    self._engine = None
    self._Session = None

TaskStore

Bases: StoreBase

Database store for managing Lilota tasks.

Source code in lilota/stores.py

class TaskStore(StoreBase):
    """Database store for managing Lilota tasks."""

    BATCH_SIZE = 50

    def __init__(self, db_url: str, logger: logging.Logger):
        """
        Args:
            db_url (str): Database connection URL.
            logger (logging.Logger): Logger for store operations.
        """
        super().__init__(db_url, logger)

    def create_task(self, name: str, input: Any = None):
        """Create a new task record in the database.

        Args:
            name (str): Registered task name.
            input (Any, optional): Input data for the task.

        Returns:
            UUID: The unique identifier of the created task.
        """
        if input is not None:
            input = normalize_data(input)

        task = Task(name=name, input=input, status=TaskStatus.CREATED)

        with self._get_session() as session:
            with session.begin():
                session.add(task)

        return task.id

    def get_all_tasks(self):
        """Return all tasks ordered by ID."""
        with self._get_session() as session:
            return session.query(Task).order_by(Task.id).all()

    def get_unfinished_tasks(self) -> list[Task]:
        """Return all tasks that are not yet completed or failed."""
        with self._get_session() as session:
            return (
                session.query(Task)
                .filter(
                    Task.status.in_(
                        [TaskStatus.CREATED, TaskStatus.SCHEDULED, TaskStatus.RUNNING]
                    )
                )
                .order_by(Task.id)
                .all()
            )

    def expire_overdue_tasks(self) -> None:
        """Set status to "expired" for running tasks whose expiration time has passed."""
        with self._get_session() as session:
            with session.begin():
                session.execute(
                    update(Task)
                    .where(
                        Task.status == TaskStatus.RUNNING,
                        Task.expires_at.is_not(None),
                        Task.expires_at < datetime.now(timezone.utc),
                    )
                    .values(status=TaskStatus.EXPIRED)
                )

    def has_unfinished_tasks(self) -> bool:
        """Check if there are any unfinished tasks in the database."""
        with self._get_session() as session:
            return session.query(
                session.query(Task)
                .filter(
                    Task.status.in_(
                        [TaskStatus.CREATED, TaskStatus.SCHEDULED, TaskStatus.RUNNING]
                    )
                )
                .exists()
            ).scalar()

    def get_task_by_id(self, id: UUID) -> Task:
        """Return a task by its UUID."""
        with self._get_session() as session:
            task = session.get(Task, id)
            if task is None:
                return None
            return task

    def get_next_task(self, worker_id: UUID) -> Task:
        """Return the next available task for a worker and lock it.

        Args:
            worker_id (UUID): Worker node locking the task.

        Returns:
            Task | None: The next scheduled task, or None if no task is available.
        """
        with self._get_session() as session:
            with session.begin():
                task_id = session.execute(
                    select(Task.id)
                    .where(Task.status == TaskStatus.CREATED)
                    .where(Task.run_at <= datetime.now(timezone.utc))
                    .order_by(Task.run_at)
                    .limit(1)
                ).scalar()

                if not task_id:
                    return None

                result = session.execute(
                    update(Task)
                    .where(Task.id == task_id)
                    .where(Task.status == TaskStatus.CREATED)
                    .values(
                        status=TaskStatus.SCHEDULED,
                        locked_at=datetime.now(timezone.utc),
                        locked_by=worker_id,
                    )
                )

            if result.rowcount != 1:
                return None

            return session.get(Task, task_id)

    def start_task(
        self, id: UUID, max_attempts: int, timeout: timedelta | None = None
    ) -> Task:
        """Mark a task as RUNNING and initialize metadata.

        Args:
            id (UUID): Id of the task.
            max_attempts (int): Upper limit on how many times the task may be executed.
            timeout (timedelta | None): Optional timeout that can be set for a task.

        Returns:
            Task | None: The started task, or None if no task is available.
        """
        expires_at = None
        start_date_time = datetime.now(timezone.utc)
        timeout_sec = int(timeout.total_seconds()) if timeout is not None else None

        with self._get_session() as session:
            with session.begin():
                task = self._load_task(session, id)
                task.timeout = timeout_sec

                if timeout is not None:
                    expires_at = start_date_time + timeout

                task.pid = os.getpid()
                task.status = TaskStatus.RUNNING
                task.max_attempts = max_attempts
                task.progress_percentage = 0
                task.start_date_time = start_date_time
                task.expires_at = expires_at
                task.end_date_time = None
                return task

    def retry_tasks(self, batch_size=BATCH_SIZE):
        """Retry eligible tasks by creating new scheduled task instances.

        This method scans for tasks that are eligible for retry based on their
        status, expiration, and retry limits. It uses optimistic locking to safely
        "claim" tasks in concurrent environments, ensuring that no task is retried
        more than once across multiple workers.

        Eligible tasks include:
            - Tasks with status "failed" or "expired"
            - Tasks with status "running" that have exceeded their expiration time

        A task is only retried if:
            - It has not already been retried (`retried_at` is None)
            - It has remaining retry attempts (`attempts < max_attempts`)

        For each claimed task:
            - A new task is created with incremented attempt count
            - The retry is scheduled with a delay proportional to the number of attempts
            - The original task is marked with `retried_at` to prevent duplicate retries

        Concurrency is handled using optimistic locking via the `version` field.
        If another worker updates the task first, the current worker skips it.

        Args:
            batch_size (int, optional):
                Maximum number of tasks to process in a single run.
                Defaults to BATCH_SIZE.

        Returns:
            int:
                The number of retry tasks successfully created.

        Raises:
            sqlalchemy.exc.SQLAlchemyError:
                If a database error occurs during the transaction.

        Notes:
            - Uses a linear backoff strategy: delay = 5 * current_attempts (seconds)
            - All operations are executed within a single transactional session
            - Safe for concurrent execution across multiple workers
        """
        now = datetime.now(timezone.utc)

        with self._get_session() as session:
            with session.begin():
                # 1. Select tasks
                stmt = (
                    select(Task)
                    .where(
                        and_(
                            Task.retried_at.is_(None),
                            Task.attempts < Task.max_attempts,
                            or_(
                                Task.status.in_(["failed", "expired"]),
                                and_(
                                    Task.status == "running",
                                    Task.expires_at.is_not(None),
                                    Task.expires_at < now,
                                ),
                            ),
                        )
                    )
                    .order_by(Task.run_at)
                    .limit(batch_size)
                )

                tasks = session.execute(stmt).scalars().all()

                if not tasks:
                    return 0

                new_tasks = []

                for task in tasks:
                    if task.retried_at is not None:
                        continue

                    current_version = task.version
                    current_attempts = task.attempts

                    # 2. Try to "claim" via optimistic locking
                    result = session.execute(
                        update(Task)
                        .where(
                            Task.id == task.id,
                            Task.version == current_version,
                            Task.retried_at.is_(None),
                        )
                        .values(retried_at=now, version=current_version + 1)
                    )

                    # 3. If no row updated: someone else took it
                    if result.rowcount == 0:
                        continue

                    # 4. Create retry task
                    retry_task = Task(
                        name=task.name,
                        status=TaskStatus.CREATED,
                        run_at=now + timedelta(seconds=5 * current_attempts),
                        attempts=current_attempts + 1,
                        max_attempts=task.max_attempts,
                        previous_task_id=task.id,
                        timeout=task.timeout,
                        input=task.input,
                    )

                    new_tasks.append(retry_task)

                session.add_all(new_tasks)
                return len(new_tasks)

    def set_progress(self, id: UUID, progress: int):
        """Update the progress percentage of a task.

        Args:
            id (UUID): Id of the task.
            progress (int): The progress of the task (0-100)
        """
        with self._get_session() as session:
            with session.begin():
                task = self._load_task(session, id)
                task.progress_percentage = max(0, min(progress, 100))

    def end_task_success(self, id: UUID, output: Any, task_progress: TaskProgress):
        """Mark a task as successfully completed.

        Args:
            id (UUID): Id of the task.
            output (Any): Task result data.
        """
        if output is not None:
            output = normalize_data(output)

        with self._get_session() as session:
            with session.begin():
                task = self._load_task(session, id)
                task.output = output
                self._complete_progress(task, TaskStatus.COMPLETED, task_progress)

    def end_task_failure(self, id: UUID, error: dict, task_progress: TaskProgress):
        """Mark a task as failed.

        Args:
            id (UUID): Id of the task.
            error (dict): Error information to store.
        """
        with self._get_session() as session:
            with session.begin():
                task = self._load_task(session, id)
                task.error = error
                self._complete_progress(task, TaskStatus.FAILED, task_progress)

    def delete_task_by_id(self, id: UUID):
        """Delete a task by its UUID.

        Returns:
            bool: True if deleted, False if task not found.
        """
        with self._get_session() as session:
            with session.begin():
                task = session.get(Task, id)
                if task is None:
                    return False
                session.delete(task)
        return True

    def _complete_progress(
        self, task: Task, task_status: TaskStatus, task_progress: TaskProgress
    ):
        if not task_progress.set_progress_manually:
            task.progress_percentage = 100
        task.status = task_status
        task.end_date_time = datetime.now(timezone.utc)

    def _load_task(self, session, id: UUID) -> Task:
        task: Task = session.get(Task, id)
        if task is None:
            raise ValueError(f"Task {id} not found")
        return task

__init__(db_url, logger)

Parameters:

Name	Type	Description	Default
db_url	str	Database connection URL.	required
logger	Logger	Logger for store operations.	required

Source code in lilota/stores.py

def __init__(self, db_url: str, logger: logging.Logger):
    """
    Args:
        db_url (str): Database connection URL.
        logger (logging.Logger): Logger for store operations.
    """
    super().__init__(db_url, logger)

create_task(name, input=None)

Create a new task record in the database.

Parameters:

Name	Type	Description	Default
name	str	Registered task name.	required
input	Any	Input data for the task.	None

Returns:

Name	Type	Description
UUID		The unique identifier of the created task.

Source code in lilota/stores.py

def create_task(self, name: str, input: Any = None):
    """Create a new task record in the database.

    Args:
        name (str): Registered task name.
        input (Any, optional): Input data for the task.

    Returns:
        UUID: The unique identifier of the created task.
    """
    if input is not None:
        input = normalize_data(input)

    task = Task(name=name, input=input, status=TaskStatus.CREATED)

    with self._get_session() as session:
        with session.begin():
            session.add(task)

    return task.id

delete_task_by_id(id)

Delete a task by its UUID.

Returns:

Name	Type	Description
bool		True if deleted, False if task not found.

Source code in lilota/stores.py

def delete_task_by_id(self, id: UUID):
    """Delete a task by its UUID.

    Returns:
        bool: True if deleted, False if task not found.
    """
    with self._get_session() as session:
        with session.begin():
            task = session.get(Task, id)
            if task is None:
                return False
            session.delete(task)
    return True

end_task_failure(id, error, task_progress)

Mark a task as failed.

Parameters:

Name	Type	Description	Default
id	UUID	Id of the task.	required
error	dict	Error information to store.	required

Source code in lilota/stores.py

def end_task_failure(self, id: UUID, error: dict, task_progress: TaskProgress):
    """Mark a task as failed.

    Args:
        id (UUID): Id of the task.
        error (dict): Error information to store.
    """
    with self._get_session() as session:
        with session.begin():
            task = self._load_task(session, id)
            task.error = error
            self._complete_progress(task, TaskStatus.FAILED, task_progress)

end_task_success(id, output, task_progress)

Mark a task as successfully completed.

Parameters:

Name	Type	Description	Default
id	UUID	Id of the task.	required
output	Any	Task result data.	required

Source code in lilota/stores.py

def end_task_success(self, id: UUID, output: Any, task_progress: TaskProgress):
    """Mark a task as successfully completed.

    Args:
        id (UUID): Id of the task.
        output (Any): Task result data.
    """
    if output is not None:
        output = normalize_data(output)

    with self._get_session() as session:
        with session.begin():
            task = self._load_task(session, id)
            task.output = output
            self._complete_progress(task, TaskStatus.COMPLETED, task_progress)

expire_overdue_tasks()

Set status to "expired" for running tasks whose expiration time has passed.

Source code in lilota/stores.py

def expire_overdue_tasks(self) -> None:
    """Set status to "expired" for running tasks whose expiration time has passed."""
    with self._get_session() as session:
        with session.begin():
            session.execute(
                update(Task)
                .where(
                    Task.status == TaskStatus.RUNNING,
                    Task.expires_at.is_not(None),
                    Task.expires_at < datetime.now(timezone.utc),
                )
                .values(status=TaskStatus.EXPIRED)
            )

get_all_tasks()

Return all tasks ordered by ID.

Source code in lilota/stores.py

def get_all_tasks(self):
    """Return all tasks ordered by ID."""
    with self._get_session() as session:
        return session.query(Task).order_by(Task.id).all()

get_next_task(worker_id)

Return the next available task for a worker and lock it.

Parameters:

Name	Type	Description	Default
worker_id	UUID	Worker node locking the task.	required

Returns:

Type	Description
Task	Task | None: The next scheduled task, or None if no task is available.

Source code in lilota/stores.py

def get_next_task(self, worker_id: UUID) -> Task:
    """Return the next available task for a worker and lock it.

    Args:
        worker_id (UUID): Worker node locking the task.

    Returns:
        Task | None: The next scheduled task, or None if no task is available.
    """
    with self._get_session() as session:
        with session.begin():
            task_id = session.execute(
                select(Task.id)
                .where(Task.status == TaskStatus.CREATED)
                .where(Task.run_at <= datetime.now(timezone.utc))
                .order_by(Task.run_at)
                .limit(1)
            ).scalar()

            if not task_id:
                return None

            result = session.execute(
                update(Task)
                .where(Task.id == task_id)
                .where(Task.status == TaskStatus.CREATED)
                .values(
                    status=TaskStatus.SCHEDULED,
                    locked_at=datetime.now(timezone.utc),
                    locked_by=worker_id,
                )
            )

        if result.rowcount != 1:
            return None

        return session.get(Task, task_id)

get_task_by_id(id)

Return a task by its UUID.

Source code in lilota/stores.py

def get_task_by_id(self, id: UUID) -> Task:
    """Return a task by its UUID."""
    with self._get_session() as session:
        task = session.get(Task, id)
        if task is None:
            return None
        return task

get_unfinished_tasks()

Return all tasks that are not yet completed or failed.

Source code in lilota/stores.py

def get_unfinished_tasks(self) -> list[Task]:
    """Return all tasks that are not yet completed or failed."""
    with self._get_session() as session:
        return (
            session.query(Task)
            .filter(
                Task.status.in_(
                    [TaskStatus.CREATED, TaskStatus.SCHEDULED, TaskStatus.RUNNING]
                )
            )
            .order_by(Task.id)
            .all()
        )

has_unfinished_tasks()

Check if there are any unfinished tasks in the database.

Source code in lilota/stores.py

def has_unfinished_tasks(self) -> bool:
    """Check if there are any unfinished tasks in the database."""
    with self._get_session() as session:
        return session.query(
            session.query(Task)
            .filter(
                Task.status.in_(
                    [TaskStatus.CREATED, TaskStatus.SCHEDULED, TaskStatus.RUNNING]
                )
            )
            .exists()
        ).scalar()

retry_tasks(batch_size=BATCH_SIZE)

Retry eligible tasks by creating new scheduled task instances.

This method scans for tasks that are eligible for retry based on their status, expiration, and retry limits. It uses optimistic locking to safely "claim" tasks in concurrent environments, ensuring that no task is retried more than once across multiple workers.

Eligible tasks include

• Tasks with status "failed" or "expired"
• Tasks with status "running" that have exceeded their expiration time

A task is only retried if

• It has not already been retried (retried_at is None)
• It has remaining retry attempts (attempts < max_attempts)

For each claimed task

• A new task is created with incremented attempt count
• The retry is scheduled with a delay proportional to the number of attempts
• The original task is marked with retried_at to prevent duplicate retries

Concurrency is handled using optimistic locking via the version field. If another worker updates the task first, the current worker skips it.

Parameters:

Name	Type	Description	Default
batch_size	int	Maximum number of tasks to process in a single run. Defaults to BATCH_SIZE.	BATCH_SIZE

Returns:

Name	Type	Description
int		The number of retry tasks successfully created.

Raises:

Type	Description
SQLAlchemyError	If a database error occurs during the transaction.

Notes

• Uses a linear backoff strategy: delay = 5 * current_attempts (seconds)
• All operations are executed within a single transactional session
• Safe for concurrent execution across multiple workers

Source code in lilota/stores.py

def retry_tasks(self, batch_size=BATCH_SIZE):
    """Retry eligible tasks by creating new scheduled task instances.

    This method scans for tasks that are eligible for retry based on their
    status, expiration, and retry limits. It uses optimistic locking to safely
    "claim" tasks in concurrent environments, ensuring that no task is retried
    more than once across multiple workers.

    Eligible tasks include:
        - Tasks with status "failed" or "expired"
        - Tasks with status "running" that have exceeded their expiration time

    A task is only retried if:
        - It has not already been retried (`retried_at` is None)
        - It has remaining retry attempts (`attempts < max_attempts`)

    For each claimed task:
        - A new task is created with incremented attempt count
        - The retry is scheduled with a delay proportional to the number of attempts
        - The original task is marked with `retried_at` to prevent duplicate retries

    Concurrency is handled using optimistic locking via the `version` field.
    If another worker updates the task first, the current worker skips it.

    Args:
        batch_size (int, optional):
            Maximum number of tasks to process in a single run.
            Defaults to BATCH_SIZE.

    Returns:
        int:
            The number of retry tasks successfully created.

    Raises:
        sqlalchemy.exc.SQLAlchemyError:
            If a database error occurs during the transaction.

    Notes:
        - Uses a linear backoff strategy: delay = 5 * current_attempts (seconds)
        - All operations are executed within a single transactional session
        - Safe for concurrent execution across multiple workers
    """
    now = datetime.now(timezone.utc)

    with self._get_session() as session:
        with session.begin():
            # 1. Select tasks
            stmt = (
                select(Task)
                .where(
                    and_(
                        Task.retried_at.is_(None),
                        Task.attempts < Task.max_attempts,
                        or_(
                            Task.status.in_(["failed", "expired"]),
                            and_(
                                Task.status == "running",
                                Task.expires_at.is_not(None),
                                Task.expires_at < now,
                            ),
                        ),
                    )
                )
                .order_by(Task.run_at)
                .limit(batch_size)
            )

            tasks = session.execute(stmt).scalars().all()

            if not tasks:
                return 0

            new_tasks = []

            for task in tasks:
                if task.retried_at is not None:
                    continue

                current_version = task.version
                current_attempts = task.attempts

                # 2. Try to "claim" via optimistic locking
                result = session.execute(
                    update(Task)
                    .where(
                        Task.id == task.id,
                        Task.version == current_version,
                        Task.retried_at.is_(None),
                    )
                    .values(retried_at=now, version=current_version + 1)
                )

                # 3. If no row updated: someone else took it
                if result.rowcount == 0:
                    continue

                # 4. Create retry task
                retry_task = Task(
                    name=task.name,
                    status=TaskStatus.CREATED,
                    run_at=now + timedelta(seconds=5 * current_attempts),
                    attempts=current_attempts + 1,
                    max_attempts=task.max_attempts,
                    previous_task_id=task.id,
                    timeout=task.timeout,
                    input=task.input,
                )

                new_tasks.append(retry_task)

            session.add_all(new_tasks)
            return len(new_tasks)

set_progress(id, progress)

Update the progress percentage of a task.

Parameters:

Name	Type	Description	Default
id	UUID	Id of the task.	required
progress	int	The progress of the task (0-100)	required

Source code in lilota/stores.py

def set_progress(self, id: UUID, progress: int):
    """Update the progress percentage of a task.

    Args:
        id (UUID): Id of the task.
        progress (int): The progress of the task (0-100)
    """
    with self._get_session() as session:
        with session.begin():
            task = self._load_task(session, id)
            task.progress_percentage = max(0, min(progress, 100))

start_task(id, max_attempts, timeout=None)

Mark a task as RUNNING and initialize metadata.

Parameters:

Name	Type	Description	Default
id	UUID	Id of the task.	required
max_attempts	int	Upper limit on how many times the task may be executed.	required
timeout	timedelta | None	Optional timeout that can be set for a task.	None

Returns:

Type	Description
Task	Task | None: The started task, or None if no task is available.

Source code in lilota/stores.py

def start_task(
    self, id: UUID, max_attempts: int, timeout: timedelta | None = None
) -> Task:
    """Mark a task as RUNNING and initialize metadata.

    Args:
        id (UUID): Id of the task.
        max_attempts (int): Upper limit on how many times the task may be executed.
        timeout (timedelta | None): Optional timeout that can be set for a task.

    Returns:
        Task | None: The started task, or None if no task is available.
    """
    expires_at = None
    start_date_time = datetime.now(timezone.utc)
    timeout_sec = int(timeout.total_seconds()) if timeout is not None else None

    with self._get_session() as session:
        with session.begin():
            task = self._load_task(session, id)
            task.timeout = timeout_sec

            if timeout is not None:
                expires_at = start_date_time + timeout

            task.pid = os.getpid()
            task.status = TaskStatus.RUNNING
            task.max_attempts = max_attempts
            task.progress_percentage = 0
            task.start_date_time = start_date_time
            task.expires_at = expires_at
            task.end_date_time = None
            return task

LogEntry

Bases: Base

Database model storing log messages generated by Lilota.

Source code in lilota/models.py

class LogEntry(Base):
    """Database model storing log messages generated by Lilota."""

    __tablename__ = "lilota_log"
    id: Mapped[int] = mapped_column(primary_key=True)
    created_at: Mapped[datetime] = mapped_column(
        UtcDateTime(),
        nullable=False,
    )
    level: Mapped[str] = mapped_column(String(20), nullable=False)
    logger: Mapped[str] = mapped_column(String(255), nullable=False)
    message: Mapped[str] = mapped_column(Text, nullable=False)
    process: Mapped[str | None] = mapped_column(String(64), default=None)
    thread: Mapped[str | None] = mapped_column(String(64), default=None)
    node_id: Mapped[UUID | None] = mapped_column(default=None)
    task_id: Mapped[UUID | None] = mapped_column(default=None)

ModelProtocol

Bases: Protocol

Protocol for models that can be serialized to dictionaries.

Source code in lilota/models.py

@runtime_checkable
class ModelProtocol(Protocol):
    """Protocol for models that can be serialized to dictionaries."""

    def as_dict(self) -> dict[str, Any]:
        """Return a dictionary representation of the model."""
        pass

as_dict()

Return a dictionary representation of the model.

Source code in lilota/models.py

def as_dict(self) -> dict[str, Any]:
    """Return a dictionary representation of the model."""
    pass

Node

Bases: Base

Database model representing a Lilota node.

A node represents a running component of the system such as a scheduler or a worker.

Source code in lilota/models.py

class Node(Base):
    """Database model representing a Lilota node.

    A node represents a running component of the system such as a
    scheduler or a worker.
    """

    __tablename__ = "lilota_node"
    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    name: Mapped[str | None] = mapped_column(String(255), nullable=True)
    type: Mapped[str] = mapped_column(String(32), nullable=False)
    status: Mapped[str] = mapped_column(String(32), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        UtcDateTime(),
        default=lambda: datetime.now(timezone.utc),
        nullable=False,
    )
    last_seen_at: Mapped[datetime] = mapped_column(
        UtcDateTime(),
        default=lambda: datetime.now(timezone.utc),
        nullable=False,
    )

    __table_args__ = (
        CheckConstraint(
            "type IN ('scheduler', 'worker')", name="lilota_node_type_check"
        ),
        CheckConstraint(
            "status IN ('idle', 'starting', 'running', 'stopping', 'stopped', 'crashed', 'dead')",
            name="lilota_node_status_check",
        ),
    )

NodeLeader

Bases: Base

Database model representing the worker node currently acting as leader.

The leader is responsible for cluster-level maintenance tasks such as cleaning up stale nodes.

Source code in lilota/models.py

class NodeLeader(Base):
    """Database model representing the worker node currently acting as leader.

    The leader is responsible for cluster-level maintenance tasks such as
    cleaning up stale nodes.
    """

    __tablename__ = "lilota_node_leader"
    id: Mapped[int] = mapped_column(Integer, primary_key=True, default=1)
    node_id: Mapped[UUID] = mapped_column(nullable=False)
    lease_expires_at: Mapped[datetime] = mapped_column(UtcDateTime(), nullable=False)

NodeStatus

Bases: StrEnum

Enumeration describing the lifecycle state of a node.

Source code in lilota/models.py

class NodeStatus(StrEnum):
    """Enumeration describing the lifecycle state of a node."""

    IDLE = "idle"
    STARTING = "starting"
    RUNNING = "running"
    STOPPING = "stopping"
    STOPPED = "stopped"
    CRASHED = "crashed"
    DEAD = "dead"

NodeType

Bases: StrEnum

Enumeration of supported node types.

Source code in lilota/models.py

class NodeType(StrEnum):
    """Enumeration of supported node types."""

    SCHEDULER = "scheduler"
    WORKER = "worker"

RegisteredTask

Wrapper representing a registered task function.

This class stores metadata about the task function such as input and output models and optionally a progress tracker.

It is responsible for: - deserializing the task input - executing the task function - serializing the output

Parameters:

Name	Type	Description	Default
func	Callable	Task function to execute.	required
input_model	Optional[Type]	Optional input model used to deserialize input payloads.	required
output_model	Optional[Type]	Optional output model used to serialize the task result.	required
input_param_name	Optional[str]	Name of the input parameter.	required
context_param_name	Optional[str]	Name of the context parameter.	required
timeout	Optional[timedelta]	Optional timeout that can be set for a task.	required
max_attempts	int	Upper limit on how many times the task may be executed.	required

Source code in lilota/models.py

class RegisteredTask:
    """Wrapper representing a registered task function.

    This class stores metadata about the task function such as input
    and output models and optionally a progress tracker.

    It is responsible for:
    - deserializing the task input
    - executing the task function
    - serializing the output

    Args:
      func (Callable): Task function to execute.
      input_model (Optional[Type]): Optional input model used to
        deserialize input payloads.
      output_model (Optional[Type]): Optional output model used to
        serialize the task result.
      input_param_name (Optional[str]): Name of the input parameter.
      context_param_name (Optional[str]): Name of the context parameter.
      timeout (Optional[timedelta]): Optional timeout that can be set for a task.
      max_attempts (int): Upper limit on how many times the task may be executed.
    """

    def __init__(
        self,
        func: Callable,
        input_model: Optional[Type],
        output_model: Optional[Type],
        input_param_name: Optional[str],
        context_param_name: Optional[str],
        timeout: Optional[timedelta],
        max_attempts: int,
    ):
        self.func = func
        self.input_model = input_model
        self.output_model = output_model
        self.input_param_name = input_param_name
        self.context_param_name = context_param_name
        self.timeout = timeout
        self.max_attempts = max_attempts

    def __call__(self, raw_input: Any, task_context: TaskContext):
        """Execute the registered task.

        Args:
          raw_input (Any): Raw input payload stored in the database.
          task_context (TaskContext): Task context that contains a TaskProgress object and a logger object.

        Returns:
          Any: Serialized task result.
        """

        # Deserialize input
        input_value = self._deserialize_input(raw_input)

        # Set the kwargs
        kwargs = {}

        # Set input parameter
        if input_value is not None:
            kwargs[self.input_param_name] = input_value

        # Set task context parameter
        if task_context is not None:
            kwargs[self.context_param_name] = task_context

        # Execute the function
        result = self.func(**kwargs)

        # Serialize output to JSON-safe dict
        return self._serialize_output(result)

    def _deserialize_input(self, raw_input: Any):
        if not self.input_model:
            return raw_input
        if isinstance(self.input_model, ModelProtocol):
            return self.input_model(**raw_input)
        if is_dataclass(self.input_model):
            return self.input_model(**raw_input)
        if isinstance(raw_input, dict):
            return raw_input
        if isinstance(raw_input, self.input_model):
            return raw_input
        raise TypeError(f"Unsupported input_model type: {self.input_model}")

    def _serialize_output(self, output: Any):
        if not self.output_model:
            return output
        if isinstance(output, dict):
            return output
        if isinstance(output, ModelProtocol):
            return output.as_dict()
        if isinstance(output, self.output_model):
            if is_dataclass(output):
                return asdict(output)
        if is_dataclass(self.output_model):
            return self.output_model(**output)
        return output

__call__(raw_input, task_context)

Execute the registered task.

Parameters:

Name	Type	Description	Default
raw_input	Any	Raw input payload stored in the database.	required
task_context	TaskContext	Task context that contains a TaskProgress object and a logger object.	required

Returns:

Name	Type	Description
Any		Serialized task result.

Source code in lilota/models.py

def __call__(self, raw_input: Any, task_context: TaskContext):
    """Execute the registered task.

    Args:
      raw_input (Any): Raw input payload stored in the database.
      task_context (TaskContext): Task context that contains a TaskProgress object and a logger object.

    Returns:
      Any: Serialized task result.
    """

    # Deserialize input
    input_value = self._deserialize_input(raw_input)

    # Set the kwargs
    kwargs = {}

    # Set input parameter
    if input_value is not None:
        kwargs[self.input_param_name] = input_value

    # Set task context parameter
    if task_context is not None:
        kwargs[self.context_param_name] = task_context

    # Execute the function
    result = self.func(**kwargs)

    # Serialize output to JSON-safe dict
    return self._serialize_output(result)

Task

Bases: Base

Database model representing a scheduled task.

Tasks are created by the scheduler and executed by worker nodes. The model stores the execution state, input/output data, and runtime metadata.

Source code in lilota/models.py

class Task(Base):
    """Database model representing a scheduled task.

    Tasks are created by the scheduler and executed by worker nodes.
    The model stores the execution state, input/output data, and
    runtime metadata.
    """

    __tablename__ = "lilota_task"
    id: Mapped[UUID] = mapped_column(primary_key=True, default=uuid4)
    name: Mapped[str] = mapped_column(String, nullable=False)
    pid: Mapped[int] = mapped_column(nullable=False, default=0)
    status: Mapped[str] = mapped_column(String(32), nullable=False, index=False)
    run_at: Mapped[datetime] = mapped_column(
        UtcDateTime(),
        nullable=False,
        default=lambda: datetime.now(timezone.utc),
    )
    attempts: Mapped[int] = mapped_column(nullable=False, default=0)
    max_attempts: Mapped[int] = mapped_column(nullable=False, default=1)
    retried_at: Mapped[datetime | None] = mapped_column(
        UtcDateTime(), nullable=True, default=None, index=True
    )
    previous_task_id: Mapped[UUID] = mapped_column(nullable=True, default=None)
    timeout: Mapped[int | None] = mapped_column(Integer, nullable=True, default=None)
    expires_at: Mapped[datetime | None] = mapped_column(
        UtcDateTime(), nullable=True, default=None
    )
    progress_percentage: Mapped[int] = mapped_column(default=0)
    start_date_time: Mapped[datetime | None] = mapped_column(
        UtcDateTime(), nullable=True, default=None
    )
    end_date_time: Mapped[datetime | None] = mapped_column(
        UtcDateTime(), nullable=True, default=None
    )
    input: Mapped[Any | None] = mapped_column(JSON)
    output: Mapped[Any | None] = mapped_column(JSON)
    error: Mapped[Any | None] = mapped_column(JSON)
    locked_by: Mapped[UUID | None] = mapped_column(default=None)
    locked_at: Mapped[datetime | None] = mapped_column(
        UtcDateTime(), nullable=True, default=None
    )
    version: Mapped[int] = mapped_column(default=0, nullable=False)

    __table_args__ = (
        CheckConstraint(
            "status IN ('created', 'scheduled', 'running', 'completed', 'failed', 'expired', 'cancelled')",
            name="lilota_task_status_check",
        ),
        Index("idx_get_next_task", "status", "run_at"),
        Index("idx_retryable_tasks", "retried_at", "status", "run_at"),
    )

    def __repr__(self):
        return f"<TaskInfo(id={self.id}, name={self.name}, progress={self.progress_percentage}%)>"

TaskContext

Runtime context injected into Lilota tasks.

The task context provides access to runtime-only services and metadata during task execution. It is created by the worker runtime and injected automatically into task functions that declare a TaskContext parameter.

Example

@lilota.task def process(data: InputModel, ctx: TaskContext): ctx.logger.info("starting") ctx.progress.update(50)

Attributes:

Name	Type	Description
task_id	Optional[str]	Unique identifier of the currently executing task.
progress	Optional[TaskProgress]	Progress tracking interface for updating task execution progress.
logger	Optional[Logger]	Logger instance associated with the current task execution.

Source code in lilota/models.py

class TaskContext:
    """Runtime context injected into Lilota tasks.

    The task context provides access to runtime-only services and metadata
    during task execution. It is created by the worker runtime and injected
    automatically into task functions that declare a ``TaskContext`` parameter.

    Example:
        @lilota.task
        def process(data: InputModel, ctx: TaskContext):
            ctx.logger.info("starting")
            ctx.progress.update(50)

    Attributes:
        task_id (Optional[str]):
            Unique identifier of the currently executing task.

        progress (Optional[TaskProgress]):
            Progress tracking interface for updating task execution progress.

        logger (Optional[logging.Logger]):
            Logger instance associated with the current task execution.
    """

    def __init__(
        self,
        *,
        task_id=None,
        progress: TaskProgress = None,
        logger: logging.Logger = None,
    ):
        """Initialize a task execution context.

        Args:
            task_id (Optional[str]):
                Unique identifier of the currently executing task.

            progress (Optional[TaskProgress]):
                Progress tracking interface for updating task progress.

            logger (Optional[logging.Logger]):
                Logger instance associated with the task execution.
        """
        self.task_id = task_id
        self.progress = progress
        self.logger = logger

__init__(*, task_id=None, progress=None, logger=None)

Initialize a task execution context.

Parameters:

Name	Type	Description	Default
task_id	Optional[str]	Unique identifier of the currently executing task.	None
progress	Optional[TaskProgress]	Progress tracking interface for updating task progress.	None
logger	Optional[Logger]	Logger instance associated with the task execution.	None

Source code in lilota/models.py

def __init__(
    self,
    *,
    task_id=None,
    progress: TaskProgress = None,
    logger: logging.Logger = None,
):
    """Initialize a task execution context.

    Args:
        task_id (Optional[str]):
            Unique identifier of the currently executing task.

        progress (Optional[TaskProgress]):
            Progress tracking interface for updating task progress.

        logger (Optional[logging.Logger]):
            Logger instance associated with the task execution.
    """
    self.task_id = task_id
    self.progress = progress
    self.logger = logger

TaskProgress

Helper object used to update task progress.

This object is passed to task functions when progress tracking is enabled. It allows the task to report its current progress to the task store.

Parameters:

Name	Type	Description	Default
task_id	int	Identifier of the task.	required
set_progress	Callable[[int, int], None]	Function used to persist progress updates.	required

Source code in lilota/models.py

class TaskProgress:
    """Helper object used to update task progress.

    This object is passed to task functions when progress tracking is
    enabled. It allows the task to report its current progress to the
    task store.

    Args:
      task_id (int): Identifier of the task.
      set_progress (Callable[[int, int], None]): Function used to persist
        progress updates.
    """

    def __init__(self, task_id: int, set_progress: Callable[[int, int], None]):
        self.task_id = task_id
        self.set_progress = set_progress
        self.set_progress_manually = False

    def set(self, progress: int):
        """Update the progress of the task.

        Args:
          progress (int): Progress value (typically 0–100).
        """
        self.set_progress(self.task_id, progress)
        self.set_progress_manually = True

set(progress)

Update the progress of the task.

Parameters:

Name	Type	Description	Default
progress	int	Progress value (typically 0–100).	required

Source code in lilota/models.py

def set(self, progress: int):
    """Update the progress of the task.

    Args:
      progress (int): Progress value (typically 0–100).
    """
    self.set_progress(self.task_id, progress)
    self.set_progress_manually = True

TaskStatus

Bases: StrEnum

Enumeration of possible task states.

Source code in lilota/models.py

class TaskStatus(StrEnum):
    """Enumeration of possible task states."""

    CREATED = "created"
    SCHEDULED = "scheduled"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    EXPIRED = "expired"
    CANCELLED = "cancelled"

Heartbeat

Bases: Thread

Background thread that executes a heartbeat task periodically.

The heartbeat repeatedly calls the provided :class:HeartbeatTask at the configured interval until the thread is stopped.

Source code in lilota/heartbeat.py

class Heartbeat(threading.Thread):
    """Background thread that executes a heartbeat task periodically.

    The heartbeat repeatedly calls the provided :class:`HeartbeatTask`
    at the configured interval until the thread is stopped.
    """

    def __init__(self, name: str, task: HeartbeatTask, logger: logging.Logger):
        """Initialize the heartbeat thread.

        Args:
          name (str): Name of the thread.
          task (HeartbeatTask): Task executed periodically by the heartbeat.
          logger (logging.Logger): Logger used for reporting errors.
        """
        super().__init__(name=name, daemon=True)
        self._task = task
        self._logger = logger
        self._stop_event = threading.Event()

    def run(self) -> None:
        """Run the heartbeat loop.

        The loop repeatedly executes the configured heartbeat task and
        waits for the defined interval before executing it again. The
        loop stops when the stop event is triggered.
        """
        while not self._stop_event.is_set():
            # Execute
            try:
                self._task.execute()
            except Exception:
                if self._logger is not None:
                    self._logger.exception("Heartbeat task failed")
                else:
                    raise

            # Wait
            interval = max(0.0, self._task.interval)
            if self._task.jitter is not None:
                jitter = random.uniform(
                    -self._task.jitter * interval, self._task.jitter * interval
                )
                interval = interval + jitter
            self._stop_event.wait(interval)

    def stop(self) -> None:
        """Signal the heartbeat thread to stop.

        The thread will stop after the current execution cycle completes.
        """
        self._stop_event.set()

    def stop_and_join(self, timeout=None):
        """Stop the heartbeat thread and wait for it to finish.

        Args:
          timeout (float | None, optional): Maximum number of seconds to
            wait for the thread to terminate. If ``None``, wait indefinitely.
        """
        self.stop()
        self.join(timeout)
        if self.is_alive():
            raise RuntimeError("Heartbeat thread did not stop!")

__init__(name, task, logger)

Initialize the heartbeat thread.

Parameters:

Name	Type	Description	Default
name	str	Name of the thread.	required
task	HeartbeatTask	Task executed periodically by the heartbeat.	required
logger	Logger	Logger used for reporting errors.	required

Source code in lilota/heartbeat.py

def __init__(self, name: str, task: HeartbeatTask, logger: logging.Logger):
    """Initialize the heartbeat thread.

    Args:
      name (str): Name of the thread.
      task (HeartbeatTask): Task executed periodically by the heartbeat.
      logger (logging.Logger): Logger used for reporting errors.
    """
    super().__init__(name=name, daemon=True)
    self._task = task
    self._logger = logger
    self._stop_event = threading.Event()

run()

Run the heartbeat loop.

The loop repeatedly executes the configured heartbeat task and waits for the defined interval before executing it again. The loop stops when the stop event is triggered.

Source code in lilota/heartbeat.py

def run(self) -> None:
    """Run the heartbeat loop.

    The loop repeatedly executes the configured heartbeat task and
    waits for the defined interval before executing it again. The
    loop stops when the stop event is triggered.
    """
    while not self._stop_event.is_set():
        # Execute
        try:
            self._task.execute()
        except Exception:
            if self._logger is not None:
                self._logger.exception("Heartbeat task failed")
            else:
                raise

        # Wait
        interval = max(0.0, self._task.interval)
        if self._task.jitter is not None:
            jitter = random.uniform(
                -self._task.jitter * interval, self._task.jitter * interval
            )
            interval = interval + jitter
        self._stop_event.wait(interval)

stop()

Signal the heartbeat thread to stop.

The thread will stop after the current execution cycle completes.

Source code in lilota/heartbeat.py

def stop(self) -> None:
    """Signal the heartbeat thread to stop.

    The thread will stop after the current execution cycle completes.
    """
    self._stop_event.set()

stop_and_join(timeout=None)

Stop the heartbeat thread and wait for it to finish.

Parameters:

Name	Type	Description	Default
timeout	float | None	Maximum number of seconds to wait for the thread to terminate. If None, wait indefinitely.	None

Source code in lilota/heartbeat.py

def stop_and_join(self, timeout=None):
    """Stop the heartbeat thread and wait for it to finish.

    Args:
      timeout (float | None, optional): Maximum number of seconds to
        wait for the thread to terminate. If ``None``, wait indefinitely.
    """
    self.stop()
    self.join(timeout)
    if self.is_alive():
        raise RuntimeError("Heartbeat thread did not stop!")

HeartbeatTask

Bases: ABC

Abstract base class for heartbeat tasks.

A heartbeat task defines logic that should be executed periodically by a :class:Heartbeat thread.

Source code in lilota/heartbeat.py

class HeartbeatTask(ABC):
    """Abstract base class for heartbeat tasks.

    A heartbeat task defines logic that should be executed periodically
    by a :class:`Heartbeat` thread.
    """

    def __init__(self, interval: float, jitter: float):
        """Initialize the heartbeat task.

        Args:
          interval (float): Interval in seconds between task executions.
          jitter (float): Jitter that is used for the heartbeat interval.
        """
        self.interval = interval
        self.jitter = jitter

    @abstractmethod
    def execute(self):
        """Execute the heartbeat logic.

        Subclasses must implement this method to define the work performed
        during each heartbeat cycle.
        """
        pass

__init__(interval, jitter)

Initialize the heartbeat task.

Parameters:

Name	Type	Description	Default
interval	float	Interval in seconds between task executions.	required
jitter	float	Jitter that is used for the heartbeat interval.	required

Source code in lilota/heartbeat.py

def __init__(self, interval: float, jitter: float):
    """Initialize the heartbeat task.

    Args:
      interval (float): Interval in seconds between task executions.
      jitter (float): Jitter that is used for the heartbeat interval.
    """
    self.interval = interval
    self.jitter = jitter

execute() abstractmethod

Execute the heartbeat logic.

Subclasses must implement this method to define the work performed during each heartbeat cycle.

Source code in lilota/heartbeat.py

@abstractmethod
def execute(self):
    """Execute the heartbeat logic.

    Subclasses must implement this method to define the work performed
    during each heartbeat cycle.
    """
    pass

ContextLogger

Bases: LoggerAdapter

Logger adapter that automatically injects contextual metadata.

This adapter attaches additional context such as node_id and task_id to log records so that log entries can be associated with specific nodes or tasks.

Source code in lilota/logging.py

class ContextLogger(logging.LoggerAdapter):
    """Logger adapter that automatically injects contextual metadata.

    This adapter attaches additional context such as ``node_id`` and
    ``task_id`` to log records so that log entries can be associated
    with specific nodes or tasks.
    """

    def process(self, msg, kwargs):
        """Inject contextual metadata into the log record.

        Args:
          msg (str): Log message.
          kwargs (dict): Keyword arguments passed to the logger.

        Returns:
          tuple: Processed ``(msg, kwargs)`` pair with injected metadata.
        """
        kwargs.setdefault("extra", {})
        kwargs["extra"].setdefault("node_id", self.extra.get("node_id"))
        kwargs["extra"].setdefault("task_id", self.extra.get("task_id"))
        return msg, kwargs

process(msg, kwargs)

Inject contextual metadata into the log record.

Parameters:

Name	Type	Description	Default
msg	str	Log message.	required
kwargs	dict	Keyword arguments passed to the logger.	required

Returns:

Name	Type	Description
tuple		Processed (msg, kwargs) pair with injected metadata.

Source code in lilota/logging.py

def process(self, msg, kwargs):
    """Inject contextual metadata into the log record.

    Args:
      msg (str): Log message.
      kwargs (dict): Keyword arguments passed to the logger.

    Returns:
      tuple: Processed ``(msg, kwargs)`` pair with injected metadata.
    """
    kwargs.setdefault("extra", {})
    kwargs["extra"].setdefault("node_id", self.extra.get("node_id"))
    kwargs["extra"].setdefault("task_id", self.extra.get("task_id"))
    return msg, kwargs

LilotaLoggingFilter

Bases: Filter

Logging filter used to suppress noisy third-party logs.

Currently filters Alembic logs so that only warnings and errors are recorded.

Source code in lilota/logging.py

class LilotaLoggingFilter(logging.Filter):
    """Logging filter used to suppress noisy third-party logs.

    Currently filters Alembic logs so that only warnings and errors
    are recorded.
    """

    def filter(self, record: logging.LogRecord) -> bool:
        """Determine whether a log record should be processed.

        Args:
          record (logging.LogRecord): Log record being evaluated.

        Returns:
          bool: ``True`` if the record should be logged, otherwise ``False``.
        """
        if record.name.startswith("alembic."):
            return record.levelno >= logging.WARNING
        return True

filter(record)

Determine whether a log record should be processed.

Parameters:

Name	Type	Description	Default
record	LogRecord	Log record being evaluated.	required

Returns:

Name	Type	Description
bool	bool	True if the record should be logged, otherwise False.

Source code in lilota/logging.py

def filter(self, record: logging.LogRecord) -> bool:
    """Determine whether a log record should be processed.

    Args:
      record (logging.LogRecord): Log record being evaluated.

    Returns:
      bool: ``True`` if the record should be logged, otherwise ``False``.
    """
    if record.name.startswith("alembic."):
        return record.levelno >= logging.WARNING
    return True

SqlAlchemyHandler

Bases: Handler

Logging handler that stores log records in the database.

This handler writes log messages to the lilota_log table using the provided :class:SqlAlchemyLogStore. Each log record is converted into a :class:LogEntry model instance.

Source code in lilota/logging.py

class SqlAlchemyHandler(logging.Handler):
    """Logging handler that stores log records in the database.

    This handler writes log messages to the ``lilota_log`` table using
    the provided :class:`SqlAlchemyLogStore`. Each log record is converted
    into a :class:`LogEntry` model instance.
    """

    def __init__(self, log_store: LogStore):
        """Initialize the logging handler.

        Args:
          log_store (SqlAlchemyLogStore): Store used to persist log entries.
        """
        super().__init__()
        self.log_store: LogStore = log_store

    def emit(self, record: logging.LogRecord) -> None:
        """Persist a log record in the database.

        Args:
          record (logging.LogRecord): Log record to store.
        """
        with self.log_store.get_session() as session:
            entry = LogEntry(
                created_at=datetime.fromtimestamp(record.created),
                level=record.levelname,
                logger=record.name,
                message=self.format(record),
                process=record.processName,
                thread=record.threadName,
                node_id=getattr(record, "node_id", None),
                task_id=getattr(record, "task_id", None),
            )
            session.add(entry)
            session.commit()

__init__(log_store)

Initialize the logging handler.

Parameters:

Name	Type	Description	Default
log_store	SqlAlchemyLogStore	Store used to persist log entries.	required

Source code in lilota/logging.py

def __init__(self, log_store: LogStore):
    """Initialize the logging handler.

    Args:
      log_store (SqlAlchemyLogStore): Store used to persist log entries.
    """
    super().__init__()
    self.log_store: LogStore = log_store

emit(record)

Persist a log record in the database.

Parameters:

Name	Type	Description	Default
record	LogRecord	Log record to store.	required

Source code in lilota/logging.py

def emit(self, record: logging.LogRecord) -> None:
    """Persist a log record in the database.

    Args:
      record (logging.LogRecord): Log record to store.
    """
    with self.log_store.get_session() as session:
        entry = LogEntry(
            created_at=datetime.fromtimestamp(record.created),
            level=record.levelname,
            logger=record.name,
            message=self.format(record),
            process=record.processName,
            thread=record.threadName,
            node_id=getattr(record, "node_id", None),
            task_id=getattr(record, "task_id", None),
        )
        session.add(entry)
        session.commit()

configure_logging(db_url, logger_name, logging_level)

Configure a Lilota logger that writes log messages to the database.

This function creates and configures a logger with the given name. The logger uses :class:SqlAlchemyHandler to persist log records in the database through :class:SqlAlchemyLogStore. Any existing handlers attached to the logger are removed before configuration.

Parameters:

Name	Type	Description	Default
db_url	str	Database connection URL used by the log store.	required
logger_name	str	Name of the logger to configure.	required
logging_level	int	Logging level to apply to both the logger and its handler.	required

Returns:

Type	Description
Logger	logging.Logger: Configured logger instance that writes log messages to the database.

Source code in lilota/logging.py

def configure_logging(
    db_url: str, logger_name: str, logging_level: int
) -> logging.Logger:
    """Configure a Lilota logger that writes log messages to the database.

    This function creates and configures a logger with the given name. The
    logger uses :class:`SqlAlchemyHandler` to persist log records in the
    database through :class:`SqlAlchemyLogStore`. Any existing handlers
    attached to the logger are removed before configuration.

    Args:
      db_url (str):
        Database connection URL used by the log store.

      logger_name (str):
        Name of the logger to configure.

      logging_level (int):
        Logging level to apply to both the logger and its handler.

    Returns:
      logging.Logger:
        Configured logger instance that writes log messages to the database.
    """
    logger = logging.getLogger(logger_name)
    logger.setLevel(logging_level)
    logger.handlers.clear()
    db_handler = SqlAlchemyHandler(LogStore(db_url))
    db_handler.setLevel(logging_level)
    db_handler.setFormatter(logging.Formatter("%(message)s"))
    db_handler.addFilter(LilotaLoggingFilter())
    logger.addHandler(db_handler)
    return logger

create_context_logger(base_logger, **kwargs)

Create a context-aware logger.

The returned logger automatically attaches contextual metadata (such as node_id and task_id) to all emitted log records.

Parameters:

Name	Type	Description	Default
base_logger	Logger	Base logger instance.	required
**kwargs		Optional context values (e.g., node_id, task_id).	{}

Returns:

Name	Type	Description
ContextLogger		Logger adapter with contextual metadata.

Source code in lilota/logging.py

def create_context_logger(base_logger: logging.Logger, **kwargs):
    """Create a context-aware logger.

    The returned logger automatically attaches contextual metadata
    (such as ``node_id`` and ``task_id``) to all emitted log records.

    Args:
      base_logger (logging.Logger): Base logger instance.
      **kwargs: Optional context values (e.g., ``node_id``, ``task_id``).

    Returns:
      ContextLogger: Logger adapter with contextual metadata.
    """
    extra = {}

    if "node_id" in kwargs:
        extra["node_id"] = kwargs["node_id"]

    if "task_id" in kwargs:
        extra["task_id"] = kwargs["task_id"]

    return ContextLogger(
        base_logger,
        extra,
    )

error_to_dict(error_message)

Wrap an error message string into a dictionary.

Parameters:

Name	Type	Description	Default
error_message	str	The error message to wrap.	required

Returns:

Name	Type	Description
dict	dict	A dictionary with key "message" containing the error message.

Source code in lilota/utils.py

def error_to_dict(error_message: str) -> dict:
    """Wrap an error message string into a dictionary.

    Args:
        error_message (str): The error message to wrap.

    Returns:
        dict: A dictionary with key "message" containing the error message.
    """
    return {"message": error_message}

exception_to_dict(ex)

Convert an exception into a dictionary containing type, message, and traceback.

Parameters:

Name	Type	Description	Default
ex	Exception	The exception to convert.	required

Returns:

Name	Type	Description
dict	dict	A dictionary with keys: - "type": Exception class name. - "message": Exception message string. - "traceback": Formatted traceback string.

Source code in lilota/utils.py

def exception_to_dict(ex: Exception) -> dict:
    """Convert an exception into a dictionary containing type, message, and traceback.

    Args:
        ex (Exception): The exception to convert.

    Returns:
        dict: A dictionary with keys:
            - "type": Exception class name.
            - "message": Exception message string.
            - "traceback": Formatted traceback string.
    """
    return {
        "type": ex.__class__.__name__,
        "message": str(ex),
        "traceback": traceback.format_exc(),
    }

normalize_data(data)

Normalize input data to a dictionary for storage or serialization.

Supports dict, ModelProtocol objects, and dataclasses.

Parameters:

Name	Type	Description	Default
data	Any	The input data to normalize.	required

Returns:

Name	Type	Description
dict	dict	A dictionary representation of the data.

Raises:

Type	Description
TypeError	If data is not a dict, ModelProtocol, or dataclass.

Source code in lilota/utils.py

def normalize_data(data: Any) -> dict:
    """Normalize input data to a dictionary for storage or serialization.

    Supports `dict`, `ModelProtocol` objects, and dataclasses.

    Args:
        data (Any): The input data to normalize.

    Returns:
        dict: A dictionary representation of the data.

    Raises:
        TypeError: If `data` is not a `dict`, `ModelProtocol`, or dataclass.
    """
    # Dict
    if isinstance(data, dict):
        return data

    # ModelProtocol
    if isinstance(data, ModelProtocol):
        return data.as_dict()

    # Dataclass
    if is_dataclass(data):
        return asdict(data)

    raise TypeError(
        f"Unsupported type: {type(data).__name__}. Expected ModelProtocol, dataclass, or dict."
    )