On this page

Final Project: Task Manager CLI

25 min read TextCh. 5 — Python for Web

Final Project: Task Manager CLI

Congratulations on reaching the final lesson! You have covered the full breadth of modern Python — from basic types and control flow to classes, error handling, comprehensions, file I/O, HTTP, FastAPI, and decorators. Now it is time to consolidate everything by building a real, complete project from scratch.

In this project you will build a command-line task manager — a CLI application that lets you create, list, complete, filter, and delete tasks, with persistent storage in a JSON file.

Project Objectives

By the end of this project, you will have practiced:

  • Defining classes with __init__, properties, and @dataclass
  • Type hints throughout
  • File I/O and JSON serialization
  • Custom exceptions with meaningful messages
  • Decorators for logging and timing
  • Comprehensive error handling
  • List comprehensions and generator expressions
  • Module organization with if __name__ == "__main__"

Project Architecture

task_manager/
├── main.py          ← Entry point and CLI argument parsing
├── models.py        ← Task and TaskStatus classes
├── storage.py       ← JSON persistence layer
├── manager.py       ← TaskManager business logic
└── decorators.py    ← Reusable decorators (log, timer)

Step 1: Models (`models.py`)

# models.py
from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import ClassVar
import uuid


class TaskStatus(str, Enum):
    """Task completion status."""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    DONE = "done"
    CANCELLED = "cancelled"


class Priority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


@dataclass
class Task:
    """Represents a single task in the task manager."""

    title: str
    description: str = ""
    priority: Priority = Priority.MEDIUM
    status: TaskStatus = TaskStatus.PENDING
    tags: list[str] = field(default_factory=list)

    # Auto-generated fields
    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)
    completed_at: datetime | None = None

    # Class-level counter for statistics
    _total_created: ClassVar[int] = 0

    def __post_init__(self) -> None:
        Task._total_created += 1
        # Normalize tags
        self.tags = [t.lower().strip() for t in self.tags]

    def complete(self) -> None:
        """Mark this task as done."""
        if self.status == TaskStatus.DONE:
            raise ValueError(f"Task '{self.title}' is already complete")
        self.status = TaskStatus.DONE
        self.completed_at = datetime.utcnow()
        self.updated_at = datetime.utcnow()

    def start(self) -> None:
        """Mark this task as in progress."""
        if self.status == TaskStatus.DONE:
            raise ValueError(f"Cannot start a completed task")
        self.status = TaskStatus.IN_PROGRESS
        self.updated_at = datetime.utcnow()

    def cancel(self) -> None:
        """Cancel this task."""
        if self.status == TaskStatus.DONE:
            raise ValueError(f"Cannot cancel a completed task")
        self.status = TaskStatus.CANCELLED
        self.updated_at = datetime.utcnow()

    def add_tag(self, tag: str) -> None:
        normalized = tag.lower().strip()
        if normalized not in self.tags:
            self.tags.append(normalized)
            self.updated_at = datetime.utcnow()

    def to_dict(self) -> dict:
        """Serialize to a JSON-compatible dictionary."""
        return {
            "id": self.id,
            "title": self.title,
            "description": self.description,
            "priority": self.priority.value,
            "status": self.status.value,
            "tags": self.tags,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat(),
            "completed_at": self.completed_at.isoformat() if self.completed_at else None,
        }

    @classmethod
    def from_dict(cls, data: dict) -> Task:
        """Deserialize from a dictionary (loaded from JSON)."""
        task = cls(
            title=data["title"],
            description=data.get("description", ""),
            priority=Priority(data.get("priority", "medium")),
            status=TaskStatus(data.get("status", "pending")),
            tags=data.get("tags", []),
        )
        task.id = data["id"]
        task.created_at = datetime.fromisoformat(data["created_at"])
        task.updated_at = datetime.fromisoformat(data["updated_at"])
        task.completed_at = (
            datetime.fromisoformat(data["completed_at"])
            if data.get("completed_at")
            else None
        )
        return task

    def __str__(self) -> str:
        status_icons = {
            TaskStatus.PENDING: "○",
            TaskStatus.IN_PROGRESS: "◐",
            TaskStatus.DONE: "●",
            TaskStatus.CANCELLED: "✗",
        }
        icon = status_icons.get(self.status, "?")
        tags_str = f" [{', '.join(self.tags)}]" if self.tags else ""
        return (
            f"{icon} [{self.id}] {self.title} "
            f"({self.priority.value}){tags_str}"
        )

Step 2: Custom Exceptions

# exceptions.py
class TaskManagerError(Exception):
    """Base error for all task manager errors."""

