File: workflow_by_chain_of_responsability_oo.py
from abc import ABC, abstractmethod
#
# Abstract classes
#
class SharedAttributes:
""" Empty object to store shared attributes among handlers """
def __str__(self):
return str(self.__dict__)
class Handler(ABC):
@abstractmethod
def handle_request(self, request, shared_attributes):
pass
class Chain:
handlers = tuple()
def handle_request(self, request):
# instantiate new shared attributes object
self.shared_attributes = SharedAttributes()
# instantiate handler class and execute method .handle_request()
for handler_class in self.handlers:
handler = handler_class()
# execute handler request
handler.handle_request(request=request, shared_attributes=self.shared_attributes)
# return shared_attributes
return self.shared_attributes
#
# Concrete classes
#
class FirstWorkflowStep(Handler):
def handle_request(self, request, shared_attributes):
shared_attributes.step = 'first'
shared_attributes.x = 1
print(f"-> Handling [request: {request}] [by: {self.__class__.__name__}] - shared attributes:", request, shared_attributes)
class SecondWorkflowStep(Handler):
def handle_request(self, request, shared_attributes):
shared_attributes.step = 'second'
shared_attributes.y = 2
print(f"-> Handling [request: {request}] [by: {self.__class__.__name__}] - shared attributes:", request, shared_attributes)
class ThirdWorkflowStep(Handler):
def handle_request(self, request, shared_attributes):
shared_attributes.step = 'third'
shared_attributes.z = 3
print(f"-> Handling [request: {request}] [by: {self.__class__.__name__}] - shared attributes:", request, shared_attributes)
class MySampleWorkflow(Chain):
handlers = [
FirstWorkflowStep,
SecondWorkflowStep,
ThirdWorkflowStep
]
if __name__ == "__main__" :
workflow = MySampleWorkflow()
requests = [ '1' , '2' , '3' ]
for request in requests:
print(f"Executing workflow for request: {request}")
workflow.handle_request(request)
print()
Executing:
$ python workflow_by_chain_of_responsability_oo.py
Executing workflow for request: 1
-> Handling [request: 1] [by: FirstWorkflowStep] - shared attributes: 1 {'step': 'first', 'x': 1}
-> Handling [request: 1] [by: SecondWorkflowStep] - shared attributes: 1 {'step': 'second', 'x': 1, 'y': 2}
-> Handling [request: 1] [by: ThirdWorkflowStep] - shared attributes: 1 {'step': 'third', 'x': 1, 'y': 2, 'z': 3}
Executing workflow for request: 2
-> Handling [request: 2] [by: FirstWorkflowStep] - shared attributes: 2 {'step': 'first', 'x': 1}
-> Handling [request: 2] [by: SecondWorkflowStep] - shared attributes: 2 {'step': 'second', 'x': 1, 'y': 2}
-> Handling [request: 2] [by: ThirdWorkflowStep] - shared attributes: 2 {'step': 'third', 'x': 1, 'y': 2, 'z': 3}
Executing workflow for request: 3
-> Handling [request: 3] [by: FirstWorkflowStep] - shared attributes: 3 {'step': 'first', 'x': 1}
-> Handling [request: 3] [by: SecondWorkflowStep] - shared attributes: 3 {'step': 'second', 'x': 1, 'y': 2}
-> Handling [request: 3] [by: ThirdWorkflowStep] - shared attributes: 3 {'step': 'third', 'x': 1, 'y': 2, 'z': 3}
File: workflow_by_chain_of_responsability_functional.py
from abc import ABC, abstractmethod
#
# Abstract classes
#
class SharedAttributes:
""" Empty object to store shared attributes among handlers """
def __str__(self):
return str(self.__dict__)
class Chain:
handlers = tuple()
def __init__(self, handlers: tuple | None = None) -> None:
self.handlers = handlers or self.handlers
def handle_request(self, request):
# instantiate new shared attributes object
self.shared_attributes = SharedAttributes()
# instantiate handler class and execute method .handle_request()
for handler_function in self.handlers:
# execute handler request
handler_function(request=request, shared_attributes=self.shared_attributes)
# return shared_attributes
return self.shared_attributes
def first_workflow_step(request, shared_attributes):
shared_attributes.step = 'first'
shared_attributes.x = 1
print(f"-> Handling [request: {request}] [by: first_workflow_step] - shared attributes:", request, shared_attributes)
def second_workflow_step(request, shared_attributes):
shared_attributes.step = 'second'
shared_attributes.y = 2
print(f"-> Handling [request: {request}] [by: second_workflow_step] - shared attributes:", request, shared_attributes)
def third_workflow_step(request, shared_attributes):
shared_attributes.step = 'third'
shared_attributes.z = 3
print(f"-> Handling [request: {request}] [by: third_workflow_step] - shared attributes:", request, shared_attributes)
class MySampleWorkflow(Chain):
handlers = [
first_workflow_step,
second_workflow_step,
third_workflow_step
]
if __name__ == "__main__" :
# Define workflow
workflow = MySampleWorkflow()
## Alternative way to declare the workflow directly without class
# my_workflow_pipeline = [
# first_workflow_step,
# second_workflow_step,
# third_workflow_step
# ]
# workflow = Chain(handlers=my_workflow_pipeline)
# Define requests
requests = [ '1' , '2' , '3' ]
# Execute workflow process for each request
for request in requests:
print(f"Executing workflow for request: {request}")
workflow.handle_request(request)
print()
Executing:
$ python workflow_by_chain_of_responsability_functional.py
Executing workflow for request: 1
-> Handling [request: 1] [by: first_workflow_step] - shared attributes: 1 {'step': 'first', 'x': 1}
-> Handling [request: 1] [by: second_workflow_step] - shared attributes: 1 {'step': 'second', 'x': 1, 'y': 2}
-> Handling [request: 1] [by: third_workflow_step] - shared attributes: 1 {'step': 'third', 'x': 1, 'y': 2, 'z': 3}
Executing workflow for request: 2
-> Handling [request: 2] [by: first_workflow_step] - shared attributes: 2 {'step': 'first', 'x': 1}
-> Handling [request: 2] [by: second_workflow_step] - shared attributes: 2 {'step': 'second', 'x': 1, 'y': 2}
-> Handling [request: 2] [by: third_workflow_step] - shared attributes: 2 {'step': 'third', 'x': 1, 'y': 2, 'z': 3}
Executing workflow for request: 3
-> Handling [request: 3] [by: first_workflow_step] - shared attributes: 3 {'step': 'first', 'x': 1}
-> Handling [request: 3] [by: second_workflow_step] - shared attributes: 3 {'step': 'second', 'x': 1, 'y': 2}
-> Handling [request: 3] [by: third_workflow_step] - shared attributes: 3 {'step': 'third', 'x': 1, 'y': 2, 'z': 3}