Skip to content

trestle.tasks.base_task

trestle.tasks.base_task ¤

Trestle tasks base templating.

Attributes¤

logger = logging.getLogger(__name__) module-attribute ¤

Classes¤

PassFail ¤

Bases: TaskBase


              flowchart TD
              trestle.tasks.base_task.PassFail[PassFail]
              trestle.tasks.base_task.TaskBase[TaskBase]

                              trestle.tasks.base_task.TaskBase --> trestle.tasks.base_task.PassFail
                


              click trestle.tasks.base_task.PassFail href "" "trestle.tasks.base_task.PassFail"
              click trestle.tasks.base_task.TaskBase href "" "trestle.tasks.base_task.TaskBase"
            

Holding pattern template for a task which does nothing and always passes.

Attributes:

Name Type Description
name

Name of the task.

Source code in trestle/tasks/base_task.py
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class PassFail(TaskBase):
    """
    Holding pattern template for a task which does nothing and always passes.

    Attributes:
        name: Name of the task.
    """

    name = 'pass-fail'

    def __init__(self, config_object: Optional[configparser.SectionProxy]) -> None:
        """
        Initialize trestle task pass-fail.

        Attributes:
            config_object: Config section associated with the task.
        """
        super().__init__(config_object)

    def print_info(self) -> None:
        """Print the help string."""
        logger.info(f'Help information for {self.name} task.')
        logger.info('This is a template task which reports pass fail depending on the specific configuration.')
        logger.info(
            'In this case if no config section is provided the task will fail. This is a task specific behavior.'
        )
        logger.info('Configuration flags sit under [task.pass-fail]')
        logger.info('with two boolean flags')
        logger.info('execute_status = True/False with a default pass')
        logger.info('simulate_status = True/False with a default fail')
        logger.info('Note that if the config file does not have the appropriate section this should fail.')
        logger.info('The principle goal is a simple development example.')

    def simulate(self) -> TaskOutcome:
        """Provide a simulated outcome."""
        if self._config:
            outcome = self._config.getboolean('simulate_status', fallback=True)
            if outcome:
                return TaskOutcome('simulated-success')
        return TaskOutcome('simulated-failure')

    def execute(self) -> TaskOutcome:
        """Provide a actual outcome."""
        if self._config:
            outcome = self._config.getboolean('execute_status', fallback=True)
            if outcome:
                return TaskOutcome('success')
        return TaskOutcome('failure')
Attributes¤
name = 'pass-fail' class-attribute instance-attribute ¤
Methods:¤
__init__(config_object) ¤

Initialize trestle task pass-fail.

Attributes:

Name Type Description
config_object

Config section associated with the task.

Source code in trestle/tasks/base_task.py
114
115
116
117
118
119
120
121
def __init__(self, config_object: Optional[configparser.SectionProxy]) -> None:
    """
    Initialize trestle task pass-fail.

    Attributes:
        config_object: Config section associated with the task.
    """
    super().__init__(config_object)
execute() ¤

Provide a actual outcome.

Source code in trestle/tasks/base_task.py
145
146
147
148
149
150
151
def execute(self) -> TaskOutcome:
    """Provide a actual outcome."""
    if self._config:
        outcome = self._config.getboolean('execute_status', fallback=True)
        if outcome:
            return TaskOutcome('success')
    return TaskOutcome('failure')
print_info() ¤

Print the help string.

Source code in trestle/tasks/base_task.py
123
124
125
126
127
128
129
130
131
132
133
134
135
def print_info(self) -> None:
    """Print the help string."""
    logger.info(f'Help information for {self.name} task.')
    logger.info('This is a template task which reports pass fail depending on the specific configuration.')
    logger.info(
        'In this case if no config section is provided the task will fail. This is a task specific behavior.'
    )
    logger.info('Configuration flags sit under [task.pass-fail]')
    logger.info('with two boolean flags')
    logger.info('execute_status = True/False with a default pass')
    logger.info('simulate_status = True/False with a default fail')
    logger.info('Note that if the config file does not have the appropriate section this should fail.')
    logger.info('The principle goal is a simple development example.')
simulate() ¤

Provide a simulated outcome.

Source code in trestle/tasks/base_task.py
137
138
139
140
141
142
143
def simulate(self) -> TaskOutcome:
    """Provide a simulated outcome."""
    if self._config:
        outcome = self._config.getboolean('simulate_status', fallback=True)
        if outcome:
            return TaskOutcome('simulated-success')
    return TaskOutcome('simulated-failure')

