Implementing Protocol Classes for Flexible Interfaces
Protocol classes, introduced in Python 3.8, provide a way to define a set of required methods without requiring explicit inheritance. This allows for more flexible and decoupled code, enabling duck typing with more clarity and enabling static type checkers to verify that objects conform to a specific interface. Your task is to implement protocol classes to define and enforce a communication protocol between different components.
Problem Description
You are tasked with designing a system for handling different types of message processors. Each processor type (e.g., EmailProcessor, SMSProcessor, LogProcessor) needs to adhere to a common protocol: they must all have a process_message method that accepts a message string and returns a status code. Instead of using traditional inheritance (which can lead to rigid hierarchies), you will use protocol classes to define this interface.
Specifically, you need to:
- Define a
MessageProcessorProtocolprotocol class with aprocess_messagemethod. This method should take a stringmessageas input and return an integer representing the processing status. - Implement three concrete processor classes:
EmailProcessor,SMSProcessor, andLogProcessor. Each class should implement theMessageProcessorProtocolby providing aprocess_messagemethod. The implementations should return different status codes based on the message content (see examples below for guidance). - Create a
MessageDispatcherclass that takes a list ofMessageProcessorProtocolobjects. This dispatcher should iterate through the processors and call theirprocess_messagemethod for a given message. The dispatcher should return a list of status codes, one for each processor. - Ensure that the code is type-hinted correctly to leverage the benefits of protocol classes with static type checkers.
Expected Behavior:
The MessageDispatcher should correctly call the process_message method of each registered processor and return a list of the returned status codes in the order the processors were registered. The code should be robust and handle cases where a processor might not implement the protocol correctly (though this is less of a concern with type checkers).
Examples
Example 1:
Input: message = "Important update", processors = [EmailProcessor(), SMSProcessor(), LogProcessor()]
Output: [200, 201, 202]
Explanation: EmailProcessor returns 200, SMSProcessor returns 201, and LogProcessor returns 202 for the given message.
Example 2:
Input: message = "Error report", processors = [SMSProcessor(), LogProcessor()]
Output: [201, 202]
Explanation: SMSProcessor returns 201, and LogProcessor returns 202 for the given message.
Example 3: (Edge Case - Empty Processor List)
Input: message = "Test message", processors = []
Output: []
Explanation: If the processor list is empty, the dispatcher should return an empty list.
Constraints
- The
process_messagemethod in each processor must accept a string and return an integer. - The
MessageDispatchermust accept a list of objects that conform to theMessageProcessorProtocol. - The status codes returned by the processors should be distinct for different messages (though this is not strictly enforced).
- The code should be well-documented and follow Python best practices.
Notes
- Consider using the
typing.Protocoltype hint to define the protocol class. - The
MessageDispatchershould not assume any specific implementation details of the processors, only that they adhere to the protocol. - Think about how you can use type hints to improve the readability and maintainability of your code.
- The processor implementations can be simple and focus on demonstrating the protocol usage. The specific logic within
process_messageis not critical to the problem.
from typing import List, Protocol
class MessageProcessorProtocol(Protocol):
def process_message(self, message: str) -> int:
...
class EmailProcessor(MessageProcessorProtocol):
def process_message(self, message: str) -> int:
if "urgent" in message.lower():
return 200
else:
return 203
class SMSProcessor(MessageProcessorProtocol):
def process_message(self, message: str) -> int:
if "error" in message.lower():
return 201
else:
return 204
class LogProcessor(MessageProcessorProtocol):
def process_message(self, message: str) -> int:
return 202
class MessageDispatcher:
def __init__(self, processors: List[MessageProcessorProtocol]):
self.processors = processors
def dispatch(self, message: str) -> List[int]:
status_codes = []
for processor in self.processors:
status_codes.append(processor.process_message(message))
return status_codes
if __name__ == '__main__':
processors = [EmailProcessor(), SMSProcessor(), LogProcessor()]
dispatcher = MessageDispatcher(processors)
message = "Important update"
status_codes = dispatcher.dispatch(message)
print(f"Status codes for message '{message}': {status_codes}")
message = "Error report"
status_codes = dispatcher.dispatch(message)
print(f"Status codes for message '{message}': {status_codes}")
processors = []
dispatcher = MessageDispatcher(processors)
message = "Test message"
status_codes = dispatcher.dispatch(message)
print(f"Status codes for message '{message}': {status_codes}")