OOP Design Patterns in Python
Difficulty: Medium-Hard | Companies: Google, Meta, Amazon, Netflix, Stripe
SOLID Principles Implementation
Single Responsibility Principle
class User:
"""User entity - only handles user data."""
def __init__(self, user_id: str, name: str, email: str):
self.user_id = user_id
self.name = name
self.email = email
self.created_at = datetime.now()
class UserRepository:
"""Handles user persistence - separate from User entity."""
def __init__(self, db_session):
self.db_session = db_session
def save(self, user: User):
self.db_session.add(user)
self.db_session.commit()
def find_by_id(self, user_id: str) -> User:
return self.db_session.query(User).filter_by(id=user_id).first()
class EmailService:
"""Handles email notifications - separate concern."""
def send_welcome_email(self, user: User):
# Email sending logic
pass
class UserService:
"""Coordinates user operations - uses other services."""
def __init__(self, repo: UserRepository, email_service: EmailService):
self.repo = repo
self.email_service = email_service
def create_user(self, user_data: dict) -> User:
user = User(**user_data)
self.repo.save(user)
self.email_service.send_welcome_email(user)
return user
βΉοΈ
Each class has a single responsibility, making the code easier to test, maintain, and extend.
Open/Closed Principle with Strategy Pattern
from abc import ABC, abstractmethod
from typing import List
class PaymentStrategy(ABC):
"""Abstract base for payment strategies."""
@abstractmethod
def process_payment(self, amount: float) -> bool:
pass
@abstractmethod
def validate(self) -> bool:
pass
class CreditCardPayment(PaymentStrategy):
def __init__(self, card_number: str, cvv: str):
self.card_number = card_number
self.cvv = cvv
def process_payment(self, amount: float) -> bool:
# Process credit card payment
return True
def validate(self) -> bool:
return len(self.card_number) == 16 and len(self.cvv) == 3
class PayPalPayment(PaymentStrategy):
def __init__(self, email: str, password: str):
self.email = email
self.password = password
def process_payment(self, amount: float) -> bool:
# Process PayPal payment
return True
def validate(self) -> bool:
return "@" in self.email
class CryptoPayment(PaymentStrategy):
def __init__(self, wallet_address: str):
self.wallet_address = wallet_address
def process_payment(self, amount: float) -> bool:
# Process cryptocurrency payment
return True
def validate(self) -> bool:
return len(self.wallet_address) >= 26
class PaymentProcessor:
"""Open for extension - can add new payment strategies without modification."""
def __init__(self, strategy: PaymentStrategy):
self.strategy = strategy
def process(self, amount: float) -> bool:
if not self.strategy.validate():
raise ValueError("Invalid payment method")
return self.strategy.process_payment(amount)
def set_strategy(self, strategy: PaymentStrategy):
self.strategy = strategy
Creational Patterns
Builder Pattern for Complex Objects
class HttpRequest:
def __init__(self):
self.method = "GET"
self.url = ""
self.headers = {}
self.body = None
self.timeout = 30
self.retry_count = 3
class HttpRequestBuilder:
"""Fluent builder for HTTP requests."""
def __init__(self):
self._request = HttpRequest()
def method(self, method: str) -> 'HttpRequestBuilder':
self._request.method = method.upper()
return self
def url(self, url: str) -> 'HttpRequestBuilder':
self._request.url = url
return self
def header(self, key: str, value: str) -> 'HttpRequestBuilder':
self._request.headers[key] = value
return self
def headers(self, headers: dict) -> 'HttpRequestBuilder':
self._request.headers.update(headers)
return self
def body(self, body: any) -> 'HttpRequestBuilder':
self._request.body = body
return self
def timeout(self, seconds: int) -> 'HttpRequestBuilder':
self._request.timeout = seconds
return self
def retry(self, count: int) -> 'HttpRequestBuilder':
self._request.retry_count = count
return self
def build(self) -> HttpRequest:
if not self._request.url:
raise ValueError("URL is required")
request = self._request
self._request = HttpRequest() # Reset for reuse
return request
# Usage with fluent interface
request = (HttpRequestBuilder()
.method("POST")
.url("https://api.example.com/users")
.header("Content-Type", "application/json")
.header("Authorization", "Bearer token123")
.body({"name": "John", "email": "john@example.com"})
.timeout(60)
.retry(5)
.build())
Factory Pattern with Registry
class Serializer(ABC):
@abstractmethod
def serialize(self, data: dict) -> str:
pass
@abstractmethod
def deserialize(self, data: str) -> dict:
pass
class JSONSerializer(Serializer):
def serialize(self, data: dict) -> str:
import json
return json.dumps(data)
def deserialize(self, data: str) -> dict:
import json
return json.loads(data)
class XMLSerializer(Serializer):
def serialize(self, data: dict) -> str:
# Simplified XML serialization
return f"<root>{data}</root>"
def deserialize(self, data: str) -> dict:
# Simplified XML deserialization
return {"data": data}
class SerializerFactory:
"""Factory with registry pattern."""
_serializers = {}
@classmethod
def register(cls, format_name: str, serializer_class):
cls._serializers[format_name] = serializer_class
@classmethod
def create(cls, format_name: str) -> Serializer:
if format_name not in cls._serializers:
raise ValueError(f"Unknown format: {format_name}")
return cls._serializers[format_name]()
# Registration
SerializerFactory.register("json", JSONSerializer)
SerializerFactory.register("xml", XMLSerializer)
# Usage
serializer = SerializerFactory.create("json")
data = serializer.serialize({"key": "value"})
Behavioral Patterns
Observer Pattern for Event System
from typing import Callable, Dict, List, Any
from collections import defaultdict
import weakref
class EventEmitter:
"""Event emitter with weak references to prevent memory leaks."""
def __init__(self):
self._listeners: Dict[str, List] = defaultdict(list)
def on(self, event: str, callback: Callable) -> Callable:
"""Register event listener. Returns unsubscribe function."""
listener = weakref.ref(callback)
self._listeners[event].append(listener)
def unsubscribe():
self._listeners[event] = [
l for l in self._listeners[event]
if l() is not None and l() != callback
]
return unsubscribe
def emit(self, event: str, *args, **kwargs) -> None:
"""Emit event to all listeners."""
dead_refs = []
for listener_ref in self._listeners[event]:
callback = listener_ref()
if callback is not None:
callback(*args, **kwargs)
else:
dead_refs.append(listener_ref)
# Cleanup dead references
for ref in dead_refs:
self._listeners[event].remove(ref)
def once(self, event: str, callback: Callable) -> Callable:
"""Register one-time listener."""
def wrapper(*args, **kwargs):
callback(*args, **kwargs)
unsubscribe()
unsubscribe = self.on(event, wrapper)
return unsubscribe
# Usage
emitter = EventEmitter()
def on_user_created(user):
print(f"User created: {user['name']}")
def send_welcome_email(user):
print(f"Sending email to {user['email']}")
emitter.on("user_created", on_user_created)
emitter.on("user_created", send_welcome_email)
emitter.emit("user_created", {"name": "John", "email": "john@example.com"})
Command Pattern with Undo/Redo
from abc import ABC, abstractmethod
from typing import List
class Command(ABC):
@abstractmethod
def execute(self) -> None:
pass
@abstractmethod
def undo(self) -> None:
pass
class TextEditor:
def __init__(self):
self.content = ""
def insert(self, text: str, position: int):
self.content = self.content[:position] + text + self.content[position:]
def delete(self, position: int, length: int) -> str:
deleted = self.content[position:position + length]
self.content = self.content[:position] + self.content[position + length:]
return deleted
class InsertCommand(Command):
def __init__(self, editor: TextEditor, text: str, position: int):
self.editor = editor
self.text = text
self.position = position
def execute(self):
self.editor.insert(self.text, self.position)
def undo(self):
self.editor.delete(self.position, len(self.text))
class DeleteCommand(Command):
def __init__(self, editor: TextEditor, position: int, length: int):
self.editor = editor
self.position = position
self.length = length
self.deleted_text = ""
def execute(self):
self.deleted_text = self.editor.delete(self.position, self.length)
def undo(self):
self.editor.insert(self.deleted_text, self.position)
class CommandHistory:
"""Manages undo/redo stacks."""
def __init__(self):
self._undo_stack: List[Command] = []
self._redo_stack: List[Command] = []
def execute(self, command: Command):
command.execute()
self._undo_stack.append(command)
self._redo_stack.clear()
def undo(self) -> bool:
if not self._undo_stack:
return False
command = self._undo_stack.pop()
command.undo()
self._redo_stack.append(command)
return True
def redo(self) -> bool:
if not self._redo_stack:
return False
command = self._redo_stack.pop()
command.execute()
self._undo_stack.append(command)
return True
β οΈ
In production systems, implement command serialization for persistence and audit trails.
Python-Specific OOP Patterns
Descriptor Protocol Implementation
class ValidatedField:
"""Descriptor that validates field values."""
def __init__(self, validator=None, error_message="Invalid value"):
self.validator = validator
self.error_message = error_message
def __set_name__(self, owner, name):
self.name = name
self.private_name = f"_{name}"
def __get__(self, obj, objtype=None):
if obj is None:
return self
return getattr(obj, self.private_name, None)
def __set__(self, obj, value):
if self.validator and not self.validator(value):
raise ValueError(f"{self.name}: {self.error_message}")
setattr(obj, self.private_name, value)
class User:
"""Example using validated descriptors."""
name = ValidatedField(
validator=lambda x: isinstance(x, str) and len(x) >= 2,
error_message="Name must be at least 2 characters"
)
email = ValidatedField(
validator=lambda x: "@" in x if x else False,
error_message="Invalid email format"
)
age = ValidatedField(
validator=lambda x: 0 <= x <= 150 if x is not None else False,
error_message="Age must be between 0 and 150"
)
def __init__(self, name: str, email: str, age: int):
self.name = name
self.email = email
self.age = age
# Usage
try:
user = User("John", "john@example.com", 30)
print(user.name) # John
user.name = "X" # Raises ValueError
except ValueError as e:
print(e) # Name: Name must be at least 2 characters
Context Manager as Class
import time
from contextlib import contextmanager
class Timer:
"""Class-based context manager for timing code blocks."""
def __init__(self, label: str = "Code block"):
self.label = label
self.start_time = None
self.end_time = None
self.elapsed = None
def __enter__(self):
self.start_time = time.perf_counter()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.end_time = time.perf_counter()
self.elapsed = self.end_time - self.start_time
if exc_type is not None:
print(f"{self.label} failed after {self.elapsed:.4f}s")
else:
print(f"{self.label} completed in {self.elapsed:.4f}s")
return False # Don't suppress exceptions
@contextmanager
def database_transaction(connection):
"""Function-based context manager for database transactions."""
transaction = connection.begin()
try:
yield connection
transaction.commit()
except Exception:
transaction.rollback()
raise
finally:
connection.close()
# Usage
with Timer("API call"):
time.sleep(0.1) # Simulate work
print("Task completed")
# Combined context managers
from contextlib import ExitStack
def process_files(filenames):
"""Process multiple files with automatic cleanup."""
with ExitStack() as stack:
files = [
stack.enter_context(open(fn))
for fn in filenames
]
# Process all files
for f in files:
print(f.read())
Follow-Up Questions
-
Explain the difference between composition and inheritance. When would you use each?
-
How does Python's Method Resolution Order (MRO) work with multiple inheritance?
-
What are the advantages of using abstract base classes over duck typing?
-
How would you implement dependency injection in a Python application?
-
Explain the difference between class methods, static methods, and instance methods.