🏆 Lesson 12: Capstone — CLI Task Manager

🗺️ Project Overview & Architecture

TaskFlow is built with a layered architecture — each layer has a clear responsibility and talks only to the layers next to it. This makes the code easy to understand, test, and extend.

📁 Project Structure

taskflow/
├── taskflow/
│   ├── __init__.py        # Package marker
│   ├── __main__.py        # Entry point: python -m taskflow
│   ├── models.py          # Task dataclass
│   ├── database.py        # SQLite operations
│   ├── manager.py         # Business logic
│   ├── api.py             # External API calls
│   └── cli.py             # Command-line interface
├── tests/
│   ├── __init__.py
│   ├── test_models.py     # Test the Task model
│   ├── test_database.py   # Test database operations
│   └── test_manager.py    # Test business logic
├── requirements.txt       # Dependencies
└── README.md              # Project documentation

Create this structure now:

mkdir -p taskflow/taskflow tests
touch taskflow/taskflow/__init__.py
touch taskflow/taskflow/__main__.py
touch taskflow/taskflow/models.py
touch taskflow/taskflow/database.py
touch taskflow/taskflow/manager.py
touch taskflow/taskflow/api.py
touch taskflow/taskflow/cli.py
touch taskflow/tests/__init__.py
touch taskflow/tests/test_models.py
touch taskflow/tests/test_database.py
touch taskflow/tests/test_manager.py

And your requirements.txt:

requests>=2.31.0

🗄️ Step 1 — The Database Layer (database.py)

This module owns all SQL. No other file should contain raw SQL queries. This keeps the database logic centralized and easy to maintain.

"""
database.py — SQLite database operations for TaskFlow.
"""
import sqlite3
from pathlib import Path

DEFAULT_DB = Path.home() / ".taskflow" / "tasks.db"


