Example of Workflows implementation like Chain Responsability but in a single Class

workflow_single_class.py

from typing import Any


class WorkflowSkip(Exception):
    ...


class WorkflowProcess:
    workflow_steps = ()

    def __init__(self, execution_id: Any = None, *args, **kwargs) -> None:
        self.execution_id = execution_id
        self.step_method = None

    #
    # Workflow engine
    #
    def run_workflow_steps(self):
        try:
            # execute workflow steps
            for step_method in self.workflow_steps:
                self.step_method = step_method
                getattr(self, step_method)()

        except WorkflowSkip as exc:
            # quiet skip invoked arbitrarily
            self.inform_status('SKIPPED', str(exc))
            self.run_workflow_handler(handler_name='skip_handler', exc=exc)

        except Exception as exc:
            # any exception invoked
            self.inform_status('FAILED', str(exc))
            self.run_workflow_handler(handler_name='error_handler', exc=exc)

        else:
            # all workflow steps executed okay
            self.run_workflow_handler(handler_name='success_handler')

        finally:
            # execute final handler generally to clean up objects
            self.run_workflow_handler(handler_name='final_handler')

    #
    # Trigger handlers
    #
    def skip_handler(self, exc):
        ...

    def error_handler(self, exc):
        ...

    def success_handler(self):
        ...

    def final_handler(self):
        ...

    def run_workflow_handler(self, handler_name: str, *args, **kwargs) -> None:
        try:
            handler_method = getattr(self, handler_name)
            handler_method(*args, **kwargs)
        except Exception as exc:
            self.step_method = handler_name
            self.inform_status('FAILED', str(exc))

    #
    # Report status
    #
    def inform_status(self, status: str, message: str = '') -> None:
        # define message line
        line = f'[{status}] [{self.__class__.__name__}]'
        if self.step_method:
            line = f'{line} [{self.step_method}]'
        if self.execution_id:
            line = f'{line} - ({self.execution_id})'
        if message:
            line = f'{line} - {message}'
        print(line)


class MyCustomWorkflow(WorkflowProcess):
    workflow_steps = (
        'step_my_first_task',
        'step_my_second_task',
        'step_my_third_task'
    )
    
    def __init__(self, external_attr: str):
        # load inheterance
        super().__init__(external_id=1)

        # external attributes
        self.external_attr = external_attr
      
        # shared attributes between step methods
        self.first_shared_attr = None
        self.second_shared_attr = None
        self.third_shared_attr = None
    
    def __example_of_custom_method(self):
        print('################################################################')
        print(f'       Step method: MyCustomWorkflow.{self.step_method}')
        print('################################################################')
        print(f'-> external_attr: {self.external_attr}')
        print(f'-> first_shared_attr: {self.first_shared_attr}')
        print(f'-> second_shared_attr: {self.second_shared_attr}')
        print(f'-> third_shared_attr: {self.third_shared_attr}\n')
    
    def step_my_first_task(self):
        self.first_shared_attr = 'My first value shared between step methods'
        self.__example_of_custom_method()
        
    def step_my_second_task(self):
        self.second_shared_attr = 'The step_my_second_task passed here'
        self.__example_of_custom_method()

    def step_my_third_task(self):
        self.third_shared_attr = 'Preparing to finish'
        self.__example_of_custom_method()


if __name__ == '__main__':
    workflow = MyCustomWorkflow(external_attr='this is an external value')
    workflow.run_workflow_steps()
    
"""
python workflow_single_class.py
################################################################
       Step method: MyCustomWorkflow.step_my_first_task
################################################################
-> external_attr: this is an external value
-> first_shared_attr: My first value shared between step methods
-> second_shared_attr: None
-> third_shared_attr: None

################################################################
       Step method: MyCustomWorkflow.step_my_second_task
################################################################
-> external_attr: this is an external value
-> first_shared_attr: My first value shared between step methods
-> second_shared_attr: The step_my_second_task passed here
-> third_shared_attr: None

################################################################
       Step method: MyCustomWorkflow.step_my_third_task
################################################################
-> external_attr: this is an external value
-> first_shared_attr: My first value shared between step methods
-> second_shared_attr: The step_my_second_task passed here
-> third_shared_attr: Preparing to finish
"""