SOLID Principles in Python - Engineering Patterns for Maintainable Code

Reading time: ~32 minutes | Level: Intermediate → Engineering

Before reading further, identify every SOLID violation in this code:

import sqlite3
import smtplib
import json

class UserManager:
    def __init__(self):
        self.db = sqlite3.connect("users.db")

    def create_user(self, username, email, password):
        hashed = password[::-1]  # terrible "hashing"
        self.db.execute(
            "INSERT INTO users VALUES (?, ?, ?)", (username, email, hashed)
        )
        self.db.commit()
        # Send welcome email
        server = smtplib.SMTP("smtp.gmail.com", 587)
        server.sendmail("[email protected]", email, f"Welcome {username}!")
        server.quit()
        # Log to file
        with open("app.log", "a") as f:
            f.write(f"User created: {username}\n")
        return {"username": username, "email": email}

    def validate_email(self, email):
        return "@" in email

    def get_all_users(self):
        return self.db.execute("SELECT * FROM users").fetchall()

    def export_to_json(self, users):
        return json.dumps(users)

    def export_to_csv(self, users):
        return "\n".join(",".join(str(v) for v in row) for row in users)

This 40-line class violates all five SOLID principles. By the end of this lesson, you will be able to identify each violation precisely and refactor it to a clean design.

What You Will Learn

• Single Responsibility Principle (SRP): one class, one reason to change
• Open/Closed Principle (OCP): extend via typing.Protocol, not modification
• Liskov Substitution Principle (LSP): subclasses must be substitutable - violations and consequences
• Interface Segregation Principle (ISP): small, focused ABCs over fat interfaces
• Dependency Inversion Principle (DIP): inject abstractions, not concretions
• Python-specific implementation patterns for each principle
• Common violations in Python codebases

Prerequisites

• Lessons 01–10 of this module
• Understanding of composition, dependency injection, ABCs, and typing.Protocol
• Familiarity with Python's abc module

SOLID as a Decision Checklist

Before diving into each principle, here is how they interact as a design checklist:

Part 1 - Single Responsibility Principle

The Principle

A class should have one, and only one, reason to change.

The word "reason" means stakeholder or concern. If your class handles both data persistence and email notifications, it will need to change when the database schema changes AND when the email provider changes. Those are two different reasons.

SRP is not "a class should do one thing". A class can do many things as long as those things are all in service of one coherent responsibility. It is about cohesion and the source of change.

tip

SRP diagnostic: if you have to use "and" to describe what a class does, it probably needs to be split. "This class validates users AND sends emails AND logs to disk" - that is three responsibilities, three reasons to change. Each "and" is a split point.

Violation

The UserManager class from the puzzle has at least five reasons to change:

1. Database implementation changes (SQLite → PostgreSQL)
2. Password hashing algorithm changes
3. Email provider changes (Gmail → SendGrid)
4. Log format changes (file → structured JSON)
5. Export format requirements change (add XML)

# Every method in this class serves a different master
class UserManager:
    def create_user(...):  # orchestration + persistence + email + logging
    def validate_email(...):  # validation logic
    def get_all_users(...):  # data access
    def export_to_json(...):  # serialisation
    def export_to_csv(...):   # serialisation (different format)

Applying SRP

Decompose the class by reason-to-change. Each resulting component has one reason to change.

# 1. Password hashing - reason: crypto policy changes
class PasswordHasher:
    def hash(self, password: str) -> str:
        import hashlib
        return hashlib.sha256(password.encode()).hexdigest()

    def verify(self, password: str, hashed: str) -> bool:
        return self.hash(password) == hashed


# 2. User repository - reason: database/storage changes
class UserRepository:
    def __init__(self, db_path: str = "users.db"):
        import sqlite3
        self._db = sqlite3.connect(db_path)
        self._db.execute(
            "CREATE TABLE IF NOT EXISTS users (username TEXT, email TEXT, password_hash TEXT)"
        )

    def save(self, username: str, email: str, password_hash: str) -> None:
        self._db.execute(
            "INSERT INTO users VALUES (?, ?, ?)",
            (username, email, password_hash)
        )
        self._db.commit()

    def find_all(self) -> list:
        return self._db.execute("SELECT username, email FROM users").fetchall()


# 3. Email service - reason: email provider or template changes
class EmailService:
    def __init__(self, smtp_host: str, smtp_port: int):
        self._host = smtp_host
        self._port = smtp_port

    def send_welcome(self, to: str, username: str) -> None:
        import smtplib
        server = smtplib.SMTP(self._host, self._port)
        server.sendmail("[email protected]", to, f"Welcome {username}!")
        server.quit()


# 4. Logger - reason: logging destination or format changes
class Logger:
    def __init__(self, log_path: str = "app.log"):
        self._path = log_path

    def info(self, message: str) -> None:
        with open(self._path, "a") as f:
            f.write(f"INFO: {message}\n")


# 5. Input validation - reason: business validation rules change
class UserValidator:
    def validate_email(self, email: str) -> bool:
        return "@" in email and "." in email.split("@")[1]

    def validate_username(self, username: str) -> bool:
        return 3 <= len(username) <= 50 and username.isalnum()


