120. ‘Capstone: Everything Together’

🧰 Aryan’s complete Python toolkit — the final epilogue exercise. A ToolkitManager class combines all real-world patterns from the series. Two bugs remain: the log-level comparison uses a string instead of the integer constant, so debug logs never appear; and the get_stats() method recomputes from scratch every call instead of using the manual cache.

💡 Fun fact: The pattern used in ToolkitManager — a @dataclass for configuration, an Enum for typed operation constants, sqlite3 for persistence, and a manual dict cache — is the same architecture used in many production Python tools. The logging module’s NOTSET=0, DEBUG=10, INFO=20, WARNING=30, ERROR=40, CRITICAL=50 integer constants were chosen with 10-point gaps specifically so you can insert custom levels (like logging.addLevelName(25, "NOTICE")) between the standard ones. This is why comparing to the integer constant (logging.DEBUG == 10) is correct, but comparing to the string "DEBUG" always fails.

⚠️ Watch out: @lru_cache cannot be used on instance methods because self is part of the cache key — and Python objects are not hashable by default. If you try @functools.lru_cache on self.get_stats, you get TypeError: unhashable type. The solution is a plain dict (self._cache) with explicit key checking: if key in self._cache: return self._cache[key]. Remember to call self._cache.clear() after every mutation (add_item, remove_item) so stale cached values are never returned.

🤔 Think about it: The ToolkitManager clears its entire stats cache after every add_item() or remove_item() call. This is simple but potentially wasteful — if you add 100 items in a loop, you clear and rebuild the cache 100 times. What would a smarter cache invalidation strategy look like? Could you make the cache lazy enough that it only rebuilds once after a batch of mutations?

Learning objectives

• Avoid @lru_cache on instance methods — use a manual dict cache instead
• Compare logger.level to logging.DEBUG (integer), never to the string ‘DEBUG’
• Combine dataclass config, enum operations, sqlite3, and logging into one class
• Invalidate a manual cache after mutating operations (add, remove)
• Apply itertools.groupby with a pre-sorted list to aggregate data by category

Key concepts

• @dataclass — config and record types
• Enum — typed operation constants
• lru_cache limitation — requires hashable args, fails on self
• Manual dict cache — self._cache = {} with .clear() on mutations
• logging.DEBUG is 10 (int) — compare with == logging.DEBUG not == ‘DEBUG’

Try it

Concept detail

Final Capstone: Bringing It All Together

This exercise combines every major Python pattern from the series into one production-quality class.

@dataclass for Config and Records

@dataclass
class ToolkitConfig:
    db_path: str = ":memory:"
    log_level: str = "INFO"

config = ToolkitConfig(db_path="prod.db", log_level="WARNING")

Enum for Operation Types

class OperationType(Enum):
    ADD = "add"
    QUERY = "query"

# Compare member to member, .value to string
if op == OperationType.ADD:   # correct

Manual Dict Cache (instead of lru_cache on self)

# lru_cache on instance methods fails — self is not hashable
# Use a plain dict instead:
def __init__(self):
    self._cache: dict = {}

def get_stats(self):
    if "stats" in self._cache:
        return self._cache["stats"]
    result = self._compute()
    self._cache["stats"] = result
    return result

def add_item(self, ...):
    ...
    self._cache.clear()   # invalidate on mutation

Log Level Comparison

import logging
# logging.DEBUG is an int (10), not a string:
if logger.level == logging.DEBUG:   # correct
if logger.level == "DEBUG":         # always False!

itertools.groupby in Practice

from itertools import groupby
items = sorted(items, key=lambda x: x.category)  # sort first!
stats = {}
for cat, group in groupby(items, key=lambda x: x.category):
    stats[cat] = sum(item.quantity for item in group)

The Complete Pattern

ToolkitConfig (@dataclass) → ToolkitManager
  ├── sqlite3: storage with commit
  ├── logging: named logger, integer level compare
  ├── Enum: typed operation constants
  ├── itertools: sorted + groupby for stats
  ├── typing: Optional, List, Dict
  └── manual cache: dict instead of lru_cache on self

Solution

