10 Python Design Patterns to Eliminate Spaghetti Code and Build Maintainable Projects

Why You Need These Patterns

When a simple script grows into a multi‑module application, developers often face problems like modifying many files for a new feature, difficult testing, unexpected side effects, and steep onboarding for new teammates. The root cause is a lack of proper architectural patterns, which leads to technical debt.

1. Dependency Injection: Say Goodbye to Hard‑Coded Coupling

Most developers avoid DI in Python, assuming its dynamic nature makes it unnecessary—until they try to unit‑test a class that directly creates a database connection.

# Bad practice: hard‑coded dependency
class BadUserManager:
    def __init__(self):
        self.email_service = EmailService()  # hard‑coded!
    def register(self, user):
        self.email_service.send(user, "Welcome!")

# Correct practice: inject dependency
class EmailService:
    def send(self, to, message):
        print(f"Sending email to {to}: {message}")

class UserManager:
    def __init__(self, email_service):
        self.email_service = email_service
    def register(self, user):
        self.email_service.send(user, "Welcome!")

email_service = EmailService()
manager = UserManager(email_service)
manager.register("[email protected]")

# In tests you can inject a mock
class MockEmailService:
    def send(self, to, message):
        print(f"[Test] Mock send to {to}")

test_manager = UserManager(MockEmailService())
test_manager.register("[email protected]")

In FastAPI or Django, DI is commonly used for database sessions, configuration objects, etc.

2. Strategy Pattern: Replace an If/Else Jungle

When a function contains a large decision tree, the code becomes a monster. The Strategy pattern swaps the tangled logic for interchangeable behavior objects.

# Before: if/else hell
def process_payment(amount, payment_method):
    if payment_method == "paypal":
        pass  # PayPal logic
    elif payment_method == "stripe":
        pass  # Stripe logic
    elif payment_method == "alipay":
        pass  # Alipay logic
    # ... more elifs

# After: strategy pattern
from abc import ABC, abstractmethod

class PaymentStrategy(ABC):
    @abstractmethod
    def pay(self, amount):
        pass

class PayPalPayment(PaymentStrategy):
    def pay(self, amount):
        print(f"Pay via PayPal {amount} yuan")
        return True

class StripePayment(PaymentStrategy):
    def pay(self, amount):
        print(f"Pay via Stripe {amount} yuan")
        return True

class AlipayPayment(PaymentStrategy):
    def pay(self, amount):
        print(f"Pay via Alipay {amount} yuan")
        return True

class PaymentProcessor:
    def __init__(self):
        self.strategies = {
            "paypal": PayPalPayment(),
            "stripe": StripePayment(),
            "alipay": AlipayPayment()
        }
    def process(self, method, amount):
        strategy = self.strategies.get(method)
        if not strategy:
            raise ValueError(f"Unsupported payment method: {method}")
        return strategy.pay(amount)

processor = PaymentProcessor()
processor.process("paypal", 100)
processor.process("alipay", 200)
# Adding a new method is as easy as adding a new class and entry
class WeChatPayment(PaymentStrategy):
    def pay(self, amount):
        print(f"Pay via WeChat {amount} yuan")
        return True
processor.strategies["wechat"] = WeChatPayment()

This adheres to the Open/Closed Principle.

3. Builder Pattern: Tame Bloated Constructors

When a class has dozens of optional parameters, the constructor becomes a hostage‑taking scenario.

# Problem: constructor with many arguments
class User:
    def __init__(self, name, email=None, phone=None, age=None,
                 address=None, city=None, country=None,
                 is_active=True, is_admin=False):
        pass

# Usage is painful
user = User(name="张三", email="[email protected]", phone="13800138000",
            age=25, address="某街道123号", city="北京", country="中国",
            is_active=True, is_admin=False)

