117. ‘Capstone: Json + Sqlite3 + Logging’

🗂️ A complete task manager stores tasks in SQLite with JSON config and structured logging. The developer forgot to call commit() inside complete_task() so status updates are lost, and get_stats() counts all rows instead of grouping by status — returning wrong numbers.

💡 Fun fact: This capstone combines three modules that together power nearly every production Python application: json for configuration (most real-world services read their config from a JSON or YAML file at startup), sqlite3 for lightweight persistence (SQLite is the default database engine for Django, Flask, and many small-scale Python apps), and logging for observability (every production service emits structured logs consumed by tools like Elasticsearch, Datadog, or CloudWatch). The pattern of “read config → open DB → configure logging” is so common it has a name: the application bootstrap sequence.

⚠️ Watch out: In SQLite, every INSERT, UPDATE, or DELETE runs inside a transaction that is not committed until you explicitly call conn.commit(). If you call complete_task() and forget conn.commit(), the UPDATE is buffered — it’s visible to subsequent queries on the same connection (because they share the same transaction), but is rolled back and lost when the connection closes. This makes the bug hard to catch in tests that use a single connection — but it would silently lose data in any real application.

🤔 Think about it: get_stats() needs counts grouped by status. One approach is two separate SELECT COUNT(*) WHERE status = ? queries. Another is a single SELECT status, COUNT(*) FROM tasks GROUP BY status query that returns all counts at once. Which approach is better for this use case? When does a single GROUP BY query become clearly superior to multiple individual count queries?

Learning objectives

Parse configuration from JSON strings and apply settings to logging and database
Commit SQLite transactions after every write operation
Use WHERE clauses in COUNT queries to group by status
Combine json, sqlite3, and logging into a single class with clear responsibilities
Test stateful classes by calling methods in sequence and asserting final state

Key concepts

json.loads() — parse config from string
conn.commit() after every INSERT/UPDATE/DELETE
SELECT COUNT(*) WHERE status = ? — filtered count
logging.getLogger(name) — module-level named logger
cursor.lastrowid — ID of last INSERT

Try it

import json
import sqlite3
import logging
logger = logging.getLogger(name)
class TaskManager:
"""A task manager backed by SQLite with JSON-based configuration."""
<pre class="giallo" style="color-scheme: light dark; color: light-dark(#24292E, #E1E4E8); background-color: light-dark(#FFFFFF, #24292E);"><code data-lang="plain">def __init__(self, config_str: str):
 &quot;&quot;&quot;Initialize from a JSON config string.&quot;&quot;&quot;
 config = json.loads(config_str)
 db_path = config.get(&quot;db_path&quot;, &quot;:memory:&quot;)
 log_level_name = config.get(&quot;log_level&quot;, &quot;INFO&quot;)
 level = getattr(logging, log_level_name.upper(), logging.INFO)
 logging.basicConfig(level=level, format=&quot;%(levelname)s:%(name)s:%(message)s&quot;)
 logger.setLevel(level)

 self.conn = sqlite3.connect(db_path)
 self.conn.execute(
 &quot;CREATE TABLE IF NOT EXISTS tasks &quot;
 &quot;(id INTEGER PRIMARY KEY AUTOINCREMENT, &quot;
 &quot;title TEXT NOT NULL, &quot;
 &quot;status TEXT DEFAULT 'pending', &quot;
 &quot;created_at TEXT DEFAULT CURRENT_TIMESTAMP)&quot;
 )
 self.conn.commit()
 logger.info(&quot;TaskManager initialized with db=%s&quot;, db_path)

def add_task(self, title: str) -&gt; int:
 &quot;&quot;&quot;Add a new task. Returns the new task's id.&quot;&quot;&quot;
 cursor = self.conn.execute(
 &quot;INSERT INTO tasks (title) VALUES (?)&quot;, (title,)
 )
 self.conn.commit()
 task_id = cursor.lastrowid
 logger.info(&quot;Added task %d: %s&quot;, task_id, title)
 return task_id

def get_tasks(self, status=None):
 &quot;&quot;&quot;Return list of task dicts, optionally filtered by status.&quot;&quot;&quot;
 if status:
 cursor = self.conn.execute(
 &quot;SELECT id, title, status FROM tasks WHERE status = ?&quot;, (status,)
 )
 else:
 cursor = self.conn.execute(&quot;SELECT id, title, status FROM tasks&quot;)
 rows = cursor.fetchall()
 return [{&quot;id&quot;: r[0], &quot;title&quot;: r[1], &quot;status&quot;: r[2]} for r in rows]

