Creational Design Patterns in Python. Part I

Design patterns are proven solutions to recurring problems in software design. Among the three main categories of design patterns, Creational patterns focus on object creation mechanisms, providing flexible ways to create objects while hiding the creation logic and making the system independent of how objects are created, composed, and represented.
This article explores the most important creational design patterns in Python, complete with practical examples and real-world use cases that you'll encounter in production systems.

A lot of my inspiration for this article comes from a particularly good book that I strongly recommend to buy, but if you don't want to, author share basically everything on the website free to view and also code examples for every major programming language! Shoutout to Refactoring.Guru

What Are Creational Design Patterns?

Creational design patterns abstract the instantiation process, making systems more flexible and reusable. They help manage object creation complexity and ensure that objects are created in a manner suitable to the situation. These patterns become especially valuable when dealing with complex object hierarchies or when the exact types of objects to be created are determined at runtime.

Factory Pattern

The Factory pattern creates objects without specifying their exact classes, delegating the creation logic to subclasses.

from abc import ABC, abstractmethod
from typing import Dict, Any
import smtplib
import requests
from email.mime.text import MIMEText

class NotificationSender(ABC):
    @abstractmethod
    def send(self, recipient: str, subject: str, message: str) -> bool:
        pass

class EmailSender(NotificationSender):
    def __init__(self, smtp_server: str, username: str, password: str):
        self.smtp_server = smtp_server
        self.username = username
        self.password = password

    def send(self, recipient: str, subject: str, message: str) -> bool:
        try:
            msg = MIMEText(message)
            msg['Subject'] = subject
            msg['From'] = self.username
            msg['To'] = recipient

            with smtplib.SMTP(self.smtp_server, 587) as server:
                server.starttls()
                server.login(self.username, self.password)
                server.send_message(msg)
            return True
        except Exception as e:
            print(f"Email failed: {e}")
            return False

class SlackSender(NotificationSender):
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url

    def send(self, recipient: str, subject: str, message: str) -> bool:
        try:
            payload = {
                "channel": recipient,
                "text": f"*{subject}*\n{message}"
            }
            response = requests.post(self.webhook_url, json=payload)
            return response.status_code == 200
        except Exception as e:
            print(f"Slack failed: {e}")
            return False

class SMSSender(NotificationSender):
    def __init__(self, api_key: str, service_url: str):
        self.api_key = api_key
        self.service_url = service_url

    def send(self, recipient: str, subject: str, message: str) -> bool:
        try:
            payload = {
                "to": recipient,
                "message": f"{subject}: {message}",
                "api_key": self.api_key
            }
            response = requests.post(self.service_url, json=payload)
            return response.status_code == 200
        except Exception as e:
            print(f"SMS failed: {e}")
            return False

class NotificationFactory:
    def create_sender(self, notification_type: str, config: Dict[str, Any]) -> NotificationSender:
        if notification_type == "email":
            return EmailSender(
                config['smtp_server'],
                config['username'], 
                config['password']
            )
        elif notification_type == "slack":
            return SlackSender(config['webhook_url'])
        elif notification_type == "sms":
            return SMSSender(config['api_key'], config['service_url'])
        else:
            raise ValueError(f"Unknown notification type: {notification_type}")

# Usage in a real application
class AlertSystem:
    def __init__(self):
        self.factory = NotificationFactory()
        self.config = ConfigManager()  # From singleton example

    def send_critical_alert(self, message: str):
        # Send via multiple channels for critical alerts
        channels = [
            ("email", {"recipient": "[e-mail]"}),
            ("slack", {"recipient": "#alerts"}),
            ("sms", {"recipient": "[phone]"})
        ]

        for channel_type, channel_config in channels:
            try:
                sender_config = self.config.get(f'{channel_type}_config')
                sender = self.factory.create_sender(channel_type, sender_config)
                sender.send(
                    channel_config['recipient'],
                    "CRITICAL ALERT",
                    message
                )
            except Exception as e:
                print(f"Failed to send {channel_type} alert: {e}")

Real-World Applications:

• E-commerce Platforms: Payment processors, shipping providers, tax calculators
• Content Management: File parsers (PDF, Word, Excel), image processors
• Cloud Services: Storage providers (AWS S3, Google Cloud, Azure), - CDN providers
• Gaming: Character classes, weapon types, enemy spawners
• IoT Systems: Device drivers, protocol handlers, sensor data processors

