Runner module

Database Task Runner

A task execution manager that integrates Luigi task scheduling with database operations in the ROMI project, enabling automated processing of plant scans either individually or in batch mode.

Key Features

• Database-aware task execution system
• Support for single or multiple task execution
• Batch processing capability for all scans in a database
• Automatic database connection management
• Luigi configuration integration
• Flexible task scheduling with local scheduler
• Logging functionality for execution tracking

Usage Examples

from plantdb.commons.fsdb.core import FSDB from romitask.runner import DBRunner from romitask.task import DummyTask db = FSDB('path/to/database') task = DummyTask config = {'TaskName': {'param1': 'value1'}} runner = DBRunner(db, task, config)

Run on specific scan

runner.run_scan('scan_001')

Run on all scans

runner.run()

DBRunner(db, tasks, config)

Bases: object

A class for executing Luigi tasks on a plant database.

This class provides functionality to run one or more Luigi tasks either on a single scan or all scans in a plant database. It handles database connections and Luigi configuration management.

Attributes:

Name	Type	Description
db	DB	The database instance being used.
tasks	list of RomiTask	List of task classes to be executed.

Examples:

>>> from plantdb.commons.fsdb.core import FSDB
>>> from romitask.runner import DBRunner
>>> from romitask.task import DummyTask
>>> db = FSDB('path/to/database')
>>> task = DummyTask
>>> config = {'TaskName': {'param1': 'value1'}}
>>> runner = DBRunner(db, task, config)
>>>
>>> # Run on specific scan
>>> runner.run_scan('scan_001')
>>>
>>> # Run on all scans
>>> runner.run()

Initialize the runner with a database, tasks, and configuration.

Parameters:

Name	Type	Description	Default
db	DB	The target database instance to run tasks on.	required
tasks	RomiTask or list of RomiTask	Single task or list of tasks to execute. Each task must be a Luigi task class (not instantiated).	required
config	dict	Luigi configuration dictionary containing task-specific settings.	required

Source code in romitask/runner.py

def __init__(self, db, tasks, config):
    """Initialize the runner with a database, tasks, and configuration.

    Parameters
    ----------
    db : plantdb.commons.db.DB
        The target database instance to run tasks on.
    tasks : RomiTask or list of RomiTask
        Single task or list of tasks to execute. Each task must be a Luigi task class
        (not instantiated).
    config : dict
        Luigi configuration dictionary containing task-specific settings.
    """
    if not isinstance(tasks, (list, tuple)):
        tasks = [tasks]
    self.db = db
    self.tasks = tasks
    luigi_config = luigi.configuration.get_config()
    luigi_config.read_dict(config)

run()

Execute tasks on all scans in the database.

This method iterates through all scans in the database and executes the configured tasks on each scan sequentially.

Notes

This method: 1. Establishes database connection 2. Iterates through all scans 3. Executes tasks on each scan 4. Closes the database connection

The process is logged, with scan IDs printed to the info log level.

Source code in romitask/runner.py

def run(self):
    """Execute tasks on all scans in the database.

    This method iterates through all scans in the database and executes
    the configured tasks on each scan sequentially.

    Notes
    -----
    This method:
    1. Establishes database connection
    2. Iterates through all scans
    3. Executes tasks on each scan
    4. Closes the database connection

    The process is logged, with scan IDs printed to the info log level.
    """
    self.db.connect()
    for scan in self.db.get_scans():
        logger.info(f"scan = {scan.id}")
        self._run_scan_connected(scan)
    logger.info("Done")
    self.db.disconnect()
    return

run_scan(scan_id)

Execute tasks on a single scan in the database.

Parameters:

Name	Type	Description	Default
scan_id	str	Identifier of the scan to process.	required

Notes

This method: 1. Establishes database connection 2. Retrieves the specified scan 3. Executes tasks on the scan 4. Closes the database connection

Source code in romitask/runner.py

def run_scan(self, scan_id):
    """Execute tasks on a single scan in the database.

    Parameters
    ----------
    scan_id : str
        Identifier of the scan to process.

    Notes
    -----
    This method:
    1. Establishes database connection
    2. Retrieves the specified scan
    3. Executes tasks on the scan
    4. Closes the database connection
    """
    self.db.connect()
    scan = self.db.get_scan(scan_id)
    self._run_scan_connected(scan)
    self.db.disconnect()
    return