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.
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": "admin@company.com"}),
("slack", {"recipient": "#alerts"}),
("sms", {"recipient": "+1234567890"})
]
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": "user@example.com", "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
Conclusion
The Singleton, Factory, and Abstract Factory design patterns are all creational patterns that manage object creation in distinct ways. Singleton ensures a class has only one instance and provides global access to it, ideal for shared resources like configuration or logging. Factory abstracts the instantiation process by delegating it to subclasses or methods, promoting loose coupling and flexibility. Abstract Factory goes a step further by creating families of related objects without specifying their concrete classes, making it perfect for systems that need to support multiple themes or platforms. Together, these patterns enhance modularity, scalability, and maintainability in software design.
As always feel free to share your thoughts and criticize this article!
Top comments (0)