# 6. Exporter - reason: export format requirements change
class UserExporter:
    def to_json(self, users: list) -> str:
        import json
        return json.dumps([{"username": u, "email": e} for u, e in users])

    def to_csv(self, users: list) -> str:
        header = "username,email"
        rows = "\n".join(f"{u},{e}" for u, e in users)
        return f"{header}\n{rows}"


# 7. Orchestration - reason: business workflow changes
class UserRegistrationService:
    def __init__(
        self,
        repo: UserRepository,
        hasher: PasswordHasher,
        email_svc: EmailService,
        validator: UserValidator,
        logger: Logger,
    ):
        self._repo = repo
        self._hasher = hasher
        self._email = email_svc
        self._validator = validator
        self._logger = logger

    def register(self, username: str, email: str, password: str) -> dict:
        if not self._validator.validate_username(username):
            raise ValueError(f"Invalid username: {username!r}")
        if not self._validator.validate_email(email):
            raise ValueError(f"Invalid email: {email!r}")
        hashed = self._hasher.hash(password)
        self._repo.save(username, email, hashed)
        self._email.send_welcome(to=email, username=username)
        self._logger.info(f"User registered: {username}")
        return {"username": username, "email": email}

Now each class has exactly one reason to change. The orchestration service (UserRegistrationService) is the only class that changes when the workflow changes.

SRP at the Module Level

SRP applies to modules too. Do not put database models, business logic, API handlers, and utility functions in a single file. Organise by cohesion.

myapp/
├── models/
│   └── user.py          # data models
├── repositories/
│   └── user_repo.py     # data access
├── services/
│   └── user_service.py  # business logic
├── api/
│   └── user_handler.py  # HTTP handlers
└── utils/
    └── validators.py    # shared utilities

Part 2 - Open/Closed Principle

The Principle

Software entities should be open for extension but closed for modification.

Adding new behaviour should not require modifying existing, working code. The classic violation is a large if/elif/isinstance chain that grows every time a new variant is added.

warning

OCP violation signal: every time you add a new if/elif branch to handle a new type or format, you are modifying existing code and risking regressions. Each new format in a ReportExporter.export(format="xml") call requires touching the same function. The fix is to extract the varying behaviour into a Protocol and let new implementations extend the system without touching the existing code.

Violation

class ReportExporter:
    def export(self, data: list, format: str) -> str:
        if format == "json":
            import json
            return json.dumps(data)
        elif format == "csv":
            return "\n".join(",".join(str(v) for v in row) for row in data)
        elif format == "xml":
            # Added later - required modifying this class
            rows = "\n".join(
                f"  <row>{''.join(f'<v>{v}</v>' for v in row)}</row>"
                for row in data
            )
            return f"<data>\n{rows}\n</data>"
        else:
            raise ValueError(f"Unknown format: {format!r}")
        # Every new format = modify this class = risk breaking existing formats

Applying OCP with typing.Protocol

from typing import Protocol

class ExportFormatter(Protocol):
    def format(self, data: list) -> str: ...

class JsonFormatter:
    def format(self, data: list) -> str:
        import json
        return json.dumps(data)

class CsvFormatter:
    def format(self, data: list) -> str:
        return "\n".join(",".join(str(v) for v in row) for row in data)

class XmlFormatter:
    def format(self, data: list) -> str:
        rows = "\n".join(
            f"  <row>{''.join(f'<v>{v}</v>' for v in row)}</row>"
            for row in data
        )
        return f"<data>\n{rows}\n</data>"

class ReportExporter:
    """Open for extension (new formatters), closed for modification."""

    def __init__(self, formatter: ExportFormatter):
        self._formatter = formatter

    def export(self, data: list) -> str:
        return self._formatter.format(data)

# Adding YAML support: create YamlFormatter - zero changes to ReportExporter
class YamlFormatter:
    def format(self, data: list) -> str:
        import yaml
        return yaml.dump(data)

# Usage
exporter = ReportExporter(formatter=JsonFormatter())
print(exporter.export([[1, "Alice"], [2, "Bob"]]))

# Switch to CSV without changing ReportExporter
exporter = ReportExporter(formatter=CsvFormatter())
print(exporter.export([[1, "Alice"], [2, "Bob"]]))

OCP with ABCs for Required Hooks

When you want to enforce that extensions implement specific methods, use an ABC:

from abc import ABC, abstractmethod

class NotificationChannel(ABC):
    """Abstract channel - open for extension via subclassing."""

    @abstractmethod
    def send(self, recipient: str, message: str) -> bool:
        """Send a notification. Returns True if successful."""
        ...

    def send_batch(self, recipients: list[str], message: str) -> dict[str, bool]:
        """Default batch implementation - overridable for efficiency."""
        return {r: self.send(r, message) for r in recipients}

class EmailChannel(NotificationChannel):
    def send(self, recipient: str, message: str) -> bool:
        print(f"Email → {recipient}: {message}")
        return True

class SlackChannel(NotificationChannel):
    def __init__(self, webhook_url: str):
        self._webhook = webhook_url

    def send(self, recipient: str, message: str) -> bool:
        print(f"Slack [{self._webhook}] → {recipient}: {message}")
        return True

class NotificationService:
    """Closed for modification - supports any NotificationChannel."""

    def __init__(self, channels: list[NotificationChannel]):
        self._channels = channels

    def notify(self, recipient: str, message: str) -> None:
        for channel in self._channels:
            channel.send(recipient, message)