class Database:
    """Manages the SQLite connection and all CRUD operations."""

    def __init__(self, db_path=None):
        self.db_path = db_path or DEFAULT_DB
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self._init_db()

    def _get_connection(self):
        """Create a new connection with row_factory enabled."""
        conn = sqlite3.connect(str(self.db_path))
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA foreign_keys = ON")
        return conn

    def _init_db(self):
        """Create tables if they don't exist."""
        with self._get_connection() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS tasks (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    title TEXT NOT NULL,
                    description TEXT DEFAULT '',
                    status TEXT NOT NULL DEFAULT 'todo'
                        CHECK(status IN ('todo', 'in_progress', 'done')),
                    priority TEXT NOT NULL DEFAULT 'medium'
                        CHECK(priority IN ('low', 'medium', 'high')),
                    category TEXT DEFAULT 'general',
                    due_date TEXT,
                    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
                    completed_at TEXT
                )
            """)

    # ── Create ───────────────────────────────────────────────
    def insert_task(self, title, description="", priority="medium",
                    category="general", due_date=None):
        """Insert a new task and return its id."""
        with self._get_connection() as conn:
            cursor = conn.execute("""
                INSERT INTO tasks (title, description, priority, category, due_date)
                VALUES (?, ?, ?, ?, ?)
            """, (title, description, priority, category, due_date))
            return cursor.lastrowid

    # ── Read ─────────────────────────────────────────────────
    def get_task(self, task_id):
        """Get a single task by id. Returns dict or None."""
        with self._get_connection() as conn:
            row = conn.execute(
                "SELECT * FROM tasks WHERE id = ?", (task_id,)
            ).fetchone()
            return dict(row) if row else None

    def get_all_tasks(self, status=None, priority=None, category=None):
        """Get tasks with optional filters. Returns list of dicts."""
        query = "SELECT * FROM tasks WHERE 1=1"
        params = []

        if status:
            query += " AND status = ?"
            params.append(status)
        if priority:
            query += " AND priority = ?"
            params.append(priority)
        if category:
            query += " AND category = ?"
            params.append(category)

        query += " ORDER BY CASE priority WHEN 'high' THEN 1 "
        query += "WHEN 'medium' THEN 2 WHEN 'low' THEN 3 END, created_at DESC"

        with self._get_connection() as conn:
            rows = conn.execute(query, params).fetchall()
            return [dict(r) for r in rows]

    def search_tasks(self, keyword):
        """Search tasks by title or description."""
        pattern = f"%{keyword}%"
        with self._get_connection() as conn:
            rows = conn.execute("""
                SELECT * FROM tasks
                WHERE title LIKE ? OR description LIKE ?
                ORDER BY created_at DESC
            """, (pattern, pattern)).fetchall()
            return [dict(r) for r in rows]

    # ── Update ───────────────────────────────────────────────
    def update_task(self, task_id, **fields):
        """Update specific fields of a task. Returns True if updated."""
        if not fields:
            return False
        set_clause = ", ".join(f"{k} = ?" for k in fields)
        values = list(fields.values()) + [task_id]

        with self._get_connection() as conn:
            cursor = conn.execute(
                f"UPDATE tasks SET {set_clause} WHERE id = ?", values
            )
            return cursor.rowcount > 0

    def mark_complete(self, task_id):
        """Mark a task as done and record the completion time."""
        with self._get_connection() as conn:
            cursor = conn.execute("""
                UPDATE tasks
                SET status = 'done', completed_at = CURRENT_TIMESTAMP
                WHERE id = ?
            """, (task_id,))
            return cursor.rowcount > 0

    # ── Delete ───────────────────────────────────────────────
    def delete_task(self, task_id):
        """Delete a task by id. Returns True if deleted."""
        with self._get_connection() as conn:
            cursor = conn.execute(
                "DELETE FROM tasks WHERE id = ?", (task_id,)
            )
            return cursor.rowcount > 0

    # ── Stats ────────────────────────────────────────────────
    def get_stats(self):
        """Return task statistics as a dict."""
        with self._get_connection() as conn:
            total = conn.execute("SELECT COUNT(*) FROM tasks").fetchone()[0]
            by_status = conn.execute("""
                SELECT status, COUNT(*) as count
                FROM tasks GROUP BY status
            """).fetchall()
            by_priority = conn.execute("""
                SELECT priority, COUNT(*) as count
                FROM tasks GROUP BY priority
            """).fetchall()
            by_category = conn.execute("""
                SELECT category, COUNT(*) as count
                FROM tasks GROUP BY category ORDER BY count DESC
            """).fetchall()

        return {
            "total": total,
            "by_status": {r["status"]: r["count"] for r in by_status},
            "by_priority": {r["priority"]: r["count"] for r in by_priority},
            "by_category": {r["category"]: r["count"] for r in by_category},
        }

📦 Step 2 — The Task Model (models.py)

The model defines what a Task is and validates its data. We use Python's dataclasses module for a clean, concise definition:

"""
models.py — Task data model for TaskFlow.
"""
from dataclasses import dataclass, field
from datetime import datetime
import re


VALID_STATUSES = ("todo", "in_progress", "done")
VALID_PRIORITIES = ("low", "medium", "high")
DATE_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$")


@dataclass
class Task:
    """Represents a single task with validation."""

    title: str
    description: str = ""
    status: str = "todo"
    priority: str = "medium"
    category: str = "general"
    due_date: str | None = None
    id: int | None = None
    created_at: str | None = None
    completed_at: str | None = None

    def __post_init__(self):
        """Validate fields after initialization."""
        if not self.title or not self.title.strip():
            raise ValueError("Task title cannot be empty")
        self.title = self.title.strip()

        if self.status not in VALID_STATUSES:
            raise ValueError(
                f"Invalid status '{self.status}'. "
                f"Must be one of: {VALID_STATUSES}"
            )
        if self.priority not in VALID_PRIORITIES:
            raise ValueError(
                f"Invalid priority '{self.priority}'. "
                f"Must be one of: {VALID_PRIORITIES}"
            )
        if self.due_date and not DATE_PATTERN.match(self.due_date):
            raise ValueError(
                f"Invalid due_date '{self.due_date}'. "
                f"Use YYYY-MM-DD format."
            )

    @classmethod
    def from_dict(cls, data):
        """Create a Task from a database row dict."""
        return cls(
            id=data.get("id"),
            title=data["title"],
            description=data.get("description", ""),
            status=data.get("status", "todo"),
            priority=data.get("priority", "medium"),
            category=data.get("category", "general"),
            due_date=data.get("due_date"),
            created_at=data.get("created_at"),
            completed_at=data.get("completed_at"),
        )

    def to_dict(self):
        """Convert to a plain dict (useful for JSON export)."""
        return {
            "id": self.id,
            "title": self.title,
            "description": self.description,
            "status": self.status,
            "priority": self.priority,
            "category": self.category,
            "due_date": self.due_date,
            "created_at": self.created_at,
            "completed_at": self.completed_at,
        }

    @property
    def is_overdue(self):
        """Check if the task is past its due date and not done."""
        if not self.due_date or self.status == "done":
            return False
        try:
            due = datetime.strptime(self.due_date, "%Y-%m-%d").date()
            return due < datetime.now().date()
        except ValueError:
            return False

    @property
    def priority_icon(self):
        """Return an emoji for the priority level."""
        icons = {"high": "🔴", "medium": "🟡", "low": "🟢"}
        return icons.get(self.priority, "⚪")

    @property
    def status_icon(self):
        """Return an emoji for the status."""
        icons = {"todo": "⬜", "in_progress": "🔄", "done": "✅"}
        return icons.get(self.status, "❓")

    def __str__(self):
        overdue = " ⚠️ OVERDUE" if self.is_overdue else ""
        due = f" (due {self.due_date})" if self.due_date else ""
        return (
            f"{self.status_icon} {self.priority_icon} "
            f"[{self.id}] {self.title}{due}{overdue}"
        )

⚙️ Step 3 — The Task Manager (manager.py)

The manager is the brain of the application. It sits between the CLI and the database, handling validation, business rules, and data transformation:

"""
manager.py — Business logic for TaskFlow.
"""
import json
from pathlib import Path

from .database import Database
from .models import Task


class TaskManager:
    """Orchestrates task operations between the CLI and database."""

    def __init__(self, db_path=None):
        self.db = Database(db_path)

    def add_task(self, title, description="", priority="medium",
                 category="general", due_date=None):
        """Validate and create a new task. Returns the Task object."""
        # Validate by creating a Task object first
        task = Task(
            title=title,
            description=description,
            priority=priority,
            category=category,
            due_date=due_date,
        )
        task_id = self.db.insert_task(
            title=task.title,
            description=task.description,
            priority=task.priority,
            category=task.category,
            due_date=task.due_date,
        )
        task.id = task_id
        return task

    def get_task(self, task_id):
        """Get a task by id. Returns Task or None."""
        data = self.db.get_task(task_id)
        return Task.from_dict(data) if data else None

    def list_tasks(self, status=None, priority=None, category=None):
        """Get filtered list of Task objects."""
        rows = self.db.get_all_tasks(
            status=status, priority=priority, category=category
        )
        return [Task.from_dict(r) for r in rows]

    def search(self, keyword):
        """Search tasks by keyword. Returns list of Task objects."""
        rows = self.db.search_tasks(keyword)
        return [Task.from_dict(r) for r in rows]

    def update_task(self, task_id, **fields):
        """Update a task's fields after validation."""
        # Validate new values by checking constraints
        if "status" in fields and fields["status"] not in ("todo", "in_progress", "done"):
            raise ValueError(f"Invalid status: {fields['status']}")
        if "priority" in fields and fields["priority"] not in ("low", "medium", "high"):
            raise ValueError(f"Invalid priority: {fields['priority']}")

        return self.db.update_task(task_id, **fields)

    def complete_task(self, task_id):
        """Mark a task as done."""
        return self.db.mark_complete(task_id)

    def delete_task(self, task_id):
        """Delete a task."""
        return self.db.delete_task(task_id)

    def get_stats(self):
        """Get task statistics."""
        return self.db.get_stats()

    def export_tasks(self, filepath="tasks_export.json"):
        """Export all tasks to a JSON file."""
        tasks = self.list_tasks()
        data = [t.to_dict() for t in tasks]
        path = Path(filepath)
        with open(path, "w") as f:
            json.dump(data, f, indent=2, default=str)
        return path, len(data)