Abstract Factory Pattern

The Abstract Factory pattern provides an interface for creating families of related objects without specifying their concrete classes.

from abc import ABC, abstractmethod
from typing import Protocol

# Abstract Products
class DatabaseConnection(Protocol):
    def connect(self) -> bool: ...
    def execute_query(self, query: str) -> Any: ...
    def close(self) -> None: ...

class CacheClient(Protocol):
    def get(self, key: str) -> Any: ...
    def set(self, key: str, value: Any, ttl: int = 3600) -> bool: ...
    def delete(self, key: str) -> bool: ...

class MessageQueue(Protocol):
    def publish(self, topic: str, message: str) -> bool: ...
    def subscribe(self, topic: str) -> Any: ...

# AWS Implementation
class AWSRDSConnection:
    def __init__(self, config: Dict):
        self.config = config
        self.connection = None

    def connect(self) -> bool:
        # AWS RDS connection logic
        print(f"Connecting to AWS RDS: {self.config['endpoint']}")
        return True

    def execute_query(self, query: str) -> Any:
        print(f"Executing on RDS: {query}")
        return {"rows": [], "status": "success"}

    def close(self) -> None:
        print("Closing RDS connection")

class AWSElastiCacheClient:
    def __init__(self, config: Dict):
        self.config = config

    def get(self, key: str) -> Any:
        print(f"Getting from ElastiCache: {key}")
        return None

    def set(self, key: str, value: Any, ttl: int = 3600) -> bool:
        print(f"Setting in ElastiCache: {key} = {value}")
        return True

    def delete(self, key: str) -> bool:
        print(f"Deleting from ElastiCache: {key}")
        return True

class AWSSQSQueue:
    def __init__(self, config: Dict):
        self.config = config

    def publish(self, topic: str, message: str) -> bool:
        print(f"Publishing to SQS {topic}: {message}")
        return True

    def subscribe(self, topic: str) -> Any:
        print(f"Subscribing to SQS {topic}")
        return []

# Google Cloud Implementation
class GCPCloudSQLConnection:
    def __init__(self, config: Dict):
        self.config = config

    def connect(self) -> bool:
        print(f"Connecting to Cloud SQL: {self.config['instance']}")
        return True

    def execute_query(self, query: str) -> Any:
        print(f"Executing on Cloud SQL: {query}")
        return {"rows": [], "status": "success"}

    def close(self) -> None:
        print("Closing Cloud SQL connection")

class GCPMemorystoreClient:
    def __init__(self, config: Dict):
        self.config = config

    def get(self, key: str) -> Any:
        print(f"Getting from Memorystore: {key}")
        return None

    def set(self, key: str, value: Any, ttl: int = 3600) -> bool:
        print(f"Setting in Memorystore: {key} = {value}")
        return True

    def delete(self, key: str) -> bool:
        print(f"Deleting from Memorystore: {key}")
        return True

class GCPPubSubQueue:
    def __init__(self, config: Dict):
        self.config = config

    def publish(self, topic: str, message: str) -> bool:
        print(f"Publishing to Pub/Sub {topic}: {message}")
        return True

    def subscribe(self, topic: str) -> Any:
        print(f"Subscribing to Pub/Sub {topic}")
        return []

# Abstract Factory
class CloudInfrastructureFactory(ABC):
    @abstractmethod
    def create_database(self, config: Dict) -> DatabaseConnection:
        pass

    @abstractmethod
    def create_cache(self, config: Dict) -> CacheClient:
        pass

    @abstractmethod
    def create_message_queue(self, config: Dict) -> MessageQueue:
        pass

# Concrete Factories
class AWSFactory(CloudInfrastructureFactory):
    def create_database(self, config: Dict) -> DatabaseConnection:
        return AWSRDSConnection(config)

    def create_cache(self, config: Dict) -> CacheClient:
        return AWSElastiCacheClient(config)

    def create_message_queue(self, config: Dict) -> MessageQueue:
        return AWSSQSQueue(config)