# Add SMS without touching NotificationService
class SMSChannel(NotificationChannel):
    def send(self, recipient: str, message: str) -> bool:
        print(f"SMS → {recipient}: {message}")
        return True

service = NotificationService(channels=[EmailChannel(), SlackChannel("https://..."), SMSChannel()])
service.notify("[email protected]", "Deploy complete")

Part 3 - Liskov Substitution Principle

The Principle

Objects of a superclass should be replaceable with objects of a subclass without altering the correctness of the program.

If you have code that works with Base, substituting Derived for Base must not break it. The subclass must honour the contract of the parent - same preconditions, same postconditions, same invariants.

danger

LSP violation: if a subclass raises NotImplementedError for a method defined in the parent, it is NOT a true subtype. Code that calls the parent method trusting its contract will crash when given the subclass. The fix is always to redesign the hierarchy - either remove the method from the parent, split into separate interfaces, or use composition. Never paper over an LSP violation with try/except.

Violation 1 - Raising NotImplementedError in Overrides

class Bird:
    def fly(self) -> str:
        return "flying"

class Penguin(Bird):
    def fly(self) -> str:
        raise NotImplementedError("Penguins cannot fly!")

def make_bird_fly(bird: Bird) -> str:
    return bird.fly()   # works for Bird, breaks for Penguin

make_bird_fly(Bird())    # "flying"
make_bird_fly(Penguin()) # NotImplementedError - LSP violated

The fix: Penguin should not inherit from Bird if it cannot honour the fly() contract. Redesign the hierarchy or use composition.

# Fix: separate the flyable capability
class Bird:
    def breathe(self): ...
    def eat(self): ...

class FlyingBird(Bird):
    def fly(self) -> str:
        return "flying"

class Penguin(Bird):
    def swim(self) -> str:
        return "swimming"

class Eagle(FlyingBird):
    pass

def make_fly(bird: FlyingBird) -> str:
    return bird.fly()   # only accepts birds that can fly

make_fly(Eagle())    # "flying"
# make_fly(Penguin()) # type error - caught by type checker

Violation 2 - Strengthening Preconditions

class DataProcessor:
    def process(self, data: list) -> list:
        """Process any list, including empty lists."""
        return [item * 2 for item in data]

class StrictProcessor(DataProcessor):
    def process(self, data: list) -> list:
        if not data:
            raise ValueError("data must not be empty")  # Stronger precondition - LSP violated
        return [item * 2 for item in data]

# Code written for DataProcessor:
def process_dataset(processor: DataProcessor, data: list) -> list:
    return processor.process(data)  # may be called with empty list

process_dataset(DataProcessor(), [])        # []  - works
process_dataset(StrictProcessor(), [])      # ValueError - contract broken

A subclass must accept at least the same inputs as the parent. It can accept more, never less.

Violation 3 - Weakening Postconditions

from typing import Optional

class Cache:
    def get(self, key: str) -> Optional[str]:
        """Returns the cached value, or None if not found. Never raises."""
        return self._store.get(key)

class BrokenCache(Cache):
    def get(self, key: str) -> Optional[str]:
        # Raises KeyError instead of returning None - weakens postcondition
        return self._store[key]   # KeyError if missing - LSP violated

def fetch_with_default(cache: Cache, key: str, default: str) -> str:
    value = cache.get(key)   # caller trusts return-or-None contract
    return value if value is not None else default

fetch_with_default(Cache(), "missing", "default")        # "default"
fetch_with_default(BrokenCache(), "missing", "default")  # KeyError - contract broken

The Practical LSP Check

For every public method in the parent class, ask:

1. Can the subclass be called with the same arguments? (preconditions not strengthened)
2. Does the subclass return the same types and value ranges? (postconditions not weakened)
3. Does the subclass preserve all invariants the parent guarantees?

# Good LSP-compliant hierarchy
class NumberList:
    def __init__(self, values: list[float]):
        if not all(isinstance(v, (int, float)) for v in values):
            raise TypeError("All values must be numbers")
        self._values = list(values)

    def sum(self) -> float:
        return sum(self._values)

    def average(self) -> float:
        if not self._values:
            return 0.0
        return self.sum() / len(self._values)

class NonEmptyNumberList(NumberList):
    def __init__(self, values: list[float]):
        if not values:
            raise ValueError("List cannot be empty")
        super().__init__(values)

    def average(self) -> float:
        # Safe - list is guaranteed non-empty by __init__
        return self.sum() / len(self._values)

# NonEmptyNumberList IS-A NumberList - but with stronger construction precondition.
# This is technically an LSP grey area: constructor preconditions are often acceptably
# strengthened. The key is that once constructed, it honours all postconditions.
# Code accepting a NumberList that was constructed already can use NonEmptyNumberList safely.

Part 4 - Interface Segregation Principle

The Principle

Clients should not be forced to depend on interfaces they do not use.

Fat interfaces force implementing classes to provide stub implementations of methods they do not need, creating coupling to irrelevant behaviour.

Violation

from abc import ABC, abstractmethod

class WorkerInterface(ABC):
    @abstractmethod
    def work(self) -> None: ...

    @abstractmethod
    def eat(self) -> None: ...

    @abstractmethod
    def sleep(self) -> None: ...

class RobotWorker(WorkerInterface):
    def work(self) -> None:
        print("Robot working")

    def eat(self) -> None:
        raise NotImplementedError("Robots don't eat")  # forced stub - ISP violated

    def sleep(self) -> None:
        raise NotImplementedError("Robots don't sleep")  # forced stub - ISP violated