import sqlite3
import logging
import functools
from enum import Enum
from dataclasses import dataclass, field
from typing import List, Optional, Dict
from itertools import groupby

logger = logging.getLogger(__name__)

class OperationType(Enum):
    ADD = "add"
    REMOVE = "remove"
    QUERY = "query"
    STATS = "stats"

@dataclass
class ToolkitConfig:
    db_path: str = ":memory:"
    log_level: str = "INFO"
    cache_size: int = 128
    owner: str = "unknown"

@dataclass
class Item:
    item_id: int
    name: str
    category: str
    quantity: int

class ToolkitManager:
    """
    The complete toolkit. Combines:
    - @dataclass for config and records
    - Enum for operation types
    - sqlite3 for storage
    - logging for observability
    - manual dict cache for expensive stats
    - itertools for grouping
    - typing for type safety
    """

    def __init__(self, config: ToolkitConfig):
        self.config = config
        level = getattr(logging, config.log_level.upper(), logging.INFO)
        logging.basicConfig(level=level, format="%(levelname)s:%(name)s:%(message)s")
        logger.setLevel(level)

        self.conn = sqlite3.connect(config.db_path)
        self.conn.execute(
            "CREATE TABLE IF NOT EXISTS items "
            "(id INTEGER PRIMARY KEY AUTOINCREMENT, "
            "name TEXT NOT NULL, category TEXT NOT NULL, quantity INTEGER DEFAULT 0)"
        )
        self.conn.commit()
        self._stats_cache: Dict = {}
        self._log_operation(OperationType.ADD, "ToolkitManager initialized")

    def _log_operation(self, op: OperationType, detail: str):
        """Log an operation at appropriate level."""
        # FIX 1: Compare logger.level to the integer constant logging.DEBUG
        if logger.level == logging.DEBUG:
            logger.debug("[%s] %s", op.value, detail)
        else:
            logger.info("[%s] %s", op.value, detail)

    def add_item(self, name: str, category: str, quantity: int = 1) -> int:
        """Add an item to the toolkit. Returns new item id."""
        cursor = self.conn.execute(
            "INSERT INTO items (name, category, quantity) VALUES (?, ?, ?)",
            (name, category, quantity),
        )
        self.conn.commit()
        item_id = cursor.lastrowid
        self._log_operation(OperationType.ADD, f"Added {name} (id={item_id})")
        self._stats_cache.clear()
        return item_id

    def get_items(self, category: Optional[str] = None) -> List[Item]:
        """Return items, optionally filtered by category."""
        self._log_operation(OperationType.QUERY, f"get_items(category={category})")
        if category:
            cursor = self.conn.execute(
                "SELECT id, name, category, quantity FROM items WHERE category = ?",
                (category,),
            )
        else:
            cursor = self.conn.execute(
                "SELECT id, name, category, quantity FROM items"
            )
        return [Item(r[0], r[1], r[2], r[3]) for r in cursor.fetchall()]

    def remove_item(self, item_id: int) -> bool:
        """Remove an item by id. Returns True if removed."""
        cursor = self.conn.execute("DELETE FROM items WHERE id = ?", (item_id,))
        self.conn.commit()
        removed = cursor.rowcount > 0
        self._log_operation(OperationType.REMOVE, f"Removed id={item_id}: {removed}")
        self._stats_cache.clear()
        return removed

    def get_stats(self) -> Dict[str, int]:
        """Return item quantity totals grouped by category. Results are cached."""
        # FIX 2: Check cache first, return early if populated
        cache_key = "stats"
        if cache_key in self._stats_cache:
            return self._stats_cache[cache_key]
        self._log_operation(OperationType.STATS, "Computing stats")
        items = self.get_items()
        sorted_items = sorted(items, key=lambda x: x.category)
        stats = {}
        for category, group in groupby(sorted_items, key=lambda x: x.category):
            stats[category] = sum(item.quantity for item in group)
        self._stats_cache[cache_key] = stats
        return stats

    def close(self):
        self.conn.close()

Tests

DEFAULT_CONFIG = ToolkitConfig(db_path=":memory:", log_level="WARNING", owner="Aryan")

def test_init_creates_manager():
    tm = ToolkitManager(DEFAULT_CONFIG)
    assert tm.config.owner == "Aryan"
    tm.close()