class GCPFactory(CloudInfrastructureFactory):
    def create_database(self, config: Dict) -> DatabaseConnection:
        return GCPCloudSQLConnection(config)

    def create_cache(self, config: Dict) -> CacheClient:
        return GCPMemorystoreClient(config)

    def create_message_queue(self, config: Dict) -> MessageQueue:
        return GCPPubSubQueue(config)

# Application Service
class UserService:
    def __init__(self, factory: CloudInfrastructureFactory, config: Dict):
        self.db = factory.create_database(config.get('database', {}))
        self.cache = factory.create_cache(config.get('cache', {}))
        self.queue = factory.create_message_queue(config.get('queue', {}))

    def get_user(self, user_id: str) -> Dict:
        # Check cache first
        cached_user = self.cache.get(f"user:{user_id}")
        if cached_user:
            return cached_user

        # Query database
        self.db.connect()
        user_data = self.db.execute_query(f"SELECT * FROM users WHERE id = '{user_id}'")
        self.db.close()

        # Cache the result
        self.cache.set(f"user:{user_id}", user_data)

        return user_data

    def create_user(self, user_data: Dict) -> bool:
        # Save to database
        self.db.connect()
        result = self.db.execute_query(f"INSERT INTO users VALUES (...)")
        self.db.close()

        # Publish event
        self.queue.publish("user.created", f"New user created: {user_data['email']}")

        return True

# Usage - Easy cloud provider switching
def create_user_service(cloud_provider: str) -> UserService:
    config = ConfigManager()

    factories = {
        'aws': AWSFactory(),
        'gcp': GCPFactory(),
        'azure': AzureFactory()  # Could be added later
    }

    factory = factories.get(cloud_provider)
    if not factory:
        raise ValueError(f"Unsupported cloud provider: {cloud_provider}")

    return UserService(factory, config.get(f'{cloud_provider}_config'))

# Switch providers easily
user_service = create_user_service('aws')  # or 'gcp'
user_service.create_user({"email": "[e-mail]", "name": "John Doe"})

Real-World Applications:

• Multi-Cloud Applications: Switch between AWS, GCP, Azure services seamlessly
• Database Migration: Support multiple database backends (PostgreSQL, MySQL, MongoDB)
• Cross-Platform Development: Different UI components for web, mobile, desktop
• Testing Environments: Mock vs real implementations for development/production
• Internationalization: Region-specific payment methods, shipping providers, tax systems
• Multi-Tenant SaaS: Different feature sets and integrations per tenant tier

Singleton Pattern

The Singleton pattern ensures that a class has only one instance and provides global access to that instance.

import threading
import logging
from typing import Optional

class Logger:
    _instance: Optional['Logger'] = None
    _lock = threading.Lock()

    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
                    cls._instance._initialized = False
        return cls._instance

    def __init__(self):
        if not self._initialized:
            self.logger = logging.getLogger('application')
            self.logger.setLevel(logging.INFO)

            # Create file handler
            handler = logging.FileHandler('app.log')
            formatter = logging.Formatter(
                '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
            )
            handler.setFormatter(formatter)
            self.logger.addHandler(handler)

            self._initialized = True

    def info(self, message: str):
        self.logger.info(message)

    def error(self, message: str):
        self.logger.error(message)

    def warning(self, message: str):
        self.logger.warning(message)

# Usage across different modules
def process_order(order_id: str):
    logger = Logger()
    logger.info(f"Processing order: {order_id}")
    # ... processing logic
    logger.info(f"Order {order_id} completed")

def handle_payment(payment_id: str):
    logger = Logger()  # Same instance as above
    logger.info(f"Processing payment: {payment_id}")
    # ... payment logic

Real-World Applications:

• Microservices: Service discovery clients, configuration managers
• Web Applications: Database connection pools, cache managers, session stores
• Desktop Applications: Settings managers, plugin registries
• Mobile Apps: Analytics managers, crash reporters
• DevOps Tools: Monitoring agents, deployment managers

When to Use Each Pattern

• Singleton: When you need exactly one instance of a class throughout your application (database connections, logging, configuration).
• Factory Method: When you need to create objects but don't know the exact class until runtime, or when you want to delegate object creation to subclasses.
• Abstract Factory: When you need to create families of related objects that must be used together (GUI components for different operating systems).

Continue reading with Part II!

As always feel free to share your thoughts and criticize this article!