Available for day contractsFrom 21st September I have availability for day and half day contracts. Please contact for more information.

Contact →
mikepreston.org

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.

Pythonic PatternsContext ManagersError HandlingDependency InjectionBehavioural PatternsObserverStrategyStructural PatternsDecoratorRepositoryCreational PatternsSingletonFactoryPython ApplicationPythonic PatternsContext ManagersError HandlingDependency InjectionBehavioural PatternsObserverStrategyStructural PatternsDecoratorRepositoryCreational PatternsSingletonFactoryPython 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.

requests instanceSingleton- instance: Singleton- init ()+get_instance() : Singleton+some_operation()Clientrequests instanceSingleton- instance: Singleton- init ()+get_instance() : Singleton+some_operation()Client

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.

«abstract»Creator+factory_method() : Product+some_operation()ConcreteCreatorA+factory_method() : ProductAConcreteCreatorB+factory_method() : ProductB«interface»Product+operation()ProductA+operation()ProductB+operation()«abstract»Creator+factory_method() : Product+some_operation()ConcreteCreatorA+factory_method() : ProductAConcreteCreatorB+factory_method() : ProductB«interface»Product+operation()ProductA+operation()ProductB+operation()

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.

Observer2Observer1SubjectObserver2Observer1SubjectState changessubscribe()subscribe()notify(data)notify(data)unsubscribe()Observer2Observer1SubjectObserver2Observer1SubjectState changessubscribe()subscribe()notify(data)notify(data)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.

Function DecorationOriginal FunctionDecorator 1Decorator 2ResultFunction DecorationOriginal FunctionDecorator 1Decorator 2Result

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.

ResourceContext ManagerCodeResourceContext ManagerCodeExecute blockwith statement__enter__()Acquire resourceReturn valueExit (normal or exception)Release resource__exit__(exc_info)ResourceContext ManagerCodeResourceContext ManagerCodeExecute blockwith statement__enter__()Acquire resourceReturn valueExit (normal or exception)Release resource__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.

With DIServiceReceives DatabaseInjector/ContainerWithout DIServiceCreates DatabaseWith DIServiceReceives DatabaseInjector/ContainerWithout DIServiceCreates Database

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 via register_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.

uses«interface»Repository+add(entity)+get(id) : Entity+list() : List <Entity>+update(entity)+delete(id)SQLRepository-session: Session+add(entity)+get(id) : Entity+list() : List <Entity>InMemoryRepository-storage: Dict+add(entity)+get(id) : Entity+list() : List <Entity>Serviceuses«interface»Repository+add(entity)+get(id) : Entity+list() : List <Entity>+update(entity)+delete(id)SQLRepository-session: Session+add(entity)+get(id) : Entity+list() : List <Entity>InMemoryRepository-storage: Dict+add(entity)+get(id) : Entity+list() : List <Entity>Service

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.

Context-strategy: Strategy+set_strategy(strategy)+execute_strategy()«interface»Strategy+execute(data)ConcreteStrategyA+execute(data)ConcreteStrategyB+execute(data)Context-strategy: Strategy+set_strategy(strategy)+execute_strategy()«interface»Strategy+execute(data)ConcreteStrategyA+execute(data)ConcreteStrategyB+execute(data)

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.

NoYestry blockException?else blockexcept blockfinally blockContinue executionNoYestry blockException?else blockexcept blockfinally blockContinue 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:

  1. Python Type Hints and Protocols - Static typing, Protocol classes, and generic types for better pattern implementations
  2. Python Testing Patterns - Unit testing, mocking, fixtures, and test doubles for pattern testing
  3. Python Async Patterns - Asynchronous implementations of these patterns using asyncio
  4. SOLID Principles in Python - Design principles that inform when and how to apply these patterns
  5. Python Metaclasses and Descriptors - Advanced features for implementing patterns like Singleton and Observer
  6. Python Dataclasses and attrs - Modern approaches to creating domain objects for Repository and Factory patterns