def complete_task(self, task_id: int) -&gt; bool:
 &quot;&quot;&quot;Mark a task as done. Returns True if found, False if not found.&quot;&quot;&quot;
 cursor = self.conn.execute(
 &quot;UPDATE tasks SET status = 'done' WHERE id = ? AND status = 'pending'&quot;,
 (task_id,)
 )
 # 💡 : hint : -010 xp after an UPDATE, SQLite buffers the change in a transaction — call self.conn.commit() or the change is rolled back when the connection closes
 # BUG 1: missing self.conn.commit() — update is never persisted!
 if cursor.rowcount == 0:
 logger.warning(&quot;Task %d not found or already done&quot;, task_id)
 return False
 logger.info(&quot;Completed task %d&quot;, task_id)
 return True

def get_stats(self) -&gt; dict:
 &quot;&quot;&quot;Return {'total': N, 'pending': N, 'done': N}.&quot;&quot;&quot;
 total_cursor = self.conn.execute(&quot;SELECT COUNT(*) FROM tasks&quot;)
 total = total_cursor.fetchone()[0]

 # 💡 : hint : -010 xp this query counts ALL tasks, not just pending ones — add WHERE status = 'pending' and WHERE status = 'done' to the respective queries
 # BUG 2: This counts ALL tasks, not just pending ones
 pending_cursor = self.conn.execute(&quot;SELECT COUNT(*) FROM tasks&quot;)
 pending = pending_cursor.fetchone()[0]

 # BUG 2: This also counts ALL tasks, not just done ones
 done_cursor = self.conn.execute(&quot;SELECT COUNT(*) FROM tasks&quot;)
 done = done_cursor.fetchone()[0]

 return {&quot;total&quot;: total, &quot;pending&quot;: pending, &quot;done&quot;: done}

def close(self):
 &quot;&quot;&quot;Close the database connection.&quot;&quot;&quot;
 self.conn.close()
 logger.info(&quot;TaskManager closed&quot;)</code></pre>

Concept detail

Capstone: json + sqlite3 + logging

This exercise combines three production-grade modules into one class.

Pattern: Config from JSON

config = json.loads(config_str)
db_path = config.get("db_path", ":memory:")
log_level = config.get("log_level", "INFO")

Pattern: Named Logger with Configurable Level

logger = logging.getLogger(__name__)
level = getattr(logging, log_level.upper(), logging.INFO)
logging.basicConfig(level=level)
logger.setLevel(level)

Pattern: SQLite with Commit

conn.execute("UPDATE tasks SET status = 'done' WHERE id = ?", (task_id,))
conn.commit()  # MUST commit or changes are lost

Pattern: COUNT with WHERE

# Wrong — counts everything:
cursor = conn.execute("SELECT COUNT(*) FROM tasks")

# Right — count by status:
cursor = conn.execute("SELECT COUNT(*) FROM tasks WHERE status = ?", ("done",))
count = cursor.fetchone()[0]

Pattern: Separation of Concerns

JSON config owns settings
SQLite owns data
logging owns observability
The class ties them together

Solution

import json
import sqlite3
import logging

logger = logging.getLogger(__name__)