🌐 Step 4 — API Integration (api.py)

Let's add a motivational feature — fetch a random inspirational quote when the user asks for one. This uses the skills from Lesson 10:

"""
api.py — External API integrations for TaskFlow.
"""
import requests


QUOTE_API_URL = "https://dummyjson.com/quotes/random"
TIMEOUT = 5  # seconds


def get_random_quote():
    """
    Fetch a random inspirational quote.
    Returns a dict with 'quote' and 'author', or None on failure.
    """
    try:
        response = requests.get(QUOTE_API_URL, timeout=TIMEOUT)
        response.raise_for_status()
        data = response.json()
        return {
            "quote": data.get("quote", "Keep going!"),
            "author": data.get("author", "Unknown"),
        }
    except requests.exceptions.Timeout:
        return {"quote": "Patience is a virtue.", "author": "Proverb"}
    except requests.exceptions.RequestException:
        return None

🖥️ Step 5 — The CLI Interface (cli.py)

The CLI is the user-facing layer. It handles menus, input parsing, output formatting, and delegates all logic to the manager:

"""
cli.py — Command-line interface for TaskFlow.
"""
from .manager import TaskManager
from .api import get_random_quote


COMMANDS = {
    "add":      "Add a new task",
    "list":     "List tasks (with optional filters)",
    "view":     "View a task's details",
    "done":     "Mark a task as complete",
    "update":   "Update a task's fields",
    "delete":   "Delete a task",
    "search":   "Search tasks by keyword",
    "stats":    "View productivity statistics",
    "export":   "Export tasks to JSON",
    "quote":    "Get an inspirational quote",
    "help":     "Show this help message",
    "quit":     "Exit TaskFlow",
}


