Project: Building a Production-Ready CLI Tool

TIP

Learning Objectives: In this capstone project, you'll apply everything you've learned by building a complete CLI application with async operations, proper testing, type hints, design patterns, and professional packaging.

Project Overview: TaskFlow

We'll build TaskFlow, a task management CLI tool that demonstrates all the advanced Python concepts from this course.

Features to Implement

TaskFlow is a CLI task manager. Here is what we'll build and the advanced concepts each piece exercises.

Features:

1. Add, list, update, delete tasks
2. Task priorities and due dates
3. Async data persistence (JSON/SQLite)
4. Plugin system for exporters
5. Rich terminal output
6. Proper error handling
7. Full test coverage
8. Packaged for distribution

Concepts applied:

Project structure:

COPY
taskflow/
├── pyproject.toml
├── README.md
├── src/
│   └── taskflow/
│       ├── __init__.py
│       ├── __main__.py
│       ├── cli.py          # Click commands
│       ├── models.py       # Task dataclass
│       ├── storage.py      # Async storage
│       ├── plugins/        # Plugin system
│       └── utils.py        # Utilities
└── tests/
    ├── conftest.py
    ├── test_cli.py
    ├── test_models.py
    └── test_storage.py

Part 1: Core Models

Task Data Model

Task Repository (Generator Pattern)

Part 2: Async Storage

Context Manager for Storage

Part 3: Plugin System

Exporter Plugin Architecture

Part 4: CLI Interface

Click Commands with Rich Output

The CLI ties everything together with Click for command parsing and Rich for formatted output. This module wires up a real terminal program, so read it as the reference implementation rather than running it in the browser sandbox.

COPY
# src/taskflow/cli.py
"""TaskFlow CLI commands."""

import click
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from rich import box

from taskflow.models import Task, Priority, Status
from taskflow.storage import storage_context
from taskflow.plugins import PluginRegistry

console = Console()

@click.group()
@click.version_option(version="1.0.0")
@click.option("--data-file", "-f", default="~/.taskflow/tasks.json",
              help="Path to data file")
@click.pass_context
def cli(ctx, data_file):
    """TaskFlow - A powerful task management CLI."""
    ctx.ensure_object(dict)
    ctx.obj["data_file"] = data_file

@cli.command()
@click.argument("title")
@click.option("--priority", "-p", type=click.Choice(["low", "medium", "high", "urgent"]),
              default="medium", help="Task priority")
@click.option("--tag", "-t", multiple=True, help="Tags for the task")
@click.option("--due", "-d", type=click.DateTime(formats=["%Y-%m-%d"]),
              help="Due date (YYYY-MM-DD)")
@click.pass_context
async def add(ctx, title, priority, tag, due):
    """Add a new task."""
    import uuid

    async with storage_context(ctx.obj["data_file"]) as storage:
        task = Task(
            id=str(uuid.uuid4())[:8],
            title=title,
            priority=Priority[priority.upper()],
            tags=list(tag),
            due_date=due.date() if due else None,
        )
        await storage.add_task(task)

        console.print(Panel(
            f"[bold green]done[/] Task created: {task.title}\n"
            f"ID: [cyan]{task.id}[/]",
            title="New Task",
            border_style="green"
        ))

@cli.command()
@click.option("--status", "-s", type=click.Choice(["todo", "in_progress", "done"]),
              help="Filter by status")
@click.option("--priority", "-p", type=click.Choice(["low", "medium", "high", "urgent"]),
              help="Filter by priority")
@click.pass_context
async def list(ctx, status, priority):
    """List all tasks."""
    async with storage_context(ctx.obj["data_file"]) as storage:
        tasks = await storage.get_tasks()

        # Apply filters
        if status:
            tasks = [t for t in tasks if t.status.value == status]
        if priority:
            tasks = [t for t in tasks if t.priority.name.lower() == priority]

        # Display table
        table = Table(title="Tasks", box=box.ROUNDED)
        table.add_column("ID", style="cyan")
        table.add_column("Title")
        table.add_column("Priority")
        table.add_column("Status")
        table.add_column("Due")

        priority_colors = {
            "LOW": "dim", "MEDIUM": "white",
            "HIGH": "yellow", "URGENT": "red bold"
        }

        for task in sorted(tasks, key=lambda t: -t.priority.value):
            color = priority_colors[task.priority.name]
            due = str(task.due_date) if task.due_date else "-"
            table.add_row(
                task.id,
                task.title,
                f"[{color}]{task.priority.name}[/]",
                task.status.value,
                due
            )

        console.print(table)