# Builder solution
class QueryBuilder:
    def __init__(self, table="users"):
        self.table = table
        self._select_fields = ["*"]
        self._where_conditions = []
        self._order_by_fields = []
        self._limit_value = None
    def select(self, *fields):
        self._select_fields = fields
        return self
    def where(self, condition):
        self._where_conditions.append(condition)
        return self
    def order_by(self, field, descending=False):
        direction = "DESC" if descending else "ASC"
        self._order_by_fields.append(f"{field} {direction}")
        return self
    def limit(self, count):
        self._limit_value = count
        return self
    def build(self):
        select_clause = f"SELECT {', '.join(self._select_fields)} FROM {self.table}"
        where_clause = f" WHERE {' AND '.join(self._where_conditions)}" if self._where_conditions else ""
        order_clause = f" ORDER BY {', '.join(self._order_by_fields)}" if self._order_by_fields else ""
        limit_clause = f" LIMIT {self._limit_value}" if self._limit_value else ""
        return select_clause + where_clause + order_clause + limit_clause

query = (QueryBuilder("users")
         .select("id", "name", "email")
         .where("age > 18")
         .where("is_active = TRUE")
         .order_by("created_at", descending=True)
         .limit(10)
         .build())
print(query)  # SELECT id, name, email FROM users WHERE age > 18 AND is_active = TRUE ORDER BY created_at DESC LIMIT 10

Modern ORMs such as SQLAlchemy, Django Q objects, and Pydantic models already employ a builder‑like API.

4. Event‑Driven Pattern: Decouple High‑Concurrency Systems

class EventBus:
    """Central hub for component communication"""
    def __init__(self):
        self._subscribers = {}
    def subscribe(self, event_type, callback):
        self._subscribers.setdefault(event_type, []).append(callback)
    def unsubscribe(self, event_type, callback):
        if event_type in self._subscribers:
            self._subscribers[event_type].remove(callback)
    def publish(self, event_type, data=None):
        for callback in self._subscribers.get(event_type, []):
            try:
                callback(data)
            except Exception as e:
                print(f"Event handling error: {e}")
    def get_subscriber_count(self, event_type):
        return len(self._subscribers.get(event_type, []))

# Example usage
USER_REGISTERED = "user_registered"
bus = EventBus()

def send_welcome_email(user):
    print(f"[Email] Send welcome to {user['email']}")

def init_user_storage(user):
    print(f"[Storage] Init storage for {user['username']}")

def log_user_registration(user):
    print(f"[Log] User registered: {user}")

bus.subscribe(USER_REGISTERED, send_welcome_email)
bus.subscribe(USER_REGISTERED, init_user_storage)
bus.subscribe(USER_REGISTERED, log_user_registration)

new_user = {"id": 123, "username": "zhangsan", "email": "[email protected]", "registered_at": "2024-01-01 10:00:00"}
print("User registration event triggered…")
bus.publish(USER_REGISTERED, new_user)
print(f"Event handling completed, {bus.get_subscriber_count(USER_REGISTERED)} subscribers")

This pattern is useful in micro‑service communication, plugin systems, and GUI event handling.

5. Repository Pattern: Keep SQL Out of Business Logic

import sqlite3
from contextlib import contextmanager
from typing import List, Optional, Dict, Any

@contextmanager
def get_db_connection():
    """Database connection context manager"""
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()

class UserRepository:
    """User data repository"""
    def __init__(self, db_connection):
        self.db = db_connection
        self._create_table()
    def _create_table(self):
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT NOT NULL UNIQUE,
                email TEXT NOT NULL UNIQUE,
                age INTEGER,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        self.db.commit()
    def get_by_id(self, user_id: int) -> Optional[Dict[str, Any]]:
        cursor = self.db.execute("SELECT * FROM users WHERE id = ?", (user_id,))
        row = cursor.fetchone()
        return dict(row) if row else None
    def get_by_email(self, email: str) -> Optional[Dict[str, Any]]:
        cursor = self.db.execute("SELECT * FROM users WHERE email = ?", (email,))
        row = cursor.fetchone()
        return dict(row) if row else None
    def get_all(self, limit: int = 100) -> List[Dict[str, Any]]:
        cursor = self.db.execute("SELECT * FROM users LIMIT ?", (limit,))
        return [dict(row) for row in cursor.fetchall()]
    def add(self, username: str, email: str, age: int = None) -> int:
        cursor = self.db.execute(
            "INSERT INTO users (username, email, age) VALUES (?, ?, ?)",
            (username, email, age)
        )
        self.db.commit()
        return cursor.lastrowid
    def update(self, user_id: int, **kwargs) -> bool:
        if not kwargs:
            return False
        set_clause = ", ".join([f"{k} = ?" for k in kwargs.keys()])
        values = list(kwargs.values()) + [user_id]
        self.db.execute(f"UPDATE users SET {set_clause} WHERE id = ?", values)
        self.db.commit()
        return True
    def delete(self, user_id: int) -> bool:
        cursor = self.db.execute("DELETE FROM users WHERE id = ?", (user_id,))
        self.db.commit()
        return cursor.rowcount > 0