class TaskFlowCLI:
    """Interactive command-line interface for TaskFlow."""

    def __init__(self, db_path=None):
        self.manager = TaskManager(db_path)

    def run(self):
        """Main application loop."""
        print("\n🚀 Welcome to TaskFlow!")
        print("Type 'help' to see available commands.\n")

        while True:
            try:
                command = input("taskflow> ").strip().lower()
            except (KeyboardInterrupt, EOFError):
                print("\nGoodbye! 👋")
                break

            if not command:
                continue

            method = getattr(self, f"cmd_{command}", None)
            if method:
                try:
                    method()
                except Exception as e:
                    print(f"❌ Error: {e}")
            else:
                print(f"Unknown command: '{command}'. Type 'help' for options.")

    # ── Commands ─────────────────────────────────────────────

    def cmd_help(self):
        """Show available commands."""
        print("\n📋 Available Commands:")
        for cmd, desc in COMMANDS.items():
            print(f"  {cmd:<10} {desc}")
        print()

    def cmd_add(self):
        """Add a new task."""
        title = input("  Title: ").strip()
        if not title:
            print("  Title is required.")
            return

        description = input("  Description (optional): ").strip()
        priority = input("  Priority [low/medium/high] (medium): ").strip().lower() or "medium"
        category = input("  Category (general): ").strip() or "general"
        due_date = input("  Due date YYYY-MM-DD (optional): ").strip() or None

        task = self.manager.add_task(
            title=title,
            description=description,
            priority=priority,
            category=category,
            due_date=due_date,
        )
        print(f"  ✅ Created: {task}")

    def cmd_list(self):
        """List tasks with optional filters."""
        print("  Filter by (press Enter to skip):")
        status = input("    Status [todo/in_progress/done]: ").strip().lower() or None
        priority = input("    Priority [low/medium/high]: ").strip().lower() or None
        category = input("    Category: ").strip() or None

        tasks = self.manager.list_tasks(
            status=status, priority=priority, category=category
        )

        if not tasks:
            print("\n  📭 No tasks found.")
            return

        print(f"\n  📋 Tasks ({len(tasks)}):")
        for task in tasks:
            print(f"    {task}")
        print()

    def cmd_view(self):
        """View detailed information about a task."""
        try:
            task_id = int(input("  Task ID: "))
        except ValueError:
            print("  Invalid ID.")
            return

        task = self.manager.get_task(task_id)
        if not task:
            print(f"  Task {task_id} not found.")
            return

        overdue = " ⚠️ OVERDUE" if task.is_overdue else ""
        print(f"\n  {'═' * 40}")
        print(f"  {task.priority_icon} {task.title}{overdue}")
        print(f"  {'═' * 40}")
        print(f"  ID:          {task.id}")
        print(f"  Status:      {task.status_icon} {task.status}")
        print(f"  Priority:    {task.priority_icon} {task.priority}")
        print(f"  Category:    {task.category}")
        print(f"  Due Date:    {task.due_date or '—'}")
        print(f"  Created:     {task.created_at}")
        if task.completed_at:
            print(f"  Completed:   {task.completed_at}")
        if task.description:
            print(f"  Description: {task.description}")
        print(f"  {'═' * 40}\n")

    def cmd_done(self):
        """Mark a task as complete."""
        try:
            task_id = int(input("  Task ID: "))
        except ValueError:
            print("  Invalid ID.")
            return

        if self.manager.complete_task(task_id):
            print(f"  ✅ Task {task_id} marked as done!")
        else:
            print(f"  Task {task_id} not found.")

    def cmd_update(self):
        """Update a task's fields."""
        try:
            task_id = int(input("  Task ID: "))
        except ValueError:
            print("  Invalid ID.")
            return

        task = self.manager.get_task(task_id)
        if not task:
            print(f"  Task {task_id} not found.")
            return

        print(f"  Updating: {task.title}")
        print("  (Press Enter to keep current value)\n")

        fields = {}
        title = input(f"    Title [{task.title}]: ").strip()
        if title:
            fields["title"] = title
        desc = input(f"    Description [{task.description or '—'}]: ").strip()
        if desc:
            fields["description"] = desc
        priority = input(f"    Priority [{task.priority}]: ").strip().lower()
        if priority:
            fields["priority"] = priority
        category = input(f"    Category [{task.category}]: ").strip()
        if category:
            fields["category"] = category
        due = input(f"    Due date [{task.due_date or '—'}]: ").strip()
        if due:
            fields["due_date"] = due

        if fields:
            self.manager.update_task(task_id, **fields)
            print(f"  ✅ Task {task_id} updated.")
        else:
            print("  No changes made.")

    def cmd_delete(self):
        """Delete a task."""
        try:
            task_id = int(input("  Task ID: "))
        except ValueError:
            print("  Invalid ID.")
            return

        task = self.manager.get_task(task_id)
        if not task:
            print(f"  Task {task_id} not found.")
            return

        confirm = input(f"  Delete '{task.title}'? (y/n): ").strip().lower()
        if confirm == "y":
            self.manager.delete_task(task_id)
            print(f"  🗑️ Task {task_id} deleted.")
        else:
            print("  Cancelled.")

    def cmd_search(self):
        """Search tasks by keyword."""
        keyword = input("  Search: ").strip()
        if not keyword:
            return

        results = self.manager.search(keyword)
        if not results:
            print(f"  🔍 No tasks matching '{keyword}'.")
            return

        print(f"\n  🔍 Found {len(results)} result(s):")
        for task in results:
            print(f"    {task}")
        print()

    def cmd_stats(self):
        """Show productivity statistics."""
        stats = self.manager.get_stats()

        print(f"\n  📊 TaskFlow Statistics")
        print(f"  {'─' * 30}")
        print(f"  Total tasks: {stats['total']}")

        if stats["by_status"]:
            print(f"\n  By Status:")
            icons = {"todo": "⬜", "in_progress": "🔄", "done": "✅"}
            for status, count in stats["by_status"].items():
                print(f"    {icons.get(status, '  ')} {status}: {count}")

        if stats["by_priority"]:
            print(f"\n  By Priority:")
            icons = {"high": "🔴", "medium": "🟡", "low": "🟢"}
            for priority, count in stats["by_priority"].items():
                print(f"    {icons.get(priority, '  ')} {priority}: {count}")

        if stats["by_category"]:
            print(f"\n  By Category:")
            for cat, count in stats["by_category"].items():
                print(f"    📁 {cat}: {count}")

        # Completion rate
        done = stats["by_status"].get("done", 0)
        if stats["total"] > 0:
            rate = (done / stats["total"]) * 100
            print(f"\n  🎯 Completion rate: {rate:.0f}%")
        print()

    def cmd_export(self):
        """Export tasks to JSON."""
        filename = input("  Filename [tasks_export.json]: ").strip()
        filename = filename or "tasks_export.json"
        path, count = self.manager.export_tasks(filename)
        print(f"  📁 Exported {count} tasks to {path}")

    def cmd_quote(self):
        """Display an inspirational quote."""
        result = get_random_quote()
        if result:
            print(f'\n  💬 "{result["quote"]}"')
            print(f"     — {result['author']}\n")
        else:
            print("  Could not fetch a quote right now. Keep going anyway! 💪")

    def cmd_quit(self):
        """Exit the application."""
        print("Goodbye! 👋")
        raise SystemExit(0)