@cli.command()
@click.argument("task_id")
@click.option("--status", "-s", type=click.Choice(["todo", "in_progress", "done"]))
@click.option("--priority", "-p", type=click.Choice(["low", "medium", "high", "urgent"]))
@click.pass_context
async def update(ctx, task_id, status, priority):
    """Update a task."""
    async with storage_context(ctx.obj["data_file"]) as storage:
        task = await storage.get_task(task_id)
        if not task:
            console.print(f"[red]Task {task_id} not found[/]")
            raise click.Abort()

        if status:
            task.status = Status(status)
        if priority:
            task.priority = Priority[priority.upper()]

        await storage.update_task(task)
        console.print(f"[green]done[/] Task {task_id} updated")

@cli.command()
@click.argument("format", type=click.Choice(["json", "csv", "markdown"]))
@click.option("--output", "-o", type=click.Path(), help="Output file")
@click.pass_context
async def export(ctx, format, output):
    """Export tasks to file."""
    async with storage_context(ctx.obj["data_file"]) as storage:
        tasks = await storage.get_tasks()

        exporter = PluginRegistry.get(format)
        content = exporter.export(tasks)

        if output:
            with open(output, "w") as f:
                f.write(content)
            console.print(f"[green]done[/] Exported to {output}")
        else:
            console.print(content)

def main():
    """Entry point."""
    import asyncio
    cli(_anyio_backend="asyncio")

if __name__ == "__main__":
    main()

Once installed, the tool is driven entirely from the shell:

COPY
# Add tasks
taskflow add "Write documentation" -p high -t docs -t important
taskflow add "Fix login bug" -p urgent --due 2024-01-15

# List tasks
taskflow list
taskflow list --status todo
taskflow list --priority high

# Update tasks
taskflow update abc123 --status done
taskflow update abc123 --priority low

# Export
taskflow export markdown -o tasks.md
taskflow export json

Part 5: Testing

Comprehensive Test Suite

