Creational Design Patterns in Python. Part I

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}") 

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"}) 

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