RobotWorker is forced to implement eat() and sleep() just to satisfy the interface, even though those methods make no sense for it.

Applying ISP: Small, Focused Interfaces

from abc import ABC, abstractmethod

class Workable(ABC):
    @abstractmethod
    def work(self) -> None: ...

class Eatable(ABC):
    @abstractmethod
    def eat(self) -> None: ...

class Sleepable(ABC):
    @abstractmethod
    def sleep(self) -> None: ...

class HumanWorker(Workable, Eatable, Sleepable):
    def work(self) -> None: print("Human working")
    def eat(self) -> None: print("Human eating")
    def sleep(self) -> None: print("Human sleeping")

class RobotWorker(Workable):
    def work(self) -> None: print("Robot working")
    # No eat() or sleep() - not forced to implement them

# Functions depend only on what they need
def schedule_work(worker: Workable) -> None:
    worker.work()

def schedule_lunch(eater: Eatable) -> None:
    eater.eat()

schedule_work(HumanWorker())   # works
schedule_work(RobotWorker())   # works
schedule_lunch(HumanWorker())  # works
# schedule_lunch(RobotWorker())  # type error - RobotWorker is not Eatable

ISP with typing.Protocol

from typing import Protocol

class Readable(Protocol):
    def read(self, size: int = -1) -> bytes: ...

class Writable(Protocol):
    def write(self, data: bytes) -> int: ...

class Seekable(Protocol):
    def seek(self, offset: int) -> int: ...
    def tell(self) -> int: ...

class ReadWriteStream(Readable, Writable, Protocol):
    pass

# Functions declare only the capabilities they need
def copy_data(source: Readable, dest: Writable) -> None:
    while chunk := source.read(4096):
        dest.write(chunk)

def get_position(stream: Seekable) -> int:
    return stream.tell()

# Standard library file objects satisfy all three without declaring them
import io
buf = io.BytesIO(b"hello")
copy_data(source=buf, dest=io.BytesIO())  # works - BytesIO satisfies Readable and Writable

note

Python's typing.Protocol enables structural subtyping: no inheritance required for interface compliance. A class satisfies a Protocol simply by having the matching methods and signatures - no implements keyword, no explicit registration. This is sometimes called "duck typing with static checking." It is the most Pythonic way to express ISP: define small Protocols for each capability, and any class that has those methods automatically satisfies the interface.

Production Example: Storage ISP

from typing import Protocol

class BlobReader(Protocol):
    def read_blob(self, blob_id: str) -> bytes: ...

class BlobWriter(Protocol):
    def write_blob(self, blob_id: str, data: bytes) -> str: ...

class BlobDeleter(Protocol):
    def delete_blob(self, blob_id: str) -> None: ...

class BlobMetadata(Protocol):
    def blob_exists(self, blob_id: str) -> bool: ...
    def blob_size(self, blob_id: str) -> int: ...

# Full storage implements all
class S3Storage:
    def read_blob(self, blob_id: str) -> bytes: ...
    def write_blob(self, blob_id: str, data: bytes) -> str: ...
    def delete_blob(self, blob_id: str) -> None: ...
    def blob_exists(self, blob_id: str) -> bool: ...
    def blob_size(self, blob_id: str) -> int: ...

# Read-only cache only needs BlobReader + BlobMetadata
class ReadThroughCache:
    def __init__(self, storage: BlobReader, metadata: BlobMetadata):
        self._storage = storage
        self._meta = metadata
        self._cache: dict = {}

    def get(self, blob_id: str) -> bytes:
        if blob_id not in self._cache:
            if not self._meta.blob_exists(blob_id):
                raise KeyError(blob_id)
            self._cache[blob_id] = self._storage.read_blob(blob_id)
        return self._cache[blob_id]

# ReadThroughCache does NOT depend on write or delete capabilities
# It is protected from changes to those interfaces

Part 5 - Dependency Inversion Principle

The Principle

High-level modules should not depend on low-level modules. Both should depend on abstractions. Abstractions should not depend on details. Details should depend on abstractions.

The key inversion: instead of high-level business logic creating and depending on concrete implementations (SQLite, SMTP, Redis), both sides depend on an abstraction (a Protocol or ABC). The concrete implementations are injected.

DIP Class Diagram

The diagram shows the key DIP insight: OrderProcessor (high-level) points only to Protocols (abstractions). The concrete classes (SQLiteOrderRepository, EmailNotifier) implement the protocols without OrderProcessor ever knowing about them.

Violation

class OrderProcessor:
    def __init__(self):
        import sqlite3
        # High-level module creates and depends on low-level module directly
        self._db = sqlite3.connect("orders.db")
        import smtplib
        self._smtp = smtplib.SMTP("smtp.gmail.com", 587)

    def process_order(self, order: dict) -> None:
        # Business logic mixed with infrastructure details
        self._db.execute("INSERT INTO orders VALUES (?)", (order["id"],))
        self._db.commit()
        self._smtp.sendmail("[email protected]", order["email"], "Order confirmed!")

Problems: OrderProcessor is untestable without a real database and SMTP server. Changing from SQLite to PostgreSQL requires modifying OrderProcessor. These are two separate concerns forced together.