class TaskNotFoundError(TaskManagerError):
    def __init__(self, task_id: str) -> None:
        self.task_id = task_id
        super().__init__(f"Task not found: {task_id!r}")

class DuplicateTaskError(TaskManagerError):
    def __init__(self, title: str) -> None:
        super().__init__(f"A task with title {title!r} already exists")

class StorageError(TaskManagerError):
    """Raised when reading or writing to the storage file fails."""

Step 3: Storage Layer (`storage.py`)

# storage.py
from __future__ import annotations

import json
from pathlib import Path

from models import Task
from exceptions import StorageError


class JSONStorage:
    """Persists tasks as JSON to a file on disk."""

    def __init__(self, file_path: str | Path = "tasks.json") -> None:
        self.path = Path(file_path)

    def load(self) -> list[Task]:
        """Load all tasks from the JSON file."""
        if not self.path.exists():
            return []
        try:
            raw = self.path.read_text(encoding="utf-8")
            data = json.loads(raw)
            return [Task.from_dict(item) for item in data]
        except json.JSONDecodeError as e:
            raise StorageError(f"Corrupted tasks file ({self.path}): {e}") from e
        except (KeyError, ValueError) as e:
            raise StorageError(f"Invalid task data in {self.path}: {e}") from e

    def save(self, tasks: list[Task]) -> None:
        """Save all tasks to the JSON file (overwrites)."""
        try:
            data = [task.to_dict() for task in tasks]
            self.path.write_text(
                json.dumps(data, indent=2, ensure_ascii=False),
                encoding="utf-8",
            )
        except OSError as e:
            raise StorageError(f"Failed to write tasks to {self.path}: {e}") from e

    def backup(self) -> Path:
        """Create a timestamped backup of the tasks file."""
        from datetime import datetime
        if not self.path.exists():
            raise StorageError("Nothing to back up — tasks file does not exist")
        timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
        backup_path = self.path.with_name(f"tasks_backup_{timestamp}.json")
        backup_path.write_bytes(self.path.read_bytes())
        return backup_path

Step 4: Business Logic (`manager.py`)

# manager.py
from __future__ import annotations

from models import Task, TaskStatus, Priority
from storage import JSONStorage
from exceptions import TaskNotFoundError, DuplicateTaskError


class TaskManager:
    """Manages the task collection — create, find, update, delete."""

    def __init__(self, storage: JSONStorage | None = None) -> None:
        self._storage = storage or JSONStorage()
        self._tasks: list[Task] = self._storage.load()

    def _save(self) -> None:
        self._storage.save(self._tasks)

    def _find(self, task_id: str) -> Task:
        for task in self._tasks:
            if task.id == task_id:
                return task
        raise TaskNotFoundError(task_id)

    def add(
        self,
        title: str,
        description: str = "",
        priority: Priority = Priority.MEDIUM,
        tags: list[str] | None = None,
    ) -> Task:
        """Create and save a new task."""
        titles_lower = {t.title.lower() for t in self._tasks}
        if title.lower() in titles_lower:
            raise DuplicateTaskError(title)
        task = Task(title=title, description=description, priority=priority, tags=tags or [])
        self._tasks.append(task)
        self._save()
        return task

    def get(self, task_id: str) -> Task:
        return self._find(task_id)

    def list_tasks(
        self,
        status: TaskStatus | None = None,
        priority: Priority | None = None,
        tag: str | None = None,
    ) -> list[Task]:
        """Return filtered list of tasks."""
        results = self._tasks
        if status:
            results = [t for t in results if t.status == status]
        if priority:
            results = [t for t in results if t.priority == priority]
        if tag:
            results = [t for t in results if tag.lower() in t.tags]
        return sorted(results, key=lambda t: (t.priority.value, t.created_at), reverse=True)

    def complete(self, task_id: str) -> Task:
        task = self._find(task_id)
        task.complete()
        self._save()
        return task

    def start(self, task_id: str) -> Task:
        task = self._find(task_id)
        task.start()
        self._save()
        return task

    def delete(self, task_id: str) -> Task:
        task = self._find(task_id)
        self._tasks.remove(task)
        self._save()
        return task

    def stats(self) -> dict:
        total = len(self._tasks)
        by_status = {s.value: 0 for s in TaskStatus}
        for task in self._tasks:
            by_status[task.status.value] += 1
        completion_rate = (by_status["done"] / total * 100) if total else 0
        return {
            "total": total,
            "by_status": by_status,
            "completion_rate": round(completion_rate, 1),
        }

Step 5: CLI Entry Point (`main.py`)

# main.py
from __future__ import annotations

import sys
from models import TaskStatus, Priority
from manager import TaskManager
from exceptions import TaskManagerError


