Python Patterns
Essential design patterns and architectural patterns for writing maintainable, scalable Python code.
Python Patterns Cheatsheet
Essential design patterns and architectural patterns for writing maintainable, scalable Python code.
Overview
Design patterns are reusable solutions to common software design problems. Python's dynamic nature and first-class functions enable elegant implementations of both traditional and Pythonic patterns.
graph TB
subgraph Creational["Creational Patterns"]
Singleton["Singleton"]
Factory["Factory"]
end
subgraph Structural["Structural Patterns"]
Decorator["Decorator"]
Repository["Repository"]
end
subgraph Behavioural["Behavioural Patterns"]
Observer["Observer"]
Strategy["Strategy"]
end
subgraph Pythonic["Pythonic Patterns"]
ContextManager["Context Managers"]
ErrorHandling["Error Handling"]
DI["Dependency Injection"]
end
Creational --> Application["Python Application"]
Structural --> Application
Behavioural --> Application
Pythonic --> Application
Singleton Pattern
Key Concepts
The Singleton pattern ensures a class has only one instance and provides a global point of access to it. Common use cases include configuration managers, connection pools, and logging systems.
classDiagram
class Singleton {
-_instance: Singleton
-__init__()
+get_instance() Singleton
+some_operation()
}
Client --> Singleton : requests instance
Common Patterns
| Approach | Description | Thread-Safe |
|---|---|---|
| Module-level | Use module as singleton | Yes |
__new__ override |
Control instance creation | No |
| Metaclass | Custom metaclass approach | Configurable |
| Decorator | Function decorator wrapper | Configurable |
Examples
Module-level Singleton (Pythonic):
# config.py - The module itself is the singleton
class _Config:
def __init__(self):
self.settings = {}
def load(self, path):
# Load configuration from file
pass
# Single instance created on import
config = _Config()
# Usage in other modules
from config import config
config.settings['debug'] = True
Using __new__ Method:
class Singleton:
_instance = None
def __new__(cls, *args, **kwargs):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, value=None):
# Only initialise once
if not hasattr(self, 'initialised'):
self.value = value
self.initialised = True
# Usage
s1 = Singleton('first')
s2 = Singleton('second')
print(s1 is s2) # True
print(s1.value) # 'first'
Thread-Safe Singleton with Metaclass:
import threading
class SingletonMeta(type):
_instances = {}
_lock = threading.Lock()
def __call__(cls, *args, **kwargs):
with cls._lock:
if cls not in cls._instances:
instance = super().__call__(*args, **kwargs)
cls._instances[cls] = instance
return cls._instances[cls]
class Database(metaclass=SingletonMeta):
def __init__(self):
self.connection = None
def connect(self, connection_string):
self.connection = connection_string
Decorator-based Singleton:
def singleton(cls):
instances = {}
def get_instance(*args, **kwargs):
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return get_instance
@singleton
class Logger:
def __init__(self):
self.logs = []
def log(self, message):
self.logs.append(message)
Factory Pattern
Key Concepts
The Factory pattern provides an interface for creating objects without specifying their exact classes. It promotes loose coupling and makes code more maintainable and extensible.
classDiagram
class Creator {
<<abstract>>
+factory_method() Product
+some_operation()
}
class ConcreteCreatorA {
+factory_method() ProductA
}
class ConcreteCreatorB {
+factory_method() ProductB
}
class Product {
<<interface>>
+operation()
}
class ProductA {
+operation()
}
class ProductB {
+operation()
}
Creator <|-- ConcreteCreatorA
Creator <|-- ConcreteCreatorB
Product <|.. ProductA
Product <|.. ProductB
ConcreteCreatorA ..> ProductA
ConcreteCreatorB ..> ProductB
Common Patterns
| Type | Use Case | Implementation |
|---|---|---|
| Simple Factory | Single creation point | Function or class method |
| Factory Method | Subclass determines type | Abstract method in base class |
| Abstract Factory | Family of related objects | Multiple factory methods |
Examples
Simple Factory Function:
from abc import ABC, abstractmethod
class Serialiser(ABC):
@abstractmethod
def serialise(self, data):
pass
class JSONSerialiser(Serialiser):
def serialise(self, data):
import json
return json.dumps(data)
class XMLSerialiser(Serialiser):
def serialise(self, data):
# Convert to XML format
return f"<data>{data}</data>"
class YAMLSerialiser(Serialiser):
def serialise(self, data):
import yaml
return yaml.dump(data)
def get_serialiser(format_type: str) -> Serialiser:
"""Factory function to create serialisers."""
serialisers = {
'json': JSONSerialiser,
'xml': XMLSerialiser,
'yaml': YAMLSerialiser,
}
serialiser_class = serialisers.get(format_type.lower())
if not serialiser_class:
raise ValueError(f"Unknown format: {format_type}")
return serialiser_class()
# Usage
serialiser = get_serialiser('json')
result = serialiser.serialise({'name': 'value'})
Factory Method Pattern:
from abc import ABC, abstractmethod
class Document(ABC):
@abstractmethod
def render(self):
pass
class PDFDocument(Document):
def render(self):
return "Rendering PDF document"
class HTMLDocument(Document):
def render(self):
return "Rendering HTML document"
class DocumentCreator(ABC):
@abstractmethod
def create_document(self) -> Document:
"""Factory method to be implemented by subclasses."""
pass
def open_document(self):
document = self.create_document()
return document.render()
class PDFCreator(DocumentCreator):
def create_document(self) -> Document:
return PDFDocument()
class HTMLCreator(DocumentCreator):
def create_document(self) -> Document:
return HTMLDocument()
# Usage
creator = PDFCreator()
print(creator.open_document())
Registry-based Factory:
class HandlerFactory:
_handlers = {}
@classmethod
def register(cls, handler_type):
"""Decorator to register handlers."""
def decorator(handler_class):
cls._handlers[handler_type] = handler_class
return handler_class
return decorator
@classmethod
def create(cls, handler_type, *args, **kwargs):
handler_class = cls._handlers.get(handler_type)
if not handler_class:
raise ValueError(f"Unknown handler: {handler_type}")
return handler_class(*args, **kwargs)
@HandlerFactory.register('email')
class EmailHandler:
def __init__(self, address):
self.address = address
def send(self, message):
print(f"Sending email to {self.address}: {message}")
@HandlerFactory.register('sms')
class SMSHandler:
def __init__(self, phone):
self.phone = phone
def send(self, message):
print(f"Sending SMS to {self.phone}: {message}")
# Usage
handler = HandlerFactory.create('email', 'user@example.com')
handler.send('Hello!')
Observer Pattern
Key Concepts
The Observer pattern defines a one-to-many dependency between objects so that when one object changes state, all its dependents are notified and updated automatically. It's fundamental for event-driven programming.
sequenceDiagram
participant Subject
participant Observer1
participant Observer2
Observer1->>Subject: subscribe()
Observer2->>Subject: subscribe()
Note over Subject: State changes
Subject->>Observer1: notify(data)
Subject->>Observer2: notify(data)
Observer1->>Subject: unsubscribe()
Common Patterns
| Component | Responsibility | Python Implementation |
|---|---|---|
| Subject | Maintains observers, sends notifications | Class with attach/detach/notify methods |
| Observer | Defines update interface | Abstract class or Protocol |
| Event | Carries notification data | Named tuple or dataclass |
Examples
Classic Observer Pattern:
from abc import ABC, abstractmethod
from typing import List
class Observer(ABC):
@abstractmethod
def update(self, subject):
pass
class Subject:
def __init__(self):
self._observers: List[Observer] = []
self._state = None
def attach(self, observer: Observer):
if observer not in self._observers:
self._observers.append(observer)
def detach(self, observer: Observer):
self._observers.remove(observer)
def notify(self):
for observer in self._observers:
observer.update(self)
@property
def state(self):
return self._state
@state.setter
def state(self, value):
self._state = value
self.notify()
class LoggingObserver(Observer):
def update(self, subject):
print(f"Logger: State changed to {subject.state}")
class EmailObserver(Observer):
def update(self, subject):
print(f"Email: Notifying about state {subject.state}")
# Usage
subject = Subject()
logger = LoggingObserver()
emailer = EmailObserver()
subject.attach(logger)
subject.attach(emailer)
subject.state = "active" # Both observers notified
Event-based Observer with Callbacks:
from collections import defaultdict
from typing import Callable, Any
class EventEmitter:
def __init__(self):
self._events = defaultdict(list)
def on(self, event: str, callback: Callable):
"""Subscribe to an event."""
self._events[event].append(callback)
return self # Allow chaining
def off(self, event: str, callback: Callable):
"""Unsubscribe from an event."""
self._events[event].remove(callback)
return self
def emit(self, event: str, *args, **kwargs):
"""Emit an event to all subscribers."""
for callback in self._events[event]:
callback(*args, **kwargs)
def once(self, event: str, callback: Callable):
"""Subscribe to an event only once."""
def wrapper(*args, **kwargs):
callback(*args, **kwargs)
self.off(event, wrapper)
self.on(event, wrapper)
return self
# Usage
emitter = EventEmitter()
def on_user_created(user):
print(f"User created: {user['name']}")
def send_welcome_email(user):
print(f"Sending welcome email to {user['email']}")
emitter.on('user_created', on_user_created)
emitter.on('user_created', send_welcome_email)
emitter.emit('user_created', {'name': 'Alice', 'email': 'alice@example.com'})
Property-based Observer with Descriptors:
class Observable:
def __init__(self, initial_value=None):
self._value = initial_value
self._callbacks = []
def subscribe(self, callback):
self._callbacks.append(callback)
def __get__(self, obj, objtype=None):
return self._value
def __set__(self, obj, value):
old_value = self._value
self._value = value
for callback in self._callbacks:
callback(old_value, value)
class Temperature:
celsius = Observable(0)
def __init__(self):
Temperature.celsius.subscribe(self._on_temperature_change)
def _on_temperature_change(self, old, new):
print(f"Temperature changed from {old}C to {new}C")
# Usage
temp = Temperature()
temp.celsius = 25 # Triggers callback
Decorator Pattern
Key Concepts
The Decorator pattern attaches additional responsibilities to an object dynamically. Python provides native support through function decorators and class decorators, making this pattern particularly elegant.
graph LR
subgraph "Function Decoration"
F["Original Function"] --> D1["Decorator 1"]
D1 --> D2["Decorator 2"]
D2 --> R["Result"]
end
Common Patterns
| Type | Syntax | Use Case |
|---|---|---|
| Function decorator | @decorator |
Add behaviour to functions |
| Class decorator | @decorator on class |
Modify class behaviour |
| Decorator with args | @decorator(args) |
Configurable decorators |
| Method decorator | @staticmethod, @classmethod |
Change method binding |
Examples
Basic Function Decorator:
import functools
import time
def timing(func):
"""Measure execution time of a function."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
end = time.perf_counter()
print(f"{func.__name__} took {end - start:.4f} seconds")
return result
return wrapper
@timing
def slow_function():
time.sleep(1)
return "done"
result = slow_function() # Prints timing information
Decorator with Arguments:
import functools
def retry(max_attempts=3, exceptions=(Exception,)):
"""Retry a function on failure."""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
print(f"Attempt {attempt + 1} failed: {e}")
raise last_exception
return wrapper
return decorator
@retry(max_attempts=3, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url):
# Simulate network request
import random
if random.random() < 0.7:
raise ConnectionError("Network error")
return "data"
Class-based Decorator:
import functools
class Cache:
"""Decorator class for caching function results."""
def __init__(self, func):
functools.update_wrapper(self, func)
self.func = func
self.cache = {}
def __call__(self, *args):
if args in self.cache:
print(f"Cache hit for {args}")
return self.cache[args]
result = self.func(*args)
self.cache[args] = result
return result
def clear_cache(self):
self.cache.clear()
@Cache
def fibonacci(n):
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
print(fibonacci(10))
fibonacci.clear_cache()
Stacking Multiple Decorators:
import functools
def bold(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
return f"<b>{func(*args, **kwargs)}</b>"
return wrapper
def italic(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
return f"<i>{func(*args, **kwargs)}</i>"
return wrapper
def validate_input(func):
@functools.wraps(func)
def wrapper(text, *args, **kwargs):
if not isinstance(text, str):
raise TypeError("Input must be a string")
return func(text, *args, **kwargs)
return wrapper
@bold
@italic
@validate_input
def format_text(text):
return text.upper()
print(format_text("hello")) # <b><i>HELLO</i></b>
Class Decorator:
def add_repr(cls):
"""Add automatic __repr__ to a class."""
def __repr__(self):
attrs = ', '.join(f"{k}={v!r}" for k, v in self.__dict__.items())
return f"{cls.__name__}({attrs})"
cls.__repr__ = __repr__
return cls
@add_repr
class Person:
def __init__(self, name, age):
self.name = name
self.age = age
person = Person("Alice", 30)
print(person) # Person(name='Alice', age=30)
Context Managers
Key Concepts
Context managers ensure proper acquisition and release of resources using the with statement. They implement __enter__ and __exit__ methods or use the contextlib module for simpler implementations.
sequenceDiagram
participant Code
participant CM as Context Manager
participant Resource
Code->>CM: with statement
CM->>CM: __enter__()
CM->>Resource: Acquire resource
CM-->>Code: Return value
Note over Code: Execute block
Code->>CM: Exit (normal or exception)
CM->>Resource: Release resource
CM->>CM: __exit__(exc_info)
Common Patterns
| Method | Purpose | Parameters |
|---|---|---|
__enter__ |
Set up resource | self |
__exit__ |
Clean up resource | self, exc_type, exc_val, exc_tb |
@contextmanager |
Generator-based CM | Decorated generator function |
Examples
Class-based Context Manager:
class FileManager:
def __init__(self, filename, mode='r'):
self.filename = filename
self.mode = mode
self.file = None
def __enter__(self):
self.file = open(self.filename, self.mode)
return self.file
def __exit__(self, exc_type, exc_val, exc_tb):
if self.file:
self.file.close()
# Return False to propagate exceptions
# Return True to suppress exceptions
return False
# Usage
with FileManager('data.txt', 'w') as f:
f.write('Hello, World!')
Using contextlib.contextmanager:
from contextlib import contextmanager
import time
@contextmanager
def timer(label):
"""Context manager to measure execution time."""
start = time.perf_counter()
try:
yield
finally:
end = time.perf_counter()
print(f"{label}: {end - start:.4f} seconds")
# Usage
with timer("Data processing"):
# Perform operations
time.sleep(1)
Database Transaction Context Manager:
from contextlib import contextmanager
class Database:
def __init__(self):
self.in_transaction = False
def begin(self):
self.in_transaction = True
print("Transaction started")
def commit(self):
self.in_transaction = False
print("Transaction committed")
def rollback(self):
self.in_transaction = False
print("Transaction rolled back")
@contextmanager
def transaction(db):
"""Ensure transaction is committed or rolled back."""
db.begin()
try:
yield db
db.commit()
except Exception:
db.rollback()
raise
# Usage
db = Database()
with transaction(db):
# Perform database operations
pass # Automatically commits
# With exception
try:
with transaction(db):
raise ValueError("Something went wrong")
except ValueError:
pass # Transaction was rolled back
Nested Context Managers:
from contextlib import ExitStack
# Multiple context managers
with open('input.txt') as infile, open('output.txt', 'w') as outfile:
outfile.write(infile.read())
# Dynamic number of context managers
files = ['file1.txt', 'file2.txt', 'file3.txt']
with ExitStack() as stack:
file_handles = [
stack.enter_context(open(fname, 'w'))
for fname in files
]
for i, fh in enumerate(file_handles):
fh.write(f"Content for file {i}")
Reusable Context Manager:
from contextlib import contextmanager
@contextmanager
def temporary_change(obj, attr, value):
"""Temporarily change an attribute value."""
original = getattr(obj, attr)
setattr(obj, attr, value)
try:
yield
finally:
setattr(obj, attr, original)
# Usage
class Config:
debug = False
config = Config()
with temporary_change(config, 'debug', True):
print(config.debug) # True
# Perform debug operations
print(config.debug) # False (restored)
Dependency Injection
Key Concepts
Dependency Injection (DI) is a technique where dependencies are passed to objects rather than created internally. This improves testability, flexibility, and adherence to SOLID principles.
graph TB
subgraph "Without DI"
A1["Service"] --> B1["Creates Database"]
end
subgraph "With DI"
A2["Service"] --> B2["Receives Database"]
C["Injector/Container"] --> B2
C --> A2
end
Common Patterns
| Type | Description | Implementation |
|---|---|---|
| Constructor Injection | Dependencies via __init__ |
Most common approach |
| Setter Injection | Dependencies via methods | For optional dependencies |
| Interface Injection | Dependency provides injector | Less common in Python |
Examples
Constructor Injection:
from abc import ABC, abstractmethod
class EmailService(ABC):
@abstractmethod
def send(self, to: str, subject: str, body: str):
pass
class SMTPEmailService(EmailService):
def __init__(self, host: str, port: int):
self.host = host
self.port = port
def send(self, to: str, subject: str, body: str):
print(f"Sending via SMTP to {to}: {subject}")
class MockEmailService(EmailService):
def __init__(self):
self.sent_emails = []
def send(self, to: str, subject: str, body: str):
self.sent_emails.append({'to': to, 'subject': subject, 'body': body})
class UserService:
def __init__(self, email_service: EmailService):
self.email_service = email_service
def register_user(self, email: str, name: str):
# Registration logic
self.email_service.send(
email,
"Welcome!",
f"Hello {name}, welcome to our platform!"
)
# Production
smtp_service = SMTPEmailService('smtp.example.com', 587)
user_service = UserService(smtp_service)
# Testing
mock_service = MockEmailService()
test_user_service = UserService(mock_service)
test_user_service.register_user('test@example.com', 'Test')
assert len(mock_service.sent_emails) == 1
Simple DI Container:
from typing import Type, TypeVar, Dict, Any
T = TypeVar('T')
class Container:
def __init__(self):
self._services: Dict[type, Any] = {}
self._factories: Dict[type, callable] = {}
def register(self, interface: type, implementation: type):
"""Register an implementation for an interface."""
self._factories[interface] = implementation
def register_instance(self, interface: type, instance):
"""Register a specific instance (singleton)."""
self._services[interface] = instance
def resolve(self, interface: Type[T]) -> T:
"""Resolve a dependency."""
# Check for registered instance
if interface in self._services:
return self._services[interface]
# Check for registered factory
if interface in self._factories:
implementation = self._factories[interface]
# Auto-resolve constructor dependencies
instance = self._create_instance(implementation)
return instance
raise ValueError(f"No registration for {interface}")
def _create_instance(self, cls):
"""Create instance with auto-resolved dependencies."""
import inspect
sig = inspect.signature(cls.__init__)
deps = {}
for name, param in sig.parameters.items():
if name == 'self':
continue
if param.annotation != inspect.Parameter.empty:
deps[name] = self.resolve(param.annotation)
return cls(**deps)
# Usage — register a concrete instance against the interface it satisfies
container = Container()
smtp = SMTPEmailService('smtp.example.com', 587)
container.register_instance(EmailService, smtp)
email_service = container.resolve(EmailService)
Limitation: auto-wiring only resolves constructor parameters whose annotated types are registered in the container; unregistered primitives (
str,int,Path) raise at resolve time — register those explicitly viaregister_instance.
Decorator-based Injection:
from functools import wraps
_dependencies = {}
def provide(name):
"""Decorator to register a dependency provider."""
def decorator(func):
_dependencies[name] = func
return func
return decorator
def inject(**deps):
"""Decorator to inject dependencies into a function."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for param_name, dep_name in deps.items():
if param_name not in kwargs:
provider = _dependencies.get(dep_name)
if provider:
kwargs[param_name] = provider()
return func(*args, **kwargs)
return wrapper
return decorator
# Register providers
@provide('database')
def get_database():
return {'connection': 'postgresql://localhost/db'}
@provide('logger')
def get_logger():
import logging
return logging.getLogger('app')
# Use injection
@inject(db='database', logger='logger')
def process_data(data, db=None, logger=None):
logger.info(f"Processing with {db['connection']}")
return f"Processed: {data}"
result = process_data("test data")
Repository Pattern
Key Concepts
The Repository pattern mediates between the domain and data mapping layers, acting like an in-memory collection of domain objects. It centralises data access logic and provides a clean API for domain operations.
classDiagram
class Repository {
<<interface>>
+add(entity)
+get(id) Entity
+list() List~Entity~
+update(entity)
+delete(id)
}
class SQLRepository {
-session: Session
+add(entity)
+get(id) Entity
+list() List~Entity~
}
class InMemoryRepository {
-storage: Dict
+add(entity)
+get(id) Entity
+list() List~Entity~
}
Repository <|.. SQLRepository
Repository <|.. InMemoryRepository
Service --> Repository : uses
Common Patterns
| Method | Purpose | Return Type |
|---|---|---|
add(entity) |
Add new entity | Entity or ID |
get(id) |
Retrieve by ID | Entity or None |
list(**filters) |
Query multiple | List[Entity] |
update(entity) |
Update existing | Entity |
delete(id) |
Remove entity | bool |
Examples
Abstract Repository Interface:
from abc import ABC, abstractmethod
from typing import List, Optional, TypeVar, Generic
from dataclasses import dataclass, field
import uuid
T = TypeVar('T')
@dataclass
class Entity:
id: str = field(default_factory=lambda: str(uuid.uuid4()))
@dataclass
class User(Entity):
name: str = ""
email: str = ""
class Repository(ABC, Generic[T]):
@abstractmethod
def add(self, entity: T) -> T:
pass
@abstractmethod
def get(self, id: str) -> Optional[T]:
pass
@abstractmethod
def list(self) -> List[T]:
pass
@abstractmethod
def update(self, entity: T) -> T:
pass
@abstractmethod
def delete(self, id: str) -> bool:
pass
In-Memory Repository Implementation:
class InMemoryRepository(Repository[T]):
def __init__(self):
self._storage: dict = {}
def add(self, entity: T) -> T:
self._storage[entity.id] = entity
return entity
def get(self, id: str) -> Optional[T]:
return self._storage.get(id)
def list(self) -> List[T]:
return list(self._storage.values())
def update(self, entity: T) -> T:
if entity.id not in self._storage:
raise ValueError(f"Entity {entity.id} not found")
self._storage[entity.id] = entity
return entity
def delete(self, id: str) -> bool:
if id in self._storage:
del self._storage[id]
return True
return False
class UserRepository(InMemoryRepository[User]):
def find_by_email(self, email: str) -> Optional[User]:
for user in self._storage.values():
if user.email == email:
return user
return None
# Usage
repo = UserRepository()
user = User(name="Alice", email="alice@example.com")
repo.add(user)
found = repo.find_by_email("alice@example.com")
print(found.name) # Alice
SQLAlchemy Repository:
from sqlalchemy import select
from sqlalchemy.orm import Session
from typing import Type
class SQLAlchemyRepository(Repository[T]):
def __init__(self, session: Session, model_class: Type[T]):
self.session = session
self.model_class = model_class
def add(self, entity: T) -> T:
self.session.add(entity)
self.session.commit()
self.session.refresh(entity)
return entity
def get(self, id: str) -> Optional[T]:
return self.session.get(self.model_class, id)
def list(self) -> List[T]:
return self.session.scalars(select(self.model_class)).all()
def update(self, entity: T) -> T:
self.session.merge(entity)
self.session.commit()
return entity
def delete(self, id: str) -> bool:
entity = self.get(id)
if entity:
self.session.delete(entity)
self.session.commit()
return True
return False
def find_by(self, **kwargs) -> List[T]:
return self.session.scalars(select(self.model_class).filter_by(**kwargs)).all()
Repository with Unit of Work:
from contextlib import contextmanager
class UnitOfWork:
def __init__(self):
self.users = InMemoryRepository[User]()
self._committed = False
def commit(self):
# In real implementation, persist all changes
self._committed = True
def rollback(self):
# In real implementation, discard all changes
pass
@contextmanager
def unit_of_work():
uow = UnitOfWork()
try:
yield uow
uow.commit()
except Exception:
uow.rollback()
raise
# Usage
with unit_of_work() as uow:
user = User(name="Bob", email="bob@example.com")
uow.users.add(user)
# Automatically commits on success
Strategy Pattern
Key Concepts
The Strategy pattern defines a family of algorithms, encapsulates each one, and makes them interchangeable. It lets the algorithm vary independently from clients that use it. Python's first-class functions make this pattern particularly clean.
classDiagram
class Context {
-strategy: Strategy
+set_strategy(strategy)
+execute_strategy()
}
class Strategy {
<<interface>>
+execute(data)
}
class ConcreteStrategyA {
+execute(data)
}
class ConcreteStrategyB {
+execute(data)
}
Context --> Strategy
Strategy <|.. ConcreteStrategyA
Strategy <|.. ConcreteStrategyB
Common Patterns
| Approach | Pros | Cons |
|---|---|---|
| Class-based | Clear structure, stateful | More boilerplate |
| Function-based | Simple, Pythonic | Less structure |
| Dict mapping | Easy lookup | Limited flexibility |
Examples
Class-based Strategy:
from abc import ABC, abstractmethod
from typing import List
class SortStrategy(ABC):
@abstractmethod
def sort(self, data: List) -> List:
pass
class QuickSortStrategy(SortStrategy):
def sort(self, data: List) -> List:
if len(data) <= 1:
return data
pivot = data[len(data) // 2]
left = [x for x in data if x < pivot]
middle = [x for x in data if x == pivot]
right = [x for x in data if x > pivot]
return self.sort(left) + middle + self.sort(right)
class MergeSortStrategy(SortStrategy):
def sort(self, data: List) -> List:
if len(data) <= 1:
return data
mid = len(data) // 2
left = self.sort(data[:mid])
right = self.sort(data[mid:])
return self._merge(left, right)
def _merge(self, left, right):
result = []
i = j = 0
while i < len(left) and j < len(right):
if left[i] <= right[j]:
result.append(left[i])
i += 1
else:
result.append(right[j])
j += 1
result.extend(left[i:])
result.extend(right[j:])
return result
class Sorter:
def __init__(self, strategy: SortStrategy = None):
self._strategy = strategy or QuickSortStrategy()
def set_strategy(self, strategy: SortStrategy):
self._strategy = strategy
def sort(self, data: List) -> List:
return self._strategy.sort(data.copy())
# Usage
sorter = Sorter()
data = [3, 1, 4, 1, 5, 9, 2, 6]
print(sorter.sort(data)) # Using QuickSort
sorter.set_strategy(MergeSortStrategy())
print(sorter.sort(data)) # Using MergeSort
Function-based Strategy (Pythonic):
from typing import Callable
def discount_none(price: float) -> float:
return price
def discount_percentage(percentage: float):
def apply(price: float) -> float:
return price * (1 - percentage / 100)
return apply
def discount_fixed(amount: float):
def apply(price: float) -> float:
return max(0, price - amount)
return apply
class ShoppingCart:
def __init__(self):
self.items = []
self._discount_strategy: Callable[[float], float] = discount_none
def add_item(self, name: str, price: float):
self.items.append({'name': name, 'price': price})
def set_discount(self, strategy: Callable[[float], float]):
self._discount_strategy = strategy
def total(self) -> float:
subtotal = sum(item['price'] for item in self.items)
return self._discount_strategy(subtotal)
# Usage
cart = ShoppingCart()
cart.add_item("Book", 20.0)
cart.add_item("Pen", 5.0)
print(f"No discount: {cart.total()}") # 25.0
cart.set_discount(discount_percentage(10))
print(f"10% off: {cart.total()}") # 22.5
cart.set_discount(discount_fixed(5))
print(f"5 off: {cart.total()}") # 20.0
Strategy with Registry:
class CompressionStrategy:
_strategies = {}
@classmethod
def register(cls, name):
def decorator(func):
cls._strategies[name] = func
return func
return decorator
@classmethod
def compress(cls, data: bytes, algorithm: str) -> bytes:
if algorithm not in cls._strategies:
raise ValueError(f"Unknown algorithm: {algorithm}")
return cls._strategies[algorithm](data)
@CompressionStrategy.register('gzip')
def gzip_compress(data: bytes) -> bytes:
import gzip
return gzip.compress(data)
@CompressionStrategy.register('zlib')
def zlib_compress(data: bytes) -> bytes:
import zlib
return zlib.compress(data)
@CompressionStrategy.register('none')
def no_compress(data: bytes) -> bytes:
return data
# Usage
data = b"Hello, World!" * 100
compressed = CompressionStrategy.compress(data, 'gzip')
print(f"Original: {len(data)}, Compressed: {len(compressed)}")
Error Handling Patterns
Key Concepts
Python's exception handling provides powerful mechanisms for managing errors. Well-designed error handling improves code reliability, debugging, and user experience.
graph TD
A["try block"] --> B{"Exception?"}
B -->|No| C["else block"]
B -->|Yes| D["except block"]
C --> E["finally block"]
D --> E
E --> F["Continue execution"]
Common Patterns
| Pattern | Use Case | Example |
|---|---|---|
| EAFP | Assume success, handle failure | try: x[key] except KeyError |
| LBYL | Check before acting | if key in x: x[key] |
| Custom exceptions | Domain-specific errors | class ValidationError(Exception) |
| Exception chaining | Preserve error context | raise NewError from original |
Examples
Custom Exception Hierarchy:
class ApplicationError(Exception):
"""Base exception for application errors."""
def __init__(self, message: str, code: str = None):
super().__init__(message)
self.message = message
self.code = code or self.__class__.__name__
class ValidationError(ApplicationError):
"""Raised when input validation fails."""
pass
class NotFoundError(ApplicationError):
"""Raised when a requested resource is not found."""
pass
class AuthenticationError(ApplicationError):
"""Raised when authentication fails."""
pass
class AuthorisationError(ApplicationError):
"""Raised when user lacks required permissions."""
pass
# Usage
def get_user(user_id: int):
if user_id < 0:
raise ValidationError(f"Invalid user ID: {user_id}", "INVALID_ID")
user = None # Simulate database lookup
if not user:
raise NotFoundError(f"User {user_id} not found", "USER_NOT_FOUND")
return user
try:
user = get_user(-1)
except ValidationError as e:
print(f"Validation failed [{e.code}]: {e.message}")
except NotFoundError as e:
print(f"Not found [{e.code}]: {e.message}")
Exception Chaining:
class DatabaseError(Exception):
pass
class ConnectionError(DatabaseError):
pass
def connect_to_database(connection_string):
try:
# Simulate connection attempt
raise OSError("Network unreachable")
except OSError as e:
raise ConnectionError(
f"Failed to connect to database"
) from e
try:
connect_to_database("postgresql://localhost/db")
except ConnectionError as e:
print(f"Error: {e}")
print(f"Caused by: {e.__cause__}")
Context Manager for Error Handling:
from contextlib import contextmanager
import logging
logger = logging.getLogger(__name__)
@contextmanager
def error_handler(operation_name: str, reraise: bool = True):
"""Context manager for consistent error handling."""
try:
yield
except ValidationError as e:
logger.warning(f"{operation_name} validation failed: {e}")
if reraise:
raise
except NotFoundError as e:
logger.info(f"{operation_name} resource not found: {e}")
if reraise:
raise
except Exception as e:
logger.error(f"{operation_name} unexpected error: {e}", exc_info=True)
if reraise:
raise
# Usage
with error_handler("User registration"):
# Perform operation
pass
Result Type Pattern (Rust-inspired):
from dataclasses import dataclass
from typing import TypeVar, Generic, Union
T = TypeVar('T')
E = TypeVar('E')
@dataclass
class Ok(Generic[T]):
value: T
def is_ok(self) -> bool:
return True
def is_err(self) -> bool:
return False
@dataclass
class Err(Generic[E]):
error: E
def is_ok(self) -> bool:
return False
def is_err(self) -> bool:
return True
Result = Union[Ok[T], Err[E]]
def divide(a: float, b: float) -> Result[float, str]:
if b == 0:
return Err("Division by zero")
return Ok(a / b)
def safe_parse_int(value: str) -> Result[int, str]:
try:
return Ok(int(value))
except ValueError:
return Err(f"Cannot parse '{value}' as integer")
# Usage
result = divide(10, 2)
if result.is_ok():
print(f"Result: {result.value}")
else:
print(f"Error: {result.error}")
# Chaining operations
parse_result = safe_parse_int("42")
if parse_result.is_ok():
div_result = divide(100, parse_result.value)
if div_result.is_ok():
print(f"Final result: {div_result.value}")
Retry Pattern with Exponential Backoff:
import time
import random
from functools import wraps
def retry_with_backoff(
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
exceptions: tuple = (Exception,)
):
"""Decorator for retrying with exponential backoff."""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
retries = 0
while True:
try:
return func(*args, **kwargs)
except exceptions as e:
retries += 1
if retries > max_retries:
raise
# Calculate delay with jitter
delay = min(base_delay * (2 ** (retries - 1)), max_delay)
jitter = random.uniform(0, delay * 0.1)
sleep_time = delay + jitter
print(f"Retry {retries}/{max_retries} after {sleep_time:.2f}s: {e}")
time.sleep(sleep_time)
return wrapper
return decorator
@retry_with_backoff(max_retries=3, base_delay=0.1, exceptions=(ConnectionError,))
def fetch_data(url):
# Simulate intermittent failures
if random.random() < 0.7:
raise ConnectionError("Connection failed")
return "data"
Quick Reference
| Pattern | Purpose | Key Implementation |
|---|---|---|
| Singleton | Single instance | __new__, metaclass, or module |
| Factory | Object creation | Factory function or method |
| Observer | Event notification | Subject/Observer with callbacks |
| Decorator | Add behaviour | @decorator with functools.wraps |
| Context Manager | Resource management | __enter__/__exit__ or @contextmanager |
| Dependency Injection | Loose coupling | Constructor injection |
| Repository | Data access abstraction | Interface with CRUD operations |
| Strategy | Interchangeable algorithms | Function or class strategies |
| Error Handling | Exception management | Custom exceptions with hierarchy |
Pattern Selection Guide
# When to use each pattern:
# Singleton - Global state, configuration, connection pools
config = get_config() # Returns same instance
# Factory - Complex object creation, multiple implementations
serialiser = SerialiserFactory.create('json')
# Observer - Event-driven systems, pub/sub
emitter.on('user_created', send_welcome_email)
# Decorator - Cross-cutting concerns (logging, caching, validation)
@cached
@logged
def expensive_operation():
pass
# Context Manager - Resource management, setup/teardown
with database.transaction():
# Operations are atomic
# Dependency Injection - Testability, flexibility
service = UserService(email_service=mock_email)
# Repository - Data access layer, domain separation
users = user_repository.find_by(active=True)
# Strategy - Multiple algorithms, runtime selection
sorter.set_strategy(QuickSortStrategy())
# Error Handling - Reliable, debuggable code
raise ValidationError("Invalid input", code="E001")
Common Issues and Solutions
Issue: Decorator Loses Function Metadata
Problem: Decorated functions lose their __name__, __doc__, and other attributes.
Solution: Use functools.wraps:
from functools import wraps
def my_decorator(func):
@wraps(func) # Preserves function metadata
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
Issue: Singleton Not Thread-Safe
Problem: Multiple threads can create multiple instances.
Solution: Use threading lock:
import threading
class Singleton:
_instance = None
_lock = threading.Lock()
def __new__(cls):
with cls._lock:
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
Issue: Circular Dependencies with Dependency Injection
Problem: Class A depends on B, and B depends on A.
Solution: Use lazy loading or restructure dependencies:
class ServiceA:
def __init__(self, service_b_factory):
self._service_b_factory = service_b_factory
self._service_b = None
@property
def service_b(self):
if self._service_b is None:
self._service_b = self._service_b_factory()
return self._service_b
Issue: Context Manager Exception Swallowing
Problem: Returning True from __exit__ suppresses all exceptions.
Solution: Only suppress specific exceptions:
def __exit__(self, exc_type, exc_val, exc_tb):
self.cleanup()
# Only suppress specific exceptions
if exc_type is ExpectedException:
return True # Suppress
return False # Propagate all other exceptions
Issue: Observer Memory Leaks
Problem: Observers hold references preventing garbage collection.
Solution: Use weak references:
import weakref
class Subject:
def __init__(self):
self._observers = weakref.WeakSet()
def attach(self, observer):
self._observers.add(observer)
Issue: Strategy Pattern Boilerplate
Problem: Too many classes for simple strategies.
Solution: Use functions instead of classes:
# Instead of multiple strategy classes
def process(data, strategy):
return strategy(data)
# Just pass functions
result = process(data, lambda x: x.upper())
Issue: Repository Testing with Real Database
Problem: Tests are slow and unreliable with real database.
Solution: Use in-memory repository for tests:
# Production
repo = SQLAlchemyUserRepository(session)
# Testing
repo = InMemoryUserRepository()
repo.add(User(id="1", name="Test"))
Issue: Custom Exceptions Without Context
Problem: Exceptions don't provide enough debugging information.
Solution: Include context and use exception chaining:
class DetailedError(Exception):
def __init__(self, message, context=None):
super().__init__(message)
self.context = context or {}
try:
process_user(user_id)
except DatabaseError as e:
raise DetailedError(
"Failed to process user",
context={'user_id': user_id, 'operation': 'update'}
) from e
Related Topics
The following topics would complement this Python Patterns cheatsheet:
- Python Type Hints and Protocols - Static typing, Protocol classes, and generic types for better pattern implementations
- Python Testing Patterns - Unit testing, mocking, fixtures, and test doubles for pattern testing
- Python Async Patterns - Asynchronous implementations of these patterns using asyncio
- SOLID Principles in Python - Design principles that inform when and how to apply these patterns
- Python Metaclasses and Descriptors - Advanced features for implementing patterns like Singleton and Observer
- Python Dataclasses and attrs - Modern approaches to creating domain objects for Repository and Factory patterns