Applying DIP

from typing import Protocol

# Abstractions - both sides depend on these
class OrderRepository(Protocol):
    def save(self, order: dict) -> None: ...
    def find(self, order_id: str) -> dict | None: ...

class Notifier(Protocol):
    def notify(self, recipient: str, message: str) -> None: ...

# Low-level details implement the abstractions
class SQLiteOrderRepository:
    def __init__(self, db_path: str):
        import sqlite3
        self._db = sqlite3.connect(db_path)

    def save(self, order: dict) -> None:
        self._db.execute("INSERT INTO orders VALUES (?)", (order["id"],))
        self._db.commit()

    def find(self, order_id: str) -> dict | None:
        row = self._db.execute(
            "SELECT * FROM orders WHERE id = ?", (order_id,)
        ).fetchone()
        return {"id": row[0]} if row else None

class EmailNotifier:
    def notify(self, recipient: str, message: str) -> None:
        import smtplib
        server = smtplib.SMTP("smtp.gmail.com", 587)
        server.sendmail("[email protected]", recipient, message)
        server.quit()

# High-level module depends ONLY on abstractions
class OrderProcessor:
    def __init__(self, repo: OrderRepository, notifier: Notifier):
        # Both dependencies are injected - DIP achieved
        self._repo = repo
        self._notifier = notifier

    def process_order(self, order: dict) -> None:
        # Pure business logic - no infrastructure details
        self._repo.save(order)
        self._notifier.notify(order["email"], f"Order {order['id']} confirmed!")

# Production wiring
processor = OrderProcessor(
    repo=SQLiteOrderRepository("orders.db"),
    notifier=EmailNotifier(),
)

# Test wiring - no database, no SMTP server
class InMemoryOrderRepository:
    def __init__(self):
        self.saved: list = []

    def save(self, order: dict) -> None:
        self.saved.append(order)

    def find(self, order_id: str) -> dict | None:
        return next((o for o in self.saved if o["id"] == order_id), None)

class RecordingNotifier:
    def __init__(self):
        self.notifications: list = []

    def notify(self, recipient: str, message: str) -> None:
        self.notifications.append({"recipient": recipient, "message": message})

# Test
repo = InMemoryOrderRepository()
notifier = RecordingNotifier()
processor = OrderProcessor(repo=repo, notifier=notifier)
processor.process_order({"id": "ORD-001", "email": "[email protected]"})

assert len(repo.saved) == 1
assert repo.saved[0]["id"] == "ORD-001"
assert len(notifier.notifications) == 1
assert notifier.notifications[0]["recipient"] == "[email protected]"
print("All assertions passed")

DIP in a Layered Architecture

from typing import Protocol

# === Domain Layer (innermost - no external dependencies) ===

class User:
    def __init__(self, user_id: str, username: str, email: str):
        self.user_id = user_id
        self.username = username
        self.email = email


# === Application Layer (depends on abstractions only) ===

class UserRepository(Protocol):
    def find_by_id(self, user_id: str) -> User | None: ...
    def save(self, user: User) -> None: ...

class EventPublisher(Protocol):
    def publish(self, event_type: str, payload: dict) -> None: ...

class UserService:
    """High-level business logic - depends only on abstractions."""

    def __init__(self, repo: UserRepository, events: EventPublisher):
        self._repo = repo
        self._events = events

    def deactivate_user(self, user_id: str) -> None:
        user = self._repo.find_by_id(user_id)
        if user is None:
            raise ValueError(f"User not found: {user_id}")
        # Business rule: cannot deactivate admin users
        if user.username.startswith("admin_"):
            raise PermissionError(f"Cannot deactivate admin user: {user.username}")
        self._repo.save(user)
        self._events.publish("user.deactivated", {"user_id": user_id})


# === Infrastructure Layer (implements abstractions) ===

class PostgresUserRepository:
    def find_by_id(self, user_id: str) -> User | None:
        # SELECT from PostgreSQL
        ...

    def save(self, user: User) -> None:
        # INSERT/UPDATE in PostgreSQL
        ...

class KafkaEventPublisher:
    def __init__(self, broker: str):
        self._broker = broker

    def publish(self, event_type: str, payload: dict) -> None:
        import json
        print(f"Kafka [{self._broker}] → {event_type}: {json.dumps(payload)}")

The domain layer knows nothing about databases or message brokers. The application layer knows nothing about PostgreSQL or Kafka. The infrastructure layer implements the interfaces that the application layer defined. This is the Dependency Rule from Clean Architecture.

Part 6 - The Complete Refactoring

Returning to the opening example - here it is fully refactored against all five SOLID principles:

from typing import Protocol
from dataclasses import dataclass
from abc import ABC, abstractmethod

# SRP: each class has one responsibility
# ISP: small, focused interfaces
# OCP: open for extension via Protocol
# DIP: high-level depends on abstractions

class PasswordHasher(Protocol):           # ISP: just hashing
    def hash(self, password: str) -> str: ...

class UserStore(Protocol):                # ISP: just storage
    def save(self, username: str, email: str, password_hash: str) -> None: ...
    def list_all(self) -> list[dict]: ...

class WelcomeMailer(Protocol):            # ISP: just welcome emails
    def send_welcome(self, to: str, username: str) -> None: ...

class AuditLogger(Protocol):              # ISP: just logging
    def log_registration(self, username: str) -> None: ...

