Open closed principle implementation python

Question

I have case like based on a type send email to different platform (this is sample use case)

So I have done below so far

from abc import ABC, abstractclassmethod
import json

class Email(ABC):
    @abstractclassmethod
    def sendEmail(self):
        pass

class Gmail(ABC):
    def sendEmail(self):
        # some implementation
        return "Sent to gmail"
    
class HotMail(ABC):
    def sendEmail(self):
        # some implementation
        return "Sent to hotmail"


def lambda_handler(event, context):
    info =  json.loads(event['body'])

    mail_platform = info['type'] # this will be gmail or hotmail
    if mail_platform == 'gmail':
        mail = Gmail()
        mail.sendEmail()
    elif mail_platform == 'hotmail':
        mail = HotMail()
        mail.sendEmail()

Problem is whenever there is new email platform (yahoo for example) I have to add another elif in lambda code. Is there better way to handle this?

Answer 1

You can put supported classes inside a dictionary and treat them as any other variable, getting proper class based on a key from dictionary:

def lambda_handler(event, context):
    info =  json.loads(event['body'])
    types = {'gmail': Gmail, 'hotmail': HotMail}
    mail = types[info['type']]()
    mail.sendEmail()

Answer 2

If I were you, I would write a separate function parse that takes the subclass corresponding to the information from the type field. Something like this:

def get_mail_platform(type: str):
    return {
        'gmail': Gmail
        'hotmail': HotMail
    }[type]


def lambda_handler(event, context):
    info =  json.loads(event['body'])
    mail_platform = get_mail_platform(info['type'])()
    mail_platform.sendEmail()

And you should create a parent class so that all mail platform subclasses inherit this class. I hope it's useful.

Answer 3

Using dictionary or, even worse, a chain of elif statements is not a good idea, because you have to modify it each time a new service is introduced and it becomes less readable with each new statement.

If you truly want to adhere to the Open/closed principle, you must be able to introduce a new email service, without having to modify anything in your existing code.

You can achieve that by using the Factory design pattern.

One approach is to delegate the logic for each particular type of service to its corresponding class.

First, we define a check_type method in the abstract base class EmailAbstract, which each new concrete class must implement.

Then we create the factory - a concrete base class named Email. It has an additional _identify_service method. Based on the particular criteria this method instantiates the corresponding service class, which will be used. The method is closed - it doesn't have to be modified when we need to add a new email service class.

Finally, to introduce a new service, we only need to define a class, which implements the interface required by the abstract base class.

from abc import ABC, abstractclassmethod
import json


class EmailAbstract(ABC):
    @staticmethod
    def check_type(service_type):
        return False

    @abstractclassmethod
    def sendEmail(self):
        raise NotImplementedError


class Gmail(EmailAbstract):
    @staticmethod
    def check_type(service_type):
        return service_type == "gmail"

    def sendEmail(self):
        # some implementation
        return "Sent to gmail"


class HotMail(EmailAbstract):
    @staticmethod
    def check_type(service_type):
        return service_type == "hotmail"

    def sendEmail(self):
        # some implementation
        return "Sent to hotmail"


class Email(EmailAbstract):
    def __init__(self, service_type):
        self._service = self._identify_service(service_type)

    def _identify_service(self, service_type):
        for email_cls in EmailAbstract.__subclasses__():
            if email_cls.check_type(service_type):
                return email_cls()
        raise ValueError("Invalid service type")

    def sendEmail(self):
        return self._service.sendEmail()


def lambda_handler(event, context):
    info = json.loads(event["body"])
    mail_platform = Email(info["type"])
    ret = mail_platform.sendEmail()
    print(ret)

The second approach is to create a class registry inside the factory class, a dictionary where the mapping between service types and their corresponding classes is stored. The method register_service is used to register new service types by associating them with their respective classes. The get_service method looks up the product type in the registry and dynamically instantiates the corresponding class.

class EmailAbstract(ABC):
    @abstractclassmethod
    def sendEmail(self):
        raise NotImplementedError


class Gmail(EmailAbstract):
    def sendEmail(self):
        # some implementation
        return "Sent to gmail"


class HotMail(EmailAbstract):
    def sendEmail(self):
        # some implementation
        return "Sent to hotmail"


class Email:
    def __init__(self):
        self._service_registry = {}

    def register_service(self, service_type, service_cls):
        self._service_registry[service_type] = service_cls

    def get_service(self, service_type):
        try:
            return self._service_registry[service_type]()
        except KeyError:
            raise ValueError("Invalid service type")


def lambda_handler(event, context):
    info = json.loads(event["body"])
    factory = Email()
    factory.register_service("gmail", Gmail)
    factory.register_service("hotmail", HotMail)
    mail_platform = factory.get_service(info["type"])
    ret = mail_platform.sendEmail()
    print(ret)

Both implementations could be tested using the same code:

if __name__ == "__main__":
    event = {"body": '{"type": "gmail"}'}
    lambda_handler(event, None)
    # Output: Sent to gmail

    event = {"body": '{"type": "hotmail"}'}
    lambda_handler(event, None)
    # Output: Sent to hotmail

    event = {"body": '{"type": "asd"}'}
    lambda_handler(event, None)
    # Output: ValueError: Invalid service type