def test_add_item_returns_id():
    tm = ToolkitManager(DEFAULT_CONFIG)
    item_id = tm.add_item("Hammer", "tools", 2)
    assert isinstance(item_id, int)
    assert item_id >= 1
    tm.close()

def test_get_items_returns_item_objects():
    tm = ToolkitManager(DEFAULT_CONFIG)
    tm.add_item("Notebook", "stationery", 5)
    items = tm.get_items()
    assert len(items) == 1
    assert isinstance(items[0], Item)
    assert items[0].name == "Notebook"
    assert items[0].quantity == 5
    tm.close()

def test_get_items_filter_by_category():
    tm = ToolkitManager(DEFAULT_CONFIG)
    tm.add_item("Pen", "stationery", 10)
    tm.add_item("Wrench", "tools", 1)
    stationery = tm.get_items(category="stationery")
    assert len(stationery) == 1
    assert stationery[0].name == "Pen"
    tm.close()

def test_remove_item():
    tm = ToolkitManager(DEFAULT_CONFIG)
    item_id = tm.add_item("Eraser", "stationery", 3)
    result = tm.remove_item(item_id)
    assert result is True
    assert len(tm.get_items()) == 0
    tm.close()

def test_remove_nonexistent_item():
    tm = ToolkitManager(DEFAULT_CONFIG)
    result = tm.remove_item(9999)
    assert result is False
    tm.close()

def test_get_stats_correct_values():
    """get_stats must group by category and sum quantities with itertools.groupby."""
    tm = ToolkitManager(DEFAULT_CONFIG)
    tm.add_item("Pen", "stationery", 5)
    tm.add_item("Ruler", "stationery", 2)
    tm.add_item("Hammer", "tools", 1)
    stats = tm.get_stats()
    assert stats.get("stationery") == 7, (
        f"Expected stationery=7 but got {stats.get('stationery')} — "
        "sum quantities within each category"
    )
    assert stats.get("tools") == 1
    tm.close()

def test_stats_cache_miss_then_hit():
    """get_stats should populate the cache on first call and use it on second."""
    tm = ToolkitManager(DEFAULT_CONFIG)
    tm.add_item("Pen", "stationery", 3)
    # First call — cache miss, computes result
    stats1 = tm.get_stats()
    assert stats1.get("stationery") == 3
    # Second call — cache hit, must return same dict object (not recomputed)
    stats2 = tm.get_stats()
    assert stats2 is stats1, (
        "get_stats should return the cached result on second call. "
        "Check the cache key lookup in get_stats()."
    )
    tm.close()

def test_stats_cache_invalidated_after_add():
    """Cache should be cleared after add_item so stats reflect new state."""
    tm = ToolkitManager(DEFAULT_CONFIG)
    tm.add_item("Pen", "stationery", 1)
    stats1 = tm.get_stats()
    tm.add_item("Pencil", "stationery", 3)
    stats2 = tm.get_stats()
    assert stats2.get("stationery") == 4, (
        f"Stats cache not cleared after add — expected 4 but got {stats2.get('stationery')}"
    )
    tm.close()

def test_log_level_comparison_is_integer():
    """logger.level is an integer — comparing to string 'DEBUG' is always False."""
    import logging
    assert logging.DEBUG == 10, "logging.DEBUG should be integer 10"
    assert isinstance(logging.DEBUG, int)
    # The fix compares logger.level (int) to logging.DEBUG (int), not to "DEBUG" (str)
    test_logger = logging.getLogger("test_level_check")
    test_logger.setLevel(logging.DEBUG)
    assert test_logger.level == logging.DEBUG
    assert test_logger.level != "DEBUG"

def test_operation_type_enum_values():
    """Verify OperationType enum values are correct strings."""
    assert OperationType.ADD.value == "add"
    assert OperationType.STATS.value == "stats"
    assert OperationType.REMOVE.value == "remove"
    assert OperationType.QUERY.value == "query"

Resources

• functools — Higher-order functions
• enum — Support for enumerations
• dataclasses — Data Classes
• logging — Logging facility for Python
• itertools — Functions creating iterators for efficient looping