class UserExporter(Protocol):             # ISP: just export
    def export(self, users: list[dict]) -> str: ...


# SRP: one class, one reason to change
class SHA256Hasher:
    def hash(self, password: str) -> str:
        import hashlib
        return hashlib.sha256(password.encode()).hexdigest()


class InMemoryUserStore:
    def __init__(self):
        self._users: list = []

    def save(self, username: str, email: str, password_hash: str) -> None:
        self._users.append({"username": username, "email": email, "hash": password_hash})

    def list_all(self) -> list[dict]:
        return [{"username": u["username"], "email": u["email"]} for u in self._users]


class PrintWelcomeMailer:
    def send_welcome(self, to: str, username: str) -> None:
        print(f"MAIL → {to}: Welcome {username}!")


class FileAuditLogger:
    def __init__(self, path: str = "audit.log"):
        self._path = path

    def log_registration(self, username: str) -> None:
        with open(self._path, "a") as f:
            f.write(f"REGISTERED: {username}\n")


class JsonUserExporter:                    # OCP: add new format = new class
    def export(self, users: list[dict]) -> str:
        import json
        return json.dumps(users)

class CsvUserExporter:
    def export(self, users: list[dict]) -> str:
        lines = ["username,email"]
        lines.extend(f"{u['username']},{u['email']}" for u in users)
        return "\n".join(lines)


# DIP: high-level orchestration depends only on abstractions
class UserRegistrationService:
    def __init__(
        self,
        store: UserStore,
        hasher: PasswordHasher,
        mailer: WelcomeMailer,
        logger: AuditLogger,
    ):
        self._store = store
        self._hasher = hasher
        self._mailer = mailer
        self._logger = logger

    def register(self, username: str, email: str, password: str) -> dict:
        if "@" not in email:
            raise ValueError(f"Invalid email: {email!r}")
        hash_value = self._hasher.hash(password)
        self._store.save(username, email, hash_value)
        self._mailer.send_welcome(to=email, username=username)
        self._logger.log_registration(username)
        return {"username": username, "email": email}


# Wire it up
service = UserRegistrationService(
    store=InMemoryUserStore(),
    hasher=SHA256Hasher(),
    mailer=PrintWelcomeMailer(),
    logger=FileAuditLogger(),
)
user = service.register("alice", "[email protected]", "password123")
print(user)   # {'username': 'alice', 'email': '[email protected]'}

Common Mistakes

Mistake 1 - Confusing SRP with "One Method Per Class"

SRP is about one reason to change, not one method. A UserValidator class can have validate_email, validate_username, and validate_password - all cohesive, all in service of the same responsibility.

Mistake 2 - Creating Proxy Classes for OCP That Add No Value

# Over-engineered OCP: abstraction with no realistic variation
class StringUppercaser(Protocol):
    def uppercase(self, s: str) -> str: ...

class DefaultUppercaser:
    def uppercase(self, s: str) -> str: return s.upper()

# Just use str.upper() directly - not everything needs to be a Protocol

OCP is about anticipated points of variation. Do not add abstractions unless you have a realistic reason to expect variation.

Mistake 3 - LSP Violation Hidden by try/except

class SafeCache(Cache):
    def get(self, key: str) -> str | None:
        try:
            return self._store[key]   # raises KeyError - LSP violation hidden
        except KeyError:
            return None   # caller cannot tell this "fixed" the violation

# The fix should be at the method level, not wrapped in try/except

Mistake 4 - DIP Without an Interface (Partial DIP)

# Not DIP: high-level depends on concrete, even though it's injected
class Service:
    def __init__(self, db: SQLiteDatabase):  # injected but still concrete
        self._db = db

# True DIP: depends on abstraction
class Service:
    def __init__(self, db: DatabaseProtocol):  # abstraction
        self._db = db

Injection alone is not DIP. The dependency must be inverted - the high-level module must define the abstraction, not depend on the concrete type.

Engineering Checklist

Before moving to the next lesson, verify you can answer these without looking:

1. What does "one reason to change" mean in SRP? How is it different from "one method"?
2. What is the canonical OCP violation pattern, and how do you fix it?
3. State the three Liskov substitution rules: preconditions, postconditions, invariants.
4. What is the penalty for an ISP violation? What happens to implementing classes?
5. How does DIP differ from simple dependency injection?
6. In a layered architecture applying DIP, which layer defines the interfaces?
7. Given class Service(Protocol) and class ServiceImpl(ABC) - which is more suited to external dependencies and which to internal hierarchies?
8. Why does raise NotImplementedError in an overridden method violate LSP?
9. How can you verify LSP compliance at the type checker level?

Graded Practice Challenges

Level 1 - Predict the Output / Identify the Violation

For each snippet, identify which SOLID principle is violated and why.

Question 1

class ReportService:
    def generate(self, data):
        ...

    def save_to_db(self, report):
        ...

    def email_report(self, report, recipient):
        ...

    def export_to_pdf(self, report):
        ...

Show Answer

SRP violation. ReportService has at least four reasons to change: report generation logic, database schema, email provider, and PDF library. Each method serves a different stakeholder.

Fix: split into ReportGenerator, ReportRepository, ReportMailer, and ReportExporter.

Question 2

class Animal:
    def speak(self) -> str:
        return "..."

class Dog(Animal):
    def speak(self) -> str:
        return "Woof"