# Usage example
with get_db_connection() as conn:
    repo = UserRepository(conn)
    user_id = repo.add("zhangsan", "[email protected]", 25)
    print(f"Added user, ID: {user_id}")
    user = repo.get_by_id(user_id)
    print(f"Fetched user: {user}")
    repo.update(user_id, age=26, email="[email protected]")
    updated_user = repo.get_by_id(user_id)
    print(f"After update: {updated_user}")
    all_users = repo.get_all()
    print(f"Total users: {len(all_users)}")

Changing the underlying database now only requires updating the repository implementation.

6. Mapper Pattern: Separate Data Structures from Domain Objects

from datetime import datetime
from typing import Optional

class User:
    """Domain model: user entity"""
    def __init__(self, id: int, username: str, email: str,
                 age: Optional[int] = None, created_at: Optional[datetime] = None):
        self.id = id
        self.username = username
        self.email = email
        self.age = age
        self.created_at = created_at or datetime.now()
    def __repr__(self):
        return f"User(id={self.id}, username='{self.username}', email='{self.email}')"
    def is_adult(self) -> bool:
        return self.age is not None and self.age >= 18
    def get_display_name(self) -> str:
        return f"{self.username} ({self.email})"

class UserMapper:
    """Convert between dicts (e.g., API JSON or DB rows) and User objects"""
    @staticmethod
    def from_dict(data: dict) -> User:
        return User(
            id=data.get("id"),
            username=data["username"],
            email=data["email"],
            age=data.get("age"),
            created_at=datetime.fromisoformat(data["created_at"]) if "created_at" in data else None
        )
    @staticmethod
    def from_database_row(row) -> User:
        return User(
            id=row["id"],
            username=row["username"],
            email=row["email"],
            age=row["age"],
            created_at=datetime.fromisoformat(row["created_at"]) if row["created_at"] else None
        )
    @staticmethod
    def to_dict(user: User) -> dict:
        return {
            "id": user.id,
            "username": user.username,
            "email": user.email,
            "age": user.age,
            "created_at": user.created_at.isoformat() if user.created_at else None,
            "is_adult": user.is_adult(),
            "display_name": user.get_display_name()
        }
    @staticmethod
    def to_database_params(user: User) -> tuple:
        return (
            user.username,
            user.email,
            user.age,
            user.created_at.isoformat() if user.created_at else None
        )

# Demo conversion
api_response = {
    "id": 123,
    "username": "zhangsan",
    "email": "[email protected]",
    "age": 25,
    "created_at": "2024-01-01T10:00:00"
}
user = UserMapper.from_dict(api_response)
print(f"Domain object: {user}")
print(f"Is adult? {user.is_adult()}")
print(f"Display name: {user.get_display_name()}")
user_dict = UserMapper.to_dict(user)
print(f"Back to dict: {user_dict}")

# Simulate a DB row
class MockDBRow:
    def __init__(self):
        self.data = {
            "id": 456,
            "username": "lisi",
            "email": "[email protected]",
            "age": 17,
            "created_at": "2024-01-02T09:00:00"
        }
    def __getitem__(self, key):
        return self.data[key]