def print_usage() -> None:
    print("""
Task Manager CLI — Commands:
  add <title> [--desc <description>] [--priority low|medium|high|critical] [--tags tag1,tag2]
  list [--status pending|in_progress|done|cancelled] [--priority ...] [--tag <tag>]
  complete <task_id>
  start <task_id>
  delete <task_id>
  stats
  backup
""")


def parse_flags(args: list[str]) -> dict[str, str]:
    """Parse --key value pairs from argument list."""
    flags: dict[str, str] = {}
    i = 0
    while i < len(args):
        if args[i].startswith("--") and i + 1 < len(args):
            flags[args[i][2:]] = args[i + 1]
            i += 2
        else:
            i += 1
    return flags


def main(argv: list[str] | None = None) -> int:
    if argv is None:
        argv = sys.argv[1:]

    if not argv:
        print_usage()
        return 0

    manager = TaskManager()
    command = argv[0]
    rest = argv[1:]
    flags = parse_flags(rest)

    try:
        if command == "add":
            if not rest or rest[0].startswith("--"):
                print("Error: Please provide a task title.")
                return 1
            title = rest[0]
            priority = Priority(flags.get("priority", "medium"))
            tags = [t.strip() for t in flags.get("tags", "").split(",") if t.strip()]
            task = manager.add(
                title=title,
                description=flags.get("desc", ""),
                priority=priority,
                tags=tags,
            )
            print(f"Created: {task}")
            return 0

        elif command == "list":
            status_filter = TaskStatus(flags["status"]) if "status" in flags else None
            priority_filter = Priority(flags["priority"]) if "priority" in flags else None
            tag_filter = flags.get("tag")
            tasks = manager.list_tasks(status=status_filter, priority=priority_filter, tag=tag_filter)
            if not tasks:
                print("No tasks found.")
            else:
                print(f"\nFound {len(tasks)} task(s):\n")
                for task in tasks:
                    print(f"  {task}")
            return 0

        elif command == "complete":
            if not rest:
                print("Error: Please provide a task ID.")
                return 1
            task = manager.complete(rest[0])
            print(f"Completed: {task}")
            return 0

        elif command == "start":
            if not rest:
                print("Error: Please provide a task ID.")
                return 1
            task = manager.start(rest[0])
            print(f"Started: {task}")
            return 0

        elif command == "delete":
            if not rest:
                print("Error: Please provide a task ID.")
                return 1
            task = manager.delete(rest[0])
            print(f"Deleted: {task}")
            return 0

        elif command == "stats":
            s = manager.stats()
            print(f"\nTotal tasks: {s['total']}")
            for status, count in s["by_status"].items():
                print(f"  {status}: {count}")
            print(f"Completion rate: {s['completion_rate']}%")
            return 0

        elif command == "backup":
            path = manager._storage.backup()
            print(f"Backup created: {path}")
            return 0

        else:
            print(f"Unknown command: {command!r}")
            print_usage()
            return 1

    except (ValueError, TaskManagerError) as e:
        print(f"Error: {e}")
        return 1


if __name__ == "__main__":
    sys.exit(main())

Testing Your CLI

Once you have all five files in a folder, test the application:

# Add tasks
python main.py add "Write unit tests" --priority high --tags python,testing
python main.py add "Update documentation" --desc "Add examples to README" --tags docs
python main.py add "Fix login bug" --priority critical --tags python,backend,auth

# List all tasks
python main.py list

# List only high-priority tasks
python main.py list --priority high

# Start and complete a task (use the ID shown in list output)
python main.py start <task_id>
python main.py complete <task_id>

# View statistics
python main.py stats

# Back up your tasks
python main.py backup

Extending the Project

Now that the core is working, here are challenges to practice further:

  1. Due dates: Add a due_date: date | None field to Task and a --overdue filter to list.
  2. Subtasks: Allow tasks to have a list of subtask IDs, with progress tracking.
  3. Export: Add a python main.py export --format csv command that writes a CSV report.
  4. Search: Add python main.py search <query> that searches titles and descriptions.
  5. FastAPI frontend: Wrap TaskManager in a FastAPI app (Lesson 14) to expose a REST API.
  6. Unit tests: Write tests for TaskManager using pytest and temporary JSON files.

tip type: tip title: "You have completed Python Essentials!"

You have covered the entire Python language — from variables and control flow to OOP, file I/O, HTTP, FastAPI, decorators, and type hints. The next step is to specialize: dive deep into data science with pandas and NumPy, build production APIs with FastAPI and databases, or explore async Python with asyncio. The foundation you have built here will serve you in all of these directions.

nextSteps: []