TaskBase ¤

Bases: ABC


              flowchart TD
              trestle.tasks.base_task.TaskBase[TaskBase]

              

              click trestle.tasks.base_task.TaskBase href "" "trestle.tasks.base_task.TaskBase"
            

Abstract base class for tasks.

Attributes:

Name Type Description
name str

Name of the task.

Source code in trestle/tasks/base_task.py
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class TaskBase(ABC):
    """
    Abstract base class for tasks.

    Attributes:
        name: Name of the task.
    """

    name: str = 'base'

    def __init__(self, config_object: Optional[configparser.SectionProxy]) -> None:
        """Initialize task base and store config."""
        self._config = config_object

    def _configure_csv_common(self) -> Tuple[bool, Optional[str]]:
        """Configure common CSV task fields shared across CSV-based tasks.

        Sets self._timestamp, self._quiet, self._verbose, self._title,
        self._version, self._csv_file, self._csv_path, self._workspace.

        Returns:
            Tuple of (success, error_message). success is True when all
            required fields are present and valid, False otherwise.
        """
        self._timestamp = datetime.datetime.now(datetime.UTC).replace(microsecond=0).isoformat()
        # config verbosity
        self._quiet = self._config.get('quiet', False)
        self._verbose = not self._quiet
        # title
        self._title = self._config.get('title')
        if self._title is None:
            return False, 'config missing "title"'
        # version
        self._version = self._config.get('version')
        if self._version is None:
            return False, 'config missing "version"'
        # config csv
        self._csv_file = self._config.get('csv-file')
        if self._csv_file is None:
            return False, 'config missing "csv-file"'
        self._csv_path = pathlib.Path(self._csv_file)
        if not self._csv_path.exists():
            return False, '"csv-file" not found'
        # announce csv
        if self._verbose:
            logger.info(f'input: {self._csv_file}')
        # workspace
        self._workspace = os.getcwd()
        return True, None

    @abstractmethod
    def print_info(self) -> None:
        """Print the help string."""

    @abstractmethod
    def execute(self) -> TaskOutcome:
        """Execute the task including potential rollback."""

    @abstractmethod
    def simulate(self) -> TaskOutcome:
        """Simulate the task and report task outcome."""
Attributes¤
name = 'base' class-attribute instance-attribute ¤
Methods:¤
__init__(config_object) ¤

Initialize task base and store config.

Source code in trestle/tasks/base_task.py
51
52
53
def __init__(self, config_object: Optional[configparser.SectionProxy]) -> None:
    """Initialize task base and store config."""
    self._config = config_object
execute() abstractmethod ¤

Execute the task including potential rollback.

Source code in trestle/tasks/base_task.py
95
96
97
@abstractmethod
def execute(self) -> TaskOutcome:
    """Execute the task including potential rollback."""
print_info() abstractmethod ¤

Print the help string.

Source code in trestle/tasks/base_task.py
91
92
93
@abstractmethod
def print_info(self) -> None:
    """Print the help string."""
simulate() abstractmethod ¤

Simulate the task and report task outcome.

Source code in trestle/tasks/base_task.py
 99
100
101
@abstractmethod
def simulate(self) -> TaskOutcome:
    """Simulate the task and report task outcome."""

TaskOutcome ¤

Bases: Enum


              flowchart TD
              trestle.tasks.base_task.TaskOutcome[TaskOutcome]

              

              click trestle.tasks.base_task.TaskOutcome href "" "trestle.tasks.base_task.TaskOutcome"
            

Enum describing possible task outcomes.

Source code in trestle/tasks/base_task.py
30
31
32
33
34
35
36
37
38
class TaskOutcome(Enum):
    """Enum describing possible task outcomes."""

    SUCCESS = 'success'
    FAILURE = 'failure'
    ROLLEDBACK = 'rolledback'
    SIM_SUCCESS = 'simulated-success'
    SIM_FAILURE = 'simulated-failure'
    NOT_IMPLEMENTED = 'not-implemented'
Attributes¤
FAILURE = 'failure' class-attribute instance-attribute ¤
NOT_IMPLEMENTED = 'not-implemented' class-attribute instance-attribute ¤
ROLLEDBACK = 'rolledback' class-attribute instance-attribute ¤
SIM_FAILURE = 'simulated-failure' class-attribute instance-attribute ¤
SIM_SUCCESS = 'simulated-success' class-attribute instance-attribute ¤
SUCCESS = 'success' class-attribute instance-attribute ¤

handler: python