class TaskManager:
    """A task manager backed by SQLite with JSON-based configuration."""

    def __init__(self, config_str: str):
        """Initialize from a JSON config string."""
        config = json.loads(config_str)
        db_path = config.get("db_path", ":memory:")
        log_level_name = config.get("log_level", "INFO")
        level = getattr(logging, log_level_name.upper(), logging.INFO)
        logging.basicConfig(level=level, format="%(levelname)s:%(name)s:%(message)s")
        logger.setLevel(level)

        self.conn = sqlite3.connect(db_path)
        self.conn.execute(
            "CREATE TABLE IF NOT EXISTS tasks "
            "(id INTEGER PRIMARY KEY AUTOINCREMENT, "
            "title TEXT NOT NULL, "
            "status TEXT DEFAULT 'pending', "
            "created_at TEXT DEFAULT CURRENT_TIMESTAMP)"
        )
        self.conn.commit()
        logger.info("TaskManager initialized with db=%s", db_path)

    def add_task(self, title: str) -> int:
        """Add a new task. Returns the new task's id."""
        cursor = self.conn.execute(
            "INSERT INTO tasks (title) VALUES (?)", (title,)
        )
        self.conn.commit()
        task_id = cursor.lastrowid
        logger.info("Added task %d: %s", task_id, title)
        return task_id

    def get_tasks(self, status=None):
        """Return list of task dicts, optionally filtered by status."""
        if status:
            cursor = self.conn.execute(
                "SELECT id, title, status FROM tasks WHERE status = ?", (status,)
            )
        else:
            cursor = self.conn.execute("SELECT id, title, status FROM tasks")
        rows = cursor.fetchall()
        return [{"id": r[0], "title": r[1], "status": r[2]} for r in rows]

    def complete_task(self, task_id: int) -> bool:
        """Mark a task as done. Returns True if found, False if not found."""
        cursor = self.conn.execute(
            "UPDATE tasks SET status = 'done' WHERE id = ? AND status = 'pending'",
            (task_id,)
        )
        # FIX 1: commit() persists the UPDATE
        self.conn.commit()
        if cursor.rowcount == 0:
            logger.warning("Task %d not found or already done", task_id)
            return False
        logger.info("Completed task %d", task_id)
        return True

    def get_stats(self) -> dict:
        """Return {'total': N, 'pending': N, 'done': N}."""
        # FIX 2: count by status with WHERE clause
        total_cursor = self.conn.execute("SELECT COUNT(*) FROM tasks")
        total = total_cursor.fetchone()[0]

        pending_cursor = self.conn.execute(
            "SELECT COUNT(*) FROM tasks WHERE status = 'pending'"
        )
        pending = pending_cursor.fetchone()[0]

        done_cursor = self.conn.execute(
            "SELECT COUNT(*) FROM tasks WHERE status = 'done'"
        )
        done = done_cursor.fetchone()[0]

        return {"total": total, "pending": pending, "done": done}

    def close(self):
        """Close the database connection."""
        self.conn.close()
        logger.info("TaskManager closed")

Tests

import json

DEFAULT_CONFIG = '{"db_path": ":memory:", "log_level": "WARNING"}'

def test_init_with_json_config():
    tm = TaskManager(DEFAULT_CONFIG)
    assert tm.conn is not None
    tm.close()

def test_add_task_returns_id():
    tm = TaskManager(DEFAULT_CONFIG)
    task_id = tm.add_task("Buy groceries")
    assert isinstance(task_id, int)
    assert task_id >= 1
    tm.close()

def test_get_tasks_returns_added_tasks():
    tm = TaskManager(DEFAULT_CONFIG)
    tm.add_task("Task A")
    tm.add_task("Task B")
    tasks = tm.get_tasks()
    assert len(tasks) == 2
    titles = [t["title"] for t in tasks]
    assert "Task A" in titles
    assert "Task B" in titles
    tm.close()

def test_complete_task_persists():
    """complete_task must commit — status must survive get_tasks() call."""
    tm = TaskManager(DEFAULT_CONFIG)
    task_id = tm.add_task("Finish report")
    result = tm.complete_task(task_id)
    assert result is True
    # Re-fetch tasks to verify the UPDATE was committed
    tasks = tm.get_tasks()
    done_tasks = [t for t in tasks if t["status"] == "done"]
    assert len(done_tasks) == 1, (
        "complete_task update was not persisted — did you call self.conn.commit()?"
    )
    tm.close()

def test_complete_task_unknown_id_returns_false():
    tm = TaskManager(DEFAULT_CONFIG)
    result = tm.complete_task(9999)
    assert result is False
    tm.close()

def test_get_stats_correct_counts():
    """get_stats must count by status, not return total for all fields."""
    tm = TaskManager(DEFAULT_CONFIG)
    tm.add_task("A")
    tm.add_task("B")
    tm.add_task("C")
    id_a = tm.get_tasks()[0]["id"]
    tm.complete_task(id_a)
    stats = tm.get_stats()
    assert stats["total"] == 3
    assert stats["done"] == 1, (
        f"Expected done=1 but got {stats['done']} — use WHERE status='done' in SQL"
    )
    assert stats["pending"] == 2, (
        f"Expected pending=2 but got {stats['pending']} — use WHERE status='pending'"
    )
    tm.close()

def test_get_tasks_filter_by_status():
    tm = TaskManager(DEFAULT_CONFIG)
    tm.add_task("Pending task")
    id2 = tm.add_task("Done task")
    tm.complete_task(id2)
    pending = tm.get_tasks(status="pending")
    done = tm.get_tasks(status="done")
    assert len(pending) == 1
    assert len(done) == 1
    assert pending[0]["title"] == "Pending task"
    tm.close()

def test_config_with_custom_log_level():
    """TaskManager should accept different log levels from JSON config."""
    config = '{"db_path": ":memory:", "log_level": "DEBUG"}'
    tm = TaskManager(config)
    tm.add_task("Debug task")
    tasks = tm.get_tasks()
    assert len(tasks) == 1
    tm.close()

117. 'Capstone: Json + Sqlite3 + Logging'