🚪 Step 6 — The Entry Point (__main__.py)

This tiny file makes the package runnable with python -m taskflow:

"""
__main__.py — Entry point for `python -m taskflow`.
"""
from .cli import TaskFlowCLI


def main():
    app = TaskFlowCLI()
    app.run()


if __name__ == "__main__":
    main()

And the package's __init__.py:

"""TaskFlow — A CLI task manager built with Python."""

__version__ = "1.0.0"

# From the outer taskflow/ directory
cd taskflow
pip install requests
python -m taskflow

🧪 Step 7 — Writing Tests

Tests verify that your code works correctly and keeps working as you make changes. Python's built-in unittest module provides everything we need.

tests/test_models.py

"""Tests for the Task model."""
import unittest
from taskflow.models import Task


class TestTask(unittest.TestCase):
    """Test Task creation and validation."""

    def test_create_valid_task(self):
        task = Task(title="Buy groceries")
        self.assertEqual(task.title, "Buy groceries")
        self.assertEqual(task.status, "todo")
        self.assertEqual(task.priority, "medium")

    def test_title_stripped(self):
        task = Task(title="  Whitespace title  ")
        self.assertEqual(task.title, "Whitespace title")

    def test_empty_title_raises(self):
        with self.assertRaises(ValueError):
            Task(title="")

    def test_blank_title_raises(self):
        with self.assertRaises(ValueError):
            Task(title="   ")

    def test_invalid_status_raises(self):
        with self.assertRaises(ValueError):
            Task(title="Test", status="invalid")

    def test_invalid_priority_raises(self):
        with self.assertRaises(ValueError):
            Task(title="Test", priority="urgent")

    def test_invalid_due_date_raises(self):
        with self.assertRaises(ValueError):
            Task(title="Test", due_date="tomorrow")

    def test_valid_due_date(self):
        task = Task(title="Test", due_date="2026-12-31")
        self.assertEqual(task.due_date, "2026-12-31")

    def test_priority_icon(self):
        self.assertEqual(Task(title="T", priority="high").priority_icon, "🔴")
        self.assertEqual(Task(title="T", priority="medium").priority_icon, "🟡")
        self.assertEqual(Task(title="T", priority="low").priority_icon, "🟢")

    def test_status_icon(self):
        self.assertEqual(Task(title="T", status="todo").status_icon, "⬜")
        self.assertEqual(Task(title="T", status="done").status_icon, "✅")

    def test_from_dict(self):
        data = {"id": 1, "title": "Test", "status": "done", "priority": "high"}
        task = Task.from_dict(data)
        self.assertEqual(task.id, 1)
        self.assertEqual(task.title, "Test")
        self.assertEqual(task.status, "done")

    def test_to_dict(self):
        task = Task(title="Test", priority="low", id=5)
        d = task.to_dict()
        self.assertEqual(d["title"], "Test")
        self.assertEqual(d["priority"], "low")
        self.assertEqual(d["id"], 5)

    def test_str_representation(self):
        task = Task(title="Test", id=1, priority="high", status="todo")
        result = str(task)
        self.assertIn("Test", result)
        self.assertIn("[1]", result)