class Fish(Animal):
    def speak(self) -> str:
        raise NotImplementedError("Fish cannot speak")

def make_noise(animal: Animal) -> str:
    return animal.speak()

Show Answer

LSP violation. Fish is a subtype of Animal but raises NotImplementedError for speak(). Code that calls make_noise(animal) trusting the Animal contract will crash when passed a Fish.

Fix: remove speak() from Animal, create a Speakable interface, and only let Dog (and similar animals) implement it.

Question 3

from abc import ABC, abstractmethod

class DataStore(ABC):
    @abstractmethod
    def read(self, key: str) -> bytes: ...

    @abstractmethod
    def write(self, key: str, data: bytes) -> None: ...

    @abstractmethod
    def delete(self, key: str) -> None: ...

    @abstractmethod
    def list_keys(self) -> list[str]: ...

class ReadOnlyArchive(DataStore):
    def read(self, key: str) -> bytes:
        return b"archived data"

    def write(self, key: str, data: bytes) -> None:
        raise NotImplementedError("Archive is read-only")

    def delete(self, key: str) -> None:
        raise NotImplementedError("Archive is read-only")

    def list_keys(self) -> list[str]:
        return ["key1", "key2"]

Show Answer

ISP violation (and a secondary LSP violation). DataStore is a fat interface - ReadOnlyArchive is forced to stub out write and delete with NotImplementedError.

Fix: split into Readable, Writable, Deletable, and Listable protocols. ReadOnlyArchive only implements Readable and Listable.

Level 2 - Debug Challenge

This code has an OCP and DIP violation. Identify both and refactor to fix them:

import sqlite3
import json

class DataExporter:
    def __init__(self):
        # DIP violation: creates its own dependency
        self._db = sqlite3.connect("app.db")

    def export(self, table: str, format: str) -> str:
        rows = self._db.execute(f"SELECT * FROM {table}").fetchall()
        # OCP violation: adding new format = modifying this method
        if format == "json":
            return json.dumps(rows)
        elif format == "csv":
            return "\n".join(",".join(str(v) for v in row) for row in rows)
        else:
            raise ValueError(f"Unknown format: {format!r}")

Show Solution

DIP violation: DataExporter creates sqlite3.connect(...) directly in __init__. It is tightly coupled to SQLite and cannot be tested without a real database.

OCP violation: Every new export format requires modifying the export method, risking regressions in existing format handling.

Refactored solution:

from typing import Protocol

# Abstraction for data source (DIP)
class DataSource(Protocol):
    def fetch(self, query_key: str) -> list: ...

# Abstraction for formatters (OCP)
class Formatter(Protocol):
    def format(self, rows: list) -> str: ...

# Concrete data source
class SQLiteDataSource:
    def __init__(self, db_path: str):
        import sqlite3
        self._db = sqlite3.connect(db_path)

    def fetch(self, query_key: str) -> list:
        return self._db.execute(f"SELECT * FROM {query_key}").fetchall()

# In-memory source for testing
class InMemoryDataSource:
    def __init__(self, data: dict[str, list]):
        self._data = data

    def fetch(self, query_key: str) -> list:
        return self._data.get(query_key, [])

# Concrete formatters - add new ones without touching DataExporter
class JsonFormatter:
    def format(self, rows: list) -> str:
        import json
        return json.dumps(rows)

class CsvFormatter:
    def format(self, rows: list) -> str:
        return "\n".join(",".join(str(v) for v in row) for row in rows)

# New format: no changes to DataExporter
class TsvFormatter:
    def format(self, rows: list) -> str:
        return "\n".join("\t".join(str(v) for v in row) for row in rows)

# DIP: depends only on abstractions; OCP: add formatter without modification
class DataExporter:
    def __init__(self, source: DataSource, formatter: Formatter):
        self._source = source
        self._formatter = formatter

    def export(self, query_key: str) -> str:
        rows = self._source.fetch(query_key)
        return self._formatter.format(rows)


# Production
exporter = DataExporter(
    source=SQLiteDataSource("app.db"),
    formatter=JsonFormatter(),
)

# Test - no database, no format coupling
test_data = {"users": [(1, "Alice"), (2, "Bob")]}
test_exporter = DataExporter(
    source=InMemoryDataSource(test_data),
    formatter=CsvFormatter(),
)
print(test_exporter.export("users"))
# 1,Alice
# 2,Bob

Level 3 - Design Challenge

Design a payment processing system that follows all five SOLID principles. Requirements:

1. Support multiple payment methods: CreditCard, PayPal, BankTransfer
2. Support multiple notification channels after payment: email and SMS
3. Support multiple audit log backends: file and database
4. The payment orchestrator (PaymentService) must be testable without real payment gateways, notification providers, or log backends
5. Adding a new payment method should require zero changes to PaymentService

Show Reference Solution

from typing import Protocol
from dataclasses import dataclass

# === Abstractions (ISP: small and focused) ===

@dataclass
class PaymentRequest:
    amount: float
    currency: str
    customer_id: str
    reference: str

@dataclass
class PaymentResult:
    success: bool
    transaction_id: str
    message: str

class PaymentGateway(Protocol):          # ISP: only payment
    def charge(self, request: PaymentRequest) -> PaymentResult: ...

class CustomerNotifier(Protocol):        # ISP: only notification
    def notify(self, customer_id: str, message: str) -> None: ...