db_row = MockDBRow()
user_from_db = UserMapper.from_database_row(db_row)
print(f"
User from DB: {user_from_db}")
print(f"Is adult? {user_from_db.is_adult()}")

The mapper eliminates the "dictionary apocalypse" and keeps data handling consistent.

7. Pipeline Pattern: Chain Processing Steps Cleanly

from abc import ABC, abstractmethod
from typing import Any, List

class PipelineStep(ABC):
    """Abstract base for a pipeline step"""
    @abstractmethod
    def process(self, data: Any) -> Any:
        pass
    def __call__(self, data: Any) -> Any:
        return self.process(data)

class TextCleanStep(PipelineStep):
    """Remove extra spaces, lower‑case, strip punctuation"""
    def process(self, text: str) -> str:
        text = ' '.join(text.split())
        text = text.lower()
        text = ''.join(c for c in text if c.isalnum() or c.isspace())
        return text

class TokenizeStep(PipelineStep):
    """Split text into tokens"""
    def __init__(self, tokenizer=None):
        self.tokenizer = tokenizer or str.split
    def process(self, text: str) -> List[str]:
        return self.tokenizer(text)

class StopwordRemovalStep(PipelineStep):
    """Remove common stopwords"""
    def __init__(self, stopwords=None):
        self.stopwords = stopwords or {"a", "an", "the", "and", "or", "in", "on", "at"}
    def process(self, tokens: List[str]) -> List[str]:
        return [t for t in tokens if t not in self.stopwords]

class StemmingStep(PipelineStep):
    """Very simple stemming example"""
    def process(self, tokens: List[str]) -> List[str]:
        stemmed = []
        for token in tokens:
            if token.endswith("ing"):
                token = token[:-3]
            elif token.endswith("ed"):
                token = token[:-2]
            elif token.endswith("s"):
                token = token[:-1]
            stemmed.append(token)
        return stemmed

class Pipeline:
    """Connect multiple steps"""
    def __init__(self, steps: List[PipelineStep] = None):
        self.steps = steps or []
    def add_step(self, step: PipelineStep):
        self.steps.append(step)
        return self
    def process(self, data: Any) -> Any:
        result = data
        for step in self.steps:
            print(f"Executing step: {step.__class__.__name__}")
            result = step.process(result)
            print(f"Current result: {result}")
        return result
    def __call__(self, data: Any) -> Any:
        return self.process(data)

# Build and run the pipeline
text_pipeline = Pipeline()
text_pipeline.add_step(TextCleanStep())
text_pipeline.add_step(TokenizeStep())
text_pipeline.add_step(StopwordRemovalStep())
text_pipeline.add_step(StemmingStep())

sample_text = "  I am RUNNING in the park and playing with dogs!  "
print("Original text:", sample_text)
print("
Starting pipeline processing…")
final_result = text_pipeline.process(sample_text)
print("
Final result:", final_result)

# Extend with a spell‑check step
class SpellCheckStep(PipelineStep):
    """Demo spell‑check (fixes a simple typo)"""
    def process(self, tokens: List[str]) -> List[str]:
        corrected = []
        for token in tokens:
            if token == "runn":
                token = "run"
            corrected.append(token)
        return corrected

text_pipeline.add_step(SpellCheckStep())
new_result = text_pipeline.process("I am runn in the parc")
print("
Result with spell‑check:", new_result)

The pipeline keeps your brain—and your project—clean and ordered.

8. Command Pattern: Enable Undo/Redo and "Do It Again"

from abc import ABC, abstractmethod
from typing import List, Dict, Any

class Command(ABC):
    """Abstract command"""
    @abstractmethod
    def execute(self) -> Any:
        pass
    @abstractmethod
    def undo(self) -> Any:
        pass
    def __str__(self):
        return self.__class__.__name__

class TextEditor:
    """Receiver that holds text and history"""
    def __init__(self):
        self.text = ""
        self.history: List[str] = []
    def add_text(self, new_text: str):
        self.history.append(self.text)
        self.text += new_text
        print(f"Added text: '{new_text}'")
        print(f"Current text: '{self.text}'")
    def delete_last_word(self) -> str:
        if self.text:
            self.history.append(self.text)
            words = self.text.split()
            if words:
                last_word = words[-1]
                self.text = ' '.join(words[:-1])
                print(f"Deleted last word: '{last_word}'")
                print(f"Current text: '{self.text}'")
                return last_word
        return ""
    def clear_text(self):
        self.history.append(self.text)
        self.text = ""
        print("Cleared text")
    def restore_from_history(self):
        if self.history:
            self.text = self.history.pop()
            print(f"Restored to: '{self.text}'")

class AddTextCommand(Command):
    """Add text command"""
    def __init__(self, editor: TextEditor, text: str):
        self.editor = editor
        self.text = text
        self.previous_text = ""
    def execute(self):
        self.previous_text = self.editor.text
        self.editor.add_text(self.text)
    def undo(self):
        self.editor.text = self.previous_text
        print(f"Undo add text, restored to: '{self.editor.text}'")

class DeleteLastWordCommand(Command):
    """Delete last word command"""
    def __init__(self, editor: TextEditor):
        self.editor = editor
        self.deleted_word = ""
        self.previous_text = ""
    def execute(self):
        self.previous_text = self.editor.text
        self.deleted_word = self.editor.delete_last_word()
    def undo(self):
        self.editor.text = self.previous_text
        print(f"Undo delete, restored to: '{self.editor.text}'")

class ClearTextCommand(Command):
    """Clear text command"""
    def __init__(self, editor: TextEditor):
        self.editor = editor
        self.previous_text = ""
    def execute(self):
        self.previous_text = self.editor.text
        self.editor.clear_text()
    def undo(self):
        self.editor.text = self.previous_text
        print(f"Undo clear, restored to: '{self.editor.text}'")

class CommandHistory:
    """Manage command execution, undo, redo"""
    def __init__(self):
        self.history: List[Command] = []
        self.undo_history: List[Command] = []
    def execute_command(self, command: Command):
        command.execute()
        self.history.append(command)
        self.undo_history.clear()
        print(f"Executed command: {command}")
    def undo(self):
        if self.history:
            command = self.history.pop()
            command.undo()
            self.undo_history.append(command)
            print(f"Undid command: {command}")
        else:
            print("No command to undo")
    def redo(self):
        if self.undo_history:
            command = self.undo_history.pop()
            command.execute()
            self.history.append(command)
            print(f"Redid command: {command}")
        else:
            print("No command to redo")
    def get_history(self) -> List[str]:
        return [str(cmd) for cmd in self.history]

# Demo
editor = TextEditor()
history = CommandHistory()
print("=== Text editor demo ===
")
add_hello = AddTextCommand(editor, "Hello ")
history.execute_command(add_hello)
add_world = AddTextCommand(editor, "World ")
history.execute_command(add_world)
add_python = AddTextCommand(editor, "Python ")
history.execute_command(add_python)
print(f"
Current text: '{editor.text}'")
print(f"Command history: {history.get_history()}")
print("
--- Undo ---")
history.undo()
print(f"After undo: '{editor.text}'")
history.undo()
print(f"After second undo: '{editor.text}'")
print("
--- Redo ---")
history.redo()
print(f"After redo: '{editor.text}'")
# Delete last word
print("
--- Delete ---")
delete_cmd = DeleteLastWordCommand(editor)
history.execute_command(delete_cmd)
# Clear text
print("
--- Clear ---")
clear_cmd = ClearTextCommand(editor)
history.execute_command(clear_cmd)
# Multiple undos
print("
--- Multiple undos ---")
for _ in range(3):
    history.undo()
print(f"
Final text: '{editor.text}'")
print(f"Final history: {history.get_history()}")

This pattern is ideal for UI applications, CLI tools, and editors where reversible actions are required.

9. Specification Pattern: Keep Filtering Logic Clean

from abc import ABC, abstractmethod
from typing import List, Dict, Any
from datetime import datetime, timedelta

class Specification(ABC):
    @abstractmethod
    def is_satisfied(self, item: Dict[str, Any]) -> bool:
        pass
    def __and__(self, other: 'Specification') -> 'AndSpecification':
        return AndSpecification(self, other)
    def __or__(self, other: 'Specification') -> 'OrSpecification':
        return OrSpecification(self, other)
    def __invert__(self) -> 'NotSpecification':
        return NotSpecification(self)
    def __call__(self, item: Dict[str, Any]) -> bool:
        return self.is_satisfied(item)

class AgeAboveSpecification(Specification):
    def __init__(self, min_age: int):
        self.min_age = min_age
    def is_satisfied(self, user: Dict[str, Any]) -> bool:
        age = user.get('age')
        return age is not None and age > self.min_age
    def __str__(self):
        return f"Age > {self.min_age}"

class HasEmailSpecification(Specification):
    def is_satisfied(self, user: Dict[str, Any]) -> bool:
        email = user.get('email')
        return bool(email and '@' in email)
    def __str__(self):
        return "Has Email"

class IsActiveSpecification(Specification):
    def __init__(self, days_threshold: int = 30):
        self.days_threshold = days_threshold
    def is_satisfied(self, user: Dict[str, Any]) -> bool:
        last_login = user.get('last_login')
        if not last_login:
            return False
        if isinstance(last_login, str):
            last_login = datetime.fromisoformat(last_login)
        days_since = (datetime.now() - last_login).days
        return days_since <= self.days_threshold
    def __str__(self):
        return f"Active within {self.days_threshold} days"

class BalanceAboveSpecification(Specification):
    def __init__(self, min_balance: float):
        self.min_balance = min_balance
    def is_satisfied(self, user: Dict[str, Any]) -> bool:
        balance = user.get('balance', 0)
        return balance >= self.min_balance
    def __str__(self):
        return f"Balance >= {self.min_balance}"

class AndSpecification(Specification):
    def __init__(self, *specs: Specification):
        self.specs = specs
    def is_satisfied(self, item: Dict[str, Any]) -> bool:
        return all(spec(item) for spec in self.specs)
    def __str__(self):
        return " AND ".join(str(s) for s in self.specs)

class OrSpecification(Specification):
    def __init__(self, *specs: Specification):
        self.specs = specs
    def is_satisfied(self, item: Dict[str, Any]) -> bool:
        return any(spec(item) for spec in self.specs)
    def __str__(self):
        return " OR ".join(str(s) for s in self.specs)

class NotSpecification(Specification):
    def __init__(self, spec: Specification):
        self.spec = spec
    def is_satisfied(self, item: Dict[str, Any]) -> bool:
        return not self.spec(item)
    def __str__(self):
        return f"NOT ({self.spec})"

class UserFilter:
    @staticmethod
    def filter_users(users: List[Dict[str, Any]], spec: Specification) -> List[Dict[str, Any]]:
        return [u for u in users if spec.is_satisfied(u)]
    @staticmethod
    def count_users(users: List[Dict[str, Any]], spec: Specification) -> int:
        return sum(1 for u in users if spec.is_satisfied(u))

# Sample data
users = [
    {"id":1, "name":"张三", "age":25, "email":"[email protected]", "balance":1000.0,
     "last_login":"2024-01-15T10:00:00", "is_active":True},
    {"id":2, "name":"李四", "age":17, "email":"[email protected]", "balance":500.0,
     "last_login":"2023-11-01T09:00:00", "is_active":True},
    {"id":3, "name":"王五", "age":30, "email":"", "balance":2000.0,
     "last_login":"2024-01-20T14:00:00", "is_active":False},
    {"id":4, "name":"赵六", "age":22, "email":"[email protected]", "balance":300.0,
     "last_login":"2024-01-18T16:00:00", "is_active":True},
    {"id":5, "name":"孙七", "age":35, "email":"[email protected]", "balance":1500.0,
     "last_login":"2023-12-01T08:00:00", "is_active":True}
]

print("=== User Filtering System ===
")
adult_spec = AgeAboveSpecification(18)
has_email_spec = HasEmailSpecification()
active_spec = IsActiveSpecification(days_threshold=30)
rich_spec = BalanceAboveSpecification(1000)

print("1. Adult users:")
adults = UserFilter.filter_users(users, adult_spec)
print(f"  Spec: {adult_spec}")
print(f"  Count: {len(adults)}")
for u in adults:
    print(f"  - {u['name']} ({u['age']}岁)")

print("
2. Active adults with email:")
complex_spec = adult_spec & has_email_spec & active_spec
filtered = UserFilter.filter_users(users, complex_spec)
print(f"  Spec: {complex_spec}")
print(f"  Count: {len(filtered)}")
for u in filtered:
    print(f"  - {u['name']}")

print("
3. Rich or active users:")
rich_or_active = rich_spec | active_spec
filtered = UserFilter.filter_users(users, rich_or_active)
print(f"  Spec: {rich_or_active}")
print(f"  Count: {len(filtered)}")
for u in filtered:
    print(f"  - {u['name']} (Balance: {u['balance']}, Last login: {u['last_login'][:10]})")

print("
4. Inactive users:")
not_active = ~active_spec
filtered = UserFilter.filter_users(users, not_active)
print(f"  Spec: {not_active}")
print(f"  Count: {len(filtered)}")
for u in filtered:
    print(f"  - {u['name']} (Last login: {u['last_login'][:10]})")

print("
5. Complex dynamic query:")
dynamic_spec = (
    adult_spec &
    has_email_spec &
    BalanceAboveSpecification(500) &
    (active_spec | BalanceAboveSpecification(1500))
)
print(f"  Spec: {dynamic_spec}")
count = UserFilter.count_users(users, dynamic_spec)
print(f"  Matching users: {count}")
filtered = UserFilter.filter_users(users, dynamic_spec)
for u in filtered:
    print(f"  - {u['name']}: Age {u['age']}, Balance {u['balance']}, Active {u['is_active']}")

The specification pattern lets you compose complex filters without tangled if‑else trees.

10. Plugin Registry Pattern: Smart Dynamic Extensions

from abc import ABC, abstractmethod
from typing import Dict, Any, Callable, Optional, Type
import importlib, pkgutil

class Plugin(ABC):
    @abstractmethod
    def execute(self, *args, **kwargs) -> Any:
        pass
    def get_name(self) -> str:
        return self.__class__.__name__
    def get_version(self) -> str:
        return "1.0.0"

class PluginRegistry:
    """Central registry for plugins"""
    _plugins: Dict[str, Type[Plugin]] = {}
    _instances: Dict[str, Plugin] = {}

    @classmethod
    def register(cls, name: Optional[str] = None):
        """Decorator to register a plugin class"""
        def decorator(plugin_cls: Type[Plugin]) -> Type[Plugin]:
            plugin_name = name or plugin_cls.__name__.lower()
            if plugin_name in cls._plugins:
                raise ValueError(f"Plugin '{plugin_name}' already registered")
            cls._plugins[plugin_name] = plugin_cls
            print(f"Registered plugin: {plugin_name} -> {plugin_cls.__name__}")
            return plugin_cls
        return decorator

    @classmethod
    def get_plugin(cls, name: str) -> Optional[Plugin]:
        if name not in cls._instances:
            if name in cls._plugins:
                cls._instances[name] = cls._plugins[name]()
            else:
                return None
        return cls._instances.get(name)

    @classmethod
    def get_all_plugins(cls) -> Dict[str, Plugin]:
        for name in cls._plugins:
            if name not in cls._instances:
                cls._instances[name] = cls._plugins[name]()
        return cls._instances.copy()

    @classmethod
    def list_plugins(cls) -> list:
        return list(cls._plugins.keys())

    @classmethod
    def execute_plugin(cls, name: str, *args, **kwargs) -> Any:
        plugin = cls.get_plugin(name)
        if not plugin:
            raise ValueError(f"Plugin not found: {name}")
        print(f"Executing plugin: {name}")
        return plugin.execute(*args, **kwargs)

    @classmethod
    def load_plugins_from_package(cls, package_name: str):
        """Automatically import all modules in a package to trigger registration"""
        try:
            package = importlib.import_module(package_name)
            for _, module_name, is_pkg in pkgutil.iter_modules(package.__path__):
                if not is_pkg:
                    full_name = f"{package_name}.{module_name}"
                    importlib.import_module(full_name)
            print(f"Loaded plugins from package '{package_name}'")
        except ImportError as e:
            print(f"Failed to load package '{package_name}': {e}")

# Define some plugins
@PluginRegistry.register("greeter")
class GreeterPlugin(Plugin):
    """Greeting plugin"""
    def execute(self, name: str = "World") -> str:
        greeting = f"Hello, {name}!"
        print(greeting)
        return greeting
    def get_version(self) -> str:
        return "2.0.0"

@PluginRegistry.register("calculator")
class CalculatorPlugin(Plugin):
    """Simple calculator"""
    def execute(self, operation: str, a: float, b: float) -> float:
        ops = {
            "add": lambda x, y: x + y,
            "subtract": lambda x, y: x - y,
            "multiply": lambda x, y: x * y,
            "divide": lambda x, y: x / y if y != 0 else float('inf')
        }
        if operation not in ops:
            raise ValueError(f"Unsupported operation: {operation}")
        result = ops[operation](a, b)
        print(f"{a} {operation} {b} = {result}")
        return result

@PluginRegistry.register("uppercase")
class UppercasePlugin(Plugin):
    """Convert text to uppercase"""
    def execute(self, text: str) -> str:
        result = text.upper()
        print(f"Uppercase: '{text}' -> '{result}'")
        return result

@PluginRegistry.register()  # default name based on class
class ReversePlugin(Plugin):
    """Reverse a string"""
    def execute(self, text: str) -> str:
        result = text[::-1]
        print(f"Reverse: '{text}' -> '{result}'")
        return result

# Dynamic plugin registration example
def register_dynamic_plugin():
    """Create and register a plugin at runtime"""
    def custom_function(x):
        return x * 2
    plugin_name = "dynamic_doubler"
    PluginRegistry._plugins[plugin_name] = type(
        "DynamicDoubler",
        (Plugin,),
        {
            "execute": lambda self, x: custom_function(x),
            "get_name": lambda self: plugin_name
        }
    )
    print(f"Dynamically registered plugin: {plugin_name}")

# Demo usage
print("=== Plugin system demo ===
")
print("Registered plugins:")
for pn in PluginRegistry.list_plugins():
    print(f"  - {pn}")

print("
1. Run greeter plugin:")
PluginRegistry.execute_plugin("greeter", "Python developer")

print("
2. Run calculator plugin:")
result = PluginRegistry.execute_plugin("calculator", "multiply", 6, 7)
print(f"   Result: {result}")

print("
3. Run uppercase plugin:")
PluginRegistry.execute_plugin("uppercase", "hello world")

print("
4. Run reverse plugin:")
PluginRegistry.execute_plugin("reverseplugin", "Python")

print("
5. Dynamic plugin demo:")
register_dynamic_plugin()
if "dynamic_doubler" in PluginRegistry.list_plugins():
    result = PluginRegistry.execute_plugin("dynamic_doubler", 21)
    print(f"   Dynamic plugin result: {result}")

print("
6. Get plugin info:")
greeter = PluginRegistry.get_plugin("greeter")
if greeter:
    print(f"   Plugin name: {greeter.get_name()}")
    print(f"   Plugin version: {greeter.get_version()}")

print("
7. Execute all plugins:")
all_plugins = PluginRegistry.get_all_plugins()
for name, plugin in all_plugins.items():
    print(f"  Executing {name}: ", end="")
    if name == "greeter":
        plugin.execute("All Plugins")
    elif name == "calculator":
        plugin.execute("add", 10, 20)
    elif name == "uppercase":
        plugin.execute("test")
    elif name == "reverseplugin":
        plugin.execute("abcd")
    elif name == "dynamic_doubler":
        res = plugin.execute(15)
        print(f"Result: {res}")
    else:
        print()

print("
8. Simulate loading plugins from a package (conceptual):")
print("   PluginRegistry.load_plugins_from_package('my_plugins')")

This registry powers extensible frameworks, CLI tools, and modular applications.

Conclusion

Most developers treat architecture as something to learn later, which leads to midnight rewrites. By consciously applying these ten patterns, code evolves from "just works" to "maintainable and elegant".

Practical Advice

Start with a single small pattern (e.g., Dependency Injection or Strategy) before trying to adopt them all.

Refactor an old project piece by piece using the patterns you’ve learned.

During code reviews, discuss whether a design pattern is appropriate for the change.

Study the classic GoF 23 patterns to understand the underlying principles.