A real test suite spans fixtures, model tests, async storage tests, and CLI tests. These run under pytest (with pytest-asyncio and Click's CliRunner), not in the browser, so study them as the testing blueprint for the project.

COPY
# tests/conftest.py
"""Shared test fixtures."""

import pytest
from pathlib import Path
import tempfile

from taskflow.models import Task, Priority, Status
from taskflow.storage import AsyncJSONStorage

@pytest.fixture
def sample_tasks():
    """Create sample tasks for testing."""
    return [
        Task(id="1", title="Task 1", priority=Priority.HIGH),
        Task(id="2", title="Task 2", priority=Priority.LOW, status=Status.DONE),
        Task(id="3", title="Task 3", priority=Priority.URGENT),
    ]

@pytest.fixture
def temp_storage(tmp_path):
    """Create temporary storage file."""
    return tmp_path / "test_tasks.json"

@pytest.fixture
async def storage(temp_storage):
    """Create async storage instance."""
    storage = AsyncJSONStorage(str(temp_storage))
    await storage.load()
    yield storage
    await storage.save()


# tests/test_models.py
"""Tests for Task model."""

import pytest
from datetime import date

from taskflow.models import Task, Priority, Status

class TestTask:
    def test_create_task_with_defaults(self):
        task = Task(id="1", title="Test")
        assert task.status == Status.TODO
        assert task.priority == Priority.MEDIUM

    def test_task_to_dict_and_back(self):
        task = Task(
            id="1",
            title="Test",
            priority=Priority.HIGH,
            due_date=date(2024, 12, 31)
        )
        data = task.to_dict()
        restored = Task.from_dict(data)

        assert restored.id == task.id
        assert restored.title == task.title
        assert restored.priority == task.priority
        assert restored.due_date == task.due_date

    @pytest.mark.parametrize("priority,expected_value", [
        (Priority.LOW, 1),
        (Priority.MEDIUM, 2),
        (Priority.HIGH, 3),
        (Priority.URGENT, 4),
    ])
    def test_priority_values(self, priority, expected_value):
        assert priority.value == expected_value

    def test_overdue_detection(self):
        past_date = date(2020, 1, 1)
        task = Task(id="1", title="Overdue", due_date=past_date)
        assert task.is_overdue() == True

        task.status = Status.DONE
        assert task.is_overdue() == False


# tests/test_storage.py
"""Tests for async storage."""

import pytest

@pytest.mark.asyncio
async def test_add_and_retrieve_task(storage, sample_tasks):
    task = sample_tasks[0]
    await storage.add_task(task)

    tasks = await storage.get_tasks()
    assert len(tasks) == 1
    assert tasks[0].id == task.id

@pytest.mark.asyncio
async def test_update_task(storage, sample_tasks):
    task = sample_tasks[0]
    await storage.add_task(task)

    task.status = Status.DONE
    success = await storage.update_task(task)

    assert success == True
    tasks = await storage.get_tasks()
    assert tasks[0].status == Status.DONE

@pytest.mark.asyncio
async def test_delete_task(storage, sample_tasks):
    task = sample_tasks[0]
    await storage.add_task(task)

    success = await storage.delete_task(task.id)

    assert success == True
    tasks = await storage.get_tasks()
    assert len(tasks) == 0


# tests/test_cli.py
"""Tests for CLI commands."""

from click.testing import CliRunner

@pytest.fixture
def runner():
    return CliRunner()

def test_add_task(runner, temp_storage):
    from taskflow.cli import cli

    result = runner.invoke(cli, [
        "-f", str(temp_storage),
        "add", "Test task", "-p", "high"
    ])

    assert result.exit_code == 0
    assert "Task created" in result.output

def test_list_tasks(runner, temp_storage):
    from taskflow.cli import cli

    # Add a task first
    runner.invoke(cli, ["-f", str(temp_storage), "add", "Test task"])

    result = runner.invoke(cli, ["-f", str(temp_storage), "list"])

    assert result.exit_code == 0
    assert "Test task" in result.output

Running pytest produces output like this:

COPY
tests/test_models.py::TestTask::test_create_task_with_defaults PASSED
tests/test_models.py::TestTask::test_task_to_dict_and_back PASSED
tests/test_models.py::TestTask::test_priority_values[LOW-1] PASSED
tests/test_models.py::TestTask::test_priority_values[MEDIUM-2] PASSED
tests/test_models.py::TestTask::test_priority_values[HIGH-3] PASSED
tests/test_models.py::TestTask::test_priority_values[URGENT-4] PASSED
tests/test_models.py::TestTask::test_overdue_detection PASSED
tests/test_storage.py::test_add_and_retrieve_task PASSED
tests/test_storage.py::test_update_task PASSED
tests/test_storage.py::test_delete_task PASSED
tests/test_cli.py::test_add_task PASSED
tests/test_cli.py::test_list_tasks PASSED

========================= 12 passed in 0.45s =========================

Part 6: Complete Package

Final pyproject.toml

The pyproject.toml declares build backend, metadata, dependencies, the taskflow entry point, and tool config for pytest, mypy, ruff, and coverage all in one place.

COPY
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "taskflow-cli"
version = "1.0.0"
description = "A powerful task management CLI tool"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.10"
authors = [
    {name = "Your Name", email = "you@example.com"}
]
keywords = ["cli", "tasks", "productivity", "todo"]
classifiers = [
    "Development Status :: 4 - Beta",
    "Environment :: Console",
    "Intended Audience :: Developers",
    "License :: OSI Approved :: MIT License",
    "Operating System :: OS Independent",
    "Programming Language :: Python :: 3",
    "Programming Language :: Python :: 3.10",
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
    "Topic :: Utilities",
]

dependencies = [
    "click>=8.1.0",
    "rich>=13.0.0",
    "aiofiles>=23.0.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=7.0.0",
    "pytest-asyncio>=0.21.0",
    "pytest-cov>=4.0.0",
    "mypy>=1.0.0",
    "ruff>=0.1.0",
]

[project.scripts]
taskflow = "taskflow.cli:main"

[project.urls]
Homepage = "https://github.com/user/taskflow"
Documentation = "https://taskflow.readthedocs.io"
Repository = "https://github.com/user/taskflow"

[tool.hatch.build.targets.wheel]
packages = ["src/taskflow"]

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
addopts = "-v --cov=taskflow --cov-report=term-missing"

[tool.mypy]
python_version = "3.11"
strict = true
warn_return_any = true
warn_unused_ignores = true

[tool.ruff]
line-length = 88
select = ["E", "F", "I", "N", "W", "UP"]
target-version = "py311"

[tool.ruff.isort]
known-first-party = ["taskflow"]

[tool.coverage.run]
source = ["src/taskflow"]
branch = true

[tool.coverage.report]
exclude_lines = [
    "pragma: no cover",
    "if TYPE_CHECKING:",
    "raise NotImplementedError",
]

README.md

A good README.md gives users a one-screen pitch, install command, and copy-paste examples. Here is the content TaskFlow ships with.

TaskFlow — a powerful, async task management CLI tool built with Python.

Features

• Beautiful CLI — Rich terminal output with colors and tables
• Async I/O — Non-blocking file operations
• Plugin system — Extensible export formats
• Fully tested — Comprehensive test suite
• Easy install — Published to PyPI

Installation

COPY
pip install taskflow-cli

Quick start

COPY
# Add a task
taskflow add "Write documentation" --priority high --tag docs

# List tasks
taskflow list

# Update task status
taskflow update abc123 --status done

# Export to Markdown
taskflow export markdown -o tasks.md

Adding tasks

COPY
taskflow add "Task title" \
    --priority high \
    --tag important \
    --due 2024-12-31

Listing tasks

COPY
# All tasks
taskflow list

# Filter by status
taskflow list --status todo

# Filter by priority
taskflow list --priority urgent

Exporting

COPY
# Export to JSON
taskflow export json

# Export to Markdown file
taskflow export markdown -o tasks.md

# Export to CSV
taskflow export csv -o tasks.csv

Development setup

COPY
# Clone repository
git clone https://github.com/user/taskflow.git
cd taskflow

# Create virtual environment
python -m venv .venv
source .venv/bin/activate

# Install with dev dependencies
pip install -e ".[dev]"

# Run tests
pytest

# Type checking
mypy src/taskflow

# Linting
ruff check src/

Licensed under the MIT License (see the LICENSE file).

Summary: Concepts Applied

This capstone exercised the entire Python Advanced toolkit. Here is where each concept landed in the project:

Congratulations — you've completed Python Advanced! Next steps:

1. Build your own CLI tool using these patterns
2. Contribute to open source Python projects
3. Explore the ML and NLP courses on AP Lab Academy
4. Keep practicing and building

Key Takeaways

Your Challenge

Now it's your turn! Extend TaskFlow with:

1. SQLite backend - Add database storage option
2. Due date reminders - Notification system (Observer pattern)
3. Task dependencies - Block tasks until prereqs done
4. Time tracking - Track time spent on tasks
5. Sync to cloud - Add cloud storage plugin

Congratulations on completing Python Advanced! You're now equipped to build professional-grade Python applications.

Further Resources

Modern CLI Frameworks

• Typer — by FastAPI's author. Type-hint driven, the friendliest modern CLI library. Generates Click commands under the hood.
• Click — the workhorse Python CLI library. What Flask's CLI, Black, and most pip-installable tools use.
• Rich — beautiful terminal output: colored text, progress bars, tables, syntax-highlighted code, markdown rendering. Drop-in upgrade to print.
• Textual — full TUI framework from Rich's author. Build full-screen interactive terminal apps with React-like components.
• prompt_toolkit — interactive prompts with autocomplete, syntax highlighting, history.

Application Architecture for CLIs

• Architecture Patterns with Python — Percival & Gregory (free online). Repository / Service-Layer / Unit-of-Work patterns scale beautifully to CLIs.
• pydantic-settings — type-safe config from env vars, .env files, secrets, and CLI args.

Distribution

• uv tool install — modern way to install CLI tools system-wide.
• pipx — install CLI tools in isolated environments. Pre-uv standard, still useful.
• PyInstaller — bundle your Python app + interpreter into a single executable. For users who don't have Python installed.
• Nuitka — alternative; compiles Python to optimized C.

Inspiration — Read the Source of Great CLIs

• httpie — beautiful HTTP client with rich output.
• black — clean Click usage, well-tested.
• pip — the canonical large-scale Python CLI.

Course Continuation

• ML Fundamentals — apply your Python mastery to machine learning.
• ML Advanced — production ML, MLOps, deployment.
• Build something real and ship it. Best learning happens after the courses end.