class PaymentAuditor(Protocol):          # ISP: only audit
    def record(self, request: PaymentRequest, result: PaymentResult) -> None: ...


# === Concrete Implementations (OCP: extend without modifying PaymentService) ===

class CreditCardGateway:
    def charge(self, request: PaymentRequest) -> PaymentResult:
        print(f"Credit card: charging {request.amount} {request.currency}")
        return PaymentResult(success=True, transaction_id="CC-001", message="Approved")

class PayPalGateway:
    def charge(self, request: PaymentRequest) -> PaymentResult:
        print(f"PayPal: charging {request.amount} {request.currency}")
        return PaymentResult(success=True, transaction_id="PP-001", message="Completed")

class BankTransferGateway:
    def charge(self, request: PaymentRequest) -> PaymentResult:
        print(f"Bank transfer: initiating {request.amount} {request.currency}")
        return PaymentResult(success=True, transaction_id="BT-001", message="Initiated")

class EmailNotifier:
    def notify(self, customer_id: str, message: str) -> None:
        print(f"EMAIL to {customer_id}: {message}")

class SMSNotifier:
    def notify(self, customer_id: str, message: str) -> None:
        print(f"SMS to {customer_id}: {message}")

class FileAuditor:
    def record(self, request: PaymentRequest, result: PaymentResult) -> None:
        print(f"FILE LOG: {request.reference} → {result.transaction_id} ({result.success})")

class DatabaseAuditor:
    def record(self, request: PaymentRequest, result: PaymentResult) -> None:
        print(f"DB INSERT: {request.reference} → {result.transaction_id}")


# === High-Level Orchestration (DIP: depends only on abstractions) ===

class PaymentService:
    """SRP: only payment orchestration. DIP: only Protocols. OCP: extend via injection."""

    def __init__(
        self,
        gateway: PaymentGateway,
        notifiers: list[CustomerNotifier],
        auditor: PaymentAuditor,
    ):
        self._gateway = gateway
        self._notifiers = notifiers
        self._auditor = auditor

    def process(self, request: PaymentRequest) -> PaymentResult:
        result = self._gateway.charge(request)
        self._auditor.record(request, result)
        message = (
            f"Payment of {request.amount} {request.currency} succeeded (ref: {request.reference})"
            if result.success
            else f"Payment failed: {result.message}"
        )
        for notifier in self._notifiers:
            notifier.notify(request.customer_id, message)
        return result


# === Production Wiring ===
service = PaymentService(
    gateway=CreditCardGateway(),
    notifiers=[EmailNotifier(), SMSNotifier()],
    auditor=FileAuditor(),
)
req = PaymentRequest(amount=99.99, currency="USD", customer_id="cust-42", reference="ORD-2024-001")
result = service.process(req)
print(result)

# === Test Wiring - no real gateways, notifiers, or file I/O ===

class MockGateway:
    def __init__(self, success: bool = True):
        self._success = success
        self.charged: list = []

    def charge(self, request: PaymentRequest) -> PaymentResult:
        self.charged.append(request)
        return PaymentResult(success=self._success, transaction_id="MOCK-001", message="Mock")

class RecordingNotifier:
    def __init__(self):
        self.sent: list = []

    def notify(self, customer_id: str, message: str) -> None:
        self.sent.append({"customer": customer_id, "msg": message})

class NullAuditor:
    def record(self, request: PaymentRequest, result: PaymentResult) -> None:
        pass  # no-op for tests

gateway = MockGateway(success=True)
notifier = RecordingNotifier()
test_service = PaymentService(gateway=gateway, notifiers=[notifier], auditor=NullAuditor())
test_req = PaymentRequest(amount=10.0, currency="USD", customer_id="test-1", reference="T-001")
test_service.process(test_req)

assert len(gateway.charged) == 1
assert gateway.charged[0].reference == "T-001"
assert len(notifier.sent) == 1
assert "succeeded" in notifier.sent[0]["msg"]
print("All test assertions passed")

Key Takeaways

• SRP: each class should have one, and only one, reason to change. If you use "and" to describe what a class does, it likely needs to be split.
• OCP: adding new behaviour should not require modifying existing code. The signal for an OCP violation is an if/elif chain that grows with every new type or format.
• LSP: subclasses must honour the parent's contract - same accepted inputs, same return types, same invariants. A raise NotImplementedError in a subclass override is always an LSP violation.
• ISP: clients should only depend on the interface capabilities they use. typing.Protocol makes this natural in Python - define small, focused protocols for each capability.
• DIP: high-level modules depend on abstractions (Protocols/ABCs), not concrete implementations. Abstractions are defined by the high-level module, not the low-level one. Injection alone is not DIP - the dependency must also be an abstraction.
• Python's typing.Protocol enables structural subtyping: a class satisfies a Protocol simply by having the matching methods, with no explicit implements declaration. This is the most Pythonic way to implement ISP and DIP.

What's Next

Lesson 12 covers Design Patterns - the Gang of Four patterns reimplemented in idiomatic Python. Python's dynamic typing, first-class functions, and metaclass system make many classic patterns simpler or look fundamentally different from their Java/C++ originals. You will see Singleton, Factory, Strategy, Observer, Decorator, Registry, and Builder - each with a GoF intent, a Pythonic implementation, and a real framework usage example.