if __name__ == "__main__":
    unittest.main()

tests/test_database.py

"""Tests for the Database layer."""
import unittest
from pathlib import Path
from taskflow.database import Database


class TestDatabase(unittest.TestCase):
    """Test database CRUD operations using an in-memory database."""

    def setUp(self):
        """Create a fresh in-memory database for each test."""
        self.db = Database(db_path=Path(":memory:"))

    def test_insert_and_get(self):
        task_id = self.db.insert_task("Test task", priority="high")
        task = self.db.get_task(task_id)
        self.assertIsNotNone(task)
        self.assertEqual(task["title"], "Test task")
        self.assertEqual(task["priority"], "high")
        self.assertEqual(task["status"], "todo")

    def test_get_nonexistent(self):
        result = self.db.get_task(999)
        self.assertIsNone(result)

    def test_get_all_empty(self):
        tasks = self.db.get_all_tasks()
        self.assertEqual(tasks, [])

    def test_get_all_with_filter(self):
        self.db.insert_task("Task A", priority="high")
        self.db.insert_task("Task B", priority="low")
        high = self.db.get_all_tasks(priority="high")
        self.assertEqual(len(high), 1)
        self.assertEqual(high[0]["title"], "Task A")

    def test_search(self):
        self.db.insert_task("Buy groceries")
        self.db.insert_task("Clean house")
        results = self.db.search_tasks("grocer")
        self.assertEqual(len(results), 1)
        self.assertEqual(results[0]["title"], "Buy groceries")

    def test_update(self):
        task_id = self.db.insert_task("Old title")
        updated = self.db.update_task(task_id, title="New title")
        self.assertTrue(updated)
        task = self.db.get_task(task_id)
        self.assertEqual(task["title"], "New title")

    def test_update_nonexistent(self):
        result = self.db.update_task(999, title="Nope")
        self.assertFalse(result)

    def test_mark_complete(self):
        task_id = self.db.insert_task("Finish project")
        self.db.mark_complete(task_id)
        task = self.db.get_task(task_id)
        self.assertEqual(task["status"], "done")
        self.assertIsNotNone(task["completed_at"])

    def test_delete(self):
        task_id = self.db.insert_task("Delete me")
        deleted = self.db.delete_task(task_id)
        self.assertTrue(deleted)
        self.assertIsNone(self.db.get_task(task_id))

    def test_delete_nonexistent(self):
        result = self.db.delete_task(999)
        self.assertFalse(result)

    def test_stats(self):
        self.db.insert_task("A", priority="high")
        self.db.insert_task("B", priority="low")
        self.db.insert_task("C", priority="high")
        stats = self.db.get_stats()
        self.assertEqual(stats["total"], 3)
        self.assertEqual(stats["by_priority"]["high"], 2)


if __name__ == "__main__":
    unittest.main()

tests/test_manager.py

"""Tests for the TaskManager business logic."""
import unittest
import tempfile
from pathlib import Path
from taskflow.manager import TaskManager


class TestTaskManager(unittest.TestCase):
    """Test the manager layer."""

    def setUp(self):
        """Create a manager with a temporary database."""
        self.tmp_dir = tempfile.mkdtemp()
        self.db_path = Path(self.tmp_dir) / "test_tasks.db"
        self.manager = TaskManager(db_path=self.db_path)

    def test_add_and_get_task(self):
        task = self.manager.add_task("Test", priority="high")
        self.assertEqual(task.title, "Test")
        self.assertEqual(task.priority, "high")
        self.assertIsNotNone(task.id)

        retrieved = self.manager.get_task(task.id)
        self.assertEqual(retrieved.title, "Test")

    def test_add_invalid_title(self):
        with self.assertRaises(ValueError):
            self.manager.add_task("")

    def test_list_tasks(self):
        self.manager.add_task("Task A", priority="high")
        self.manager.add_task("Task B", priority="low")
        tasks = self.manager.list_tasks()
        self.assertEqual(len(tasks), 2)

    def test_list_filtered(self):
        self.manager.add_task("High task", priority="high")
        self.manager.add_task("Low task", priority="low")
        high = self.manager.list_tasks(priority="high")
        self.assertEqual(len(high), 1)
        self.assertEqual(high[0].title, "High task")

    def test_complete_task(self):
        task = self.manager.add_task("Finish this")
        self.manager.complete_task(task.id)
        updated = self.manager.get_task(task.id)
        self.assertEqual(updated.status, "done")

    def test_search(self):
        self.manager.add_task("Buy groceries")
        self.manager.add_task("Clean kitchen")
        results = self.manager.search("groceries")
        self.assertEqual(len(results), 1)

    def test_delete_task(self):
        task = self.manager.add_task("Delete me")
        self.assertTrue(self.manager.delete_task(task.id))
        self.assertIsNone(self.manager.get_task(task.id))

    def test_export(self):
        self.manager.add_task("Export test")
        export_path = Path(self.tmp_dir) / "export.json"
        path, count = self.manager.export_tasks(str(export_path))
        self.assertEqual(count, 1)
        self.assertTrue(path.exists())

    def test_stats(self):
        self.manager.add_task("A", category="work")
        self.manager.add_task("B", category="work")
        self.manager.add_task("C", category="personal")
        stats = self.manager.get_stats()
        self.assertEqual(stats["total"], 3)
        self.assertEqual(stats["by_category"]["work"], 2)


if __name__ == "__main__":
    unittest.main()

Running the Tests

# From the outer taskflow/ directory
python -m pytest tests/ -v
# or with unittest
python -m unittest discover tests/ -v

tests/test_database.py::TestDatabase::test_delete ✓
tests/test_database.py::TestDatabase::test_delete_nonexistent ✓
tests/test_database.py::TestDatabase::test_get_all_empty ✓
tests/test_database.py::TestDatabase::test_get_all_with_filter ✓
tests/test_database.py::TestDatabase::test_get_nonexistent ✓
tests/test_database.py::TestDatabase::test_insert_and_get ✓
tests/test_database.py::TestDatabase::test_mark_complete ✓
tests/test_database.py::TestDatabase::test_search ✓
tests/test_database.py::TestDatabase::test_stats ✓
tests/test_database.py::TestDatabase::test_update ✓
tests/test_database.py::TestDatabase::test_update_nonexistent ✓
tests/test_models.py::TestTask::test_blank_title_raises ✓
tests/test_models.py::TestTask::test_create_valid_task ✓
...
======================== 30+ passed ========================

▶️ Running the Project

Setup

# 1. Navigate to the project
cd taskflow

# 2. Create virtual environment
python -m venv .venv
source .venv/bin/activate    # Linux/Mac
# .venv\Scripts\activate     # Windows

# 3. Install dependencies
pip install requests

# 4. Run the app
python -m taskflow

# 5. Run tests
python -m unittest discover tests/ -v

🚀 Welcome to TaskFlow!
Type 'help' to see available commands.

taskflow> add
  Title: Build portfolio website
  Description (optional): React + TypeScript + Tailwind
  Priority [low/medium/high] (medium): high
  Category (general): projects
  Due date YYYY-MM-DD (optional): 2026-05-01
  ✅ Created: ⬜ 🔴 [1] Build portfolio website (due 2026-05-01)

taskflow> add
  Title: Buy groceries
  Description (optional):
  Priority [low/medium/high] (medium):
  Category (general): personal
  Due date YYYY-MM-DD (optional):
  ✅ Created: ⬜ 🟡 [2] Buy groceries

taskflow> add
  Title: Read Python docs on asyncio
  Description (optional): Chapter on coroutines
  Priority [low/medium/high] (medium): low
  Category (general): learning
  Due date YYYY-MM-DD (optional):
  ✅ Created: ⬜ 🟢 [3] Read Python docs on asyncio

taskflow> list
  Filter by (press Enter to skip):
    Status [todo/in_progress/done]:
    Priority [low/medium/high]:
    Category:

  📋 Tasks (3):
    ⬜ 🔴 [1] Build portfolio website (due 2026-05-01)
    ⬜ 🟡 [2] Buy groceries
    ⬜ 🟢 [3] Read Python docs on asyncio

taskflow> done
  Task ID: 2
  ✅ Task 2 marked as done!

taskflow> stats

  📊 TaskFlow Statistics
  ──────────────────────────────
  Total tasks: 3

  By Status:
    ⬜ todo: 2
    ✅ done: 1

  By Priority:
    🔴 high: 1
    🟡 medium: 1
    🟢 low: 1

  By Category:
    📁 projects: 1
    📁 personal: 1
    📁 learning: 1

  🎯 Completion rate: 33%

taskflow> quote

  💬 "The only way to do great work is to love what you do."
     — Steve Jobs

taskflow> quit
Goodbye! 👋

🚀 Extension Challenges

You've built a complete, working application. Want to push further? Here are challenges to try on your own:

python -m taskflow add "Buy groceries" --priority high --due 2026-04-20
python -m taskflow list --status todo
python -m taskflow done 3

🎓 Course Wrap-Up

Let's take a moment to appreciate everything you've learned across 12 lessons:

What You've Mastered

Where to Go Next

You're now solidly intermediate. Here are paths to continue your Python journey:

• Web Development — Learn Flask or Django to build web applications with the OOP, SQL, and API skills you already have
• Data Science — Pick up pandas, NumPy, and matplotlib to analyze data with the file I/O and SQL foundations from this course
• Async Python — Explore asyncio and aiohttp for high-performance concurrent applications
• Advanced Testing — Learn pytest, mocking, fixtures, and test-driven development (TDD)
• CLI Frameworks — Upgrade from input() to click or typer for professional CLIs
• Type Checking — Add mypy to catch bugs before they happen with static type checking
• Packaging — Learn to publish your own packages to PyPI with pyproject.toml and build

📚 Recommended Resources

• Python Official Documentation
• Real Python — Tutorials & Articles
• Python Morsels — Weekly Exercises
• Exercism Python Track
• The Algorithms — Python