True A→Z → PhD-level Python Mastery Course Part2

Perfect choice.
We’ll now build a real, production-grade CLI tool step by step, using everything you learned across ALL phases.

This will not feel like a tutorial — it will feel like building a small framework.


🛠️ CAPSTONE PROJECT 1 — PROFESSIONAL CLI TOOL

📌 “TaskRunner” (Plugin-based CLI like git / pytest)

What you’re building
A CLI tool that:

  • Has subcommands (taskrunner clean, taskrunner build)
  • Discovers commands via plugins
  • Has clean package structure
  • Uses logging, exceptions, testing
  • Is installable via pip
  • Is interview-defensible at senior level

🧭 STEP 0 — WHAT YOU’LL BE ABLE TO SAY IN INTERVIEWS

“I built a plugin-based CLI framework in Python using dynamic imports, decorators for command registration, structured logging, pytest, and packaged it using pyproject.toml with entry points.”

This sentence alone filters you into strong Python engineer territory.


📁 STEP 1 — PROJECT STRUCTURE (NON-NEGOTIABLE)

We use src layout (Phase 7).

taskrunner/
├── pyproject.toml
├── README.md
├── src/
│   └── taskrunner/
│       ├── __init__.py
│       ├── __main__.py
│       ├── cli.py
│       ├── registry.py
│       ├── logging_config.py
│       ├── errors.py
│       ├── plugins/
│       │   ├── __init__.py
│       │   ├── clean.py
│       │   └── build.py
└── tests/
    ├── test_registry.py
    └── test_cli.py

📌 Phase mapping

  • Phase 7 → imports & packages
  • Phase 5 → OOP boundaries
  • Phase 10 → testing
  • Phase 12 → packaging

🧱 STEP 2 — CUSTOM EXCEPTIONS (PHASE 8)

errors.py

class TaskRunnerError(Exception):
    """Base exception for taskrunner"""

class CommandNotFound(TaskRunnerError):
    pass

Why this matters:

  • Clean error hierarchy
  • Targeted handling
  • Interview-grade design

🧠 STEP 3 — COMMAND REGISTRY (PLUGIN CORE)

This is the heart of the system.

registry.py

from typing import Callable, Dict

_COMMANDS: Dict[str, Callable] = {}

def register(name: str):
    def decorator(func: Callable):
        if name in _COMMANDS:
            raise ValueError(f"Command '{name}' already registered")
        _COMMANDS[name] = func
        return func
    return decorator

def get_command(name: str) -> Callable:
    return _COMMANDS.get(name)

def all_commands():
    return dict(_COMMANDS)

📌 Uses:

  • Decorators (Phase 4)
  • Module-level state (Phase 7)
  • Clean API

🔌 STEP 4 — PLUGIN COMMANDS (PHASE 7.3)

plugins/clean.py

from taskrunner.registry import register

@register("clean")
def clean():
    print("Cleaning build artifacts...")

plugins/build.py

from taskrunner.registry import register

@register("build")
def build():
    print("Building project...")

📌 Key insight
Plugins register themselves at import time.

This is how:

  • pytest
  • Django
  • Click
    work internally.

🔍 STEP 5 — PLUGIN DISCOVERY (DYNAMIC IMPORTS)

cli.py

import importlib
import pkgutil
from taskrunner import plugins

def load_plugins():
    for _, module_name, _ in pkgutil.iter_modules(plugins.__path__):
        importlib.import_module(f"{plugins.__name__}.{module_name}")

Why this is powerful:

  • Zero hard-coding
  • New command = new file
  • Core never changes

📌 Phase 7.3 mastery


🧪 STEP 6 — CLI ENTRY LOGIC

cli.py (continued)

import sys
from taskrunner.registry import get_command
from taskrunner.errors import CommandNotFound

def main():
    load_plugins()

    if len(sys.argv) < 2:
        print("Available commands:")
        for cmd in sorted(all_commands()):
            print(f"  {cmd}")
        sys.exit(1)

    cmd_name = sys.argv[1]
    cmd = get_command(cmd_name)

    if not cmd:
        raise CommandNotFound(f"Unknown command: {cmd_name}")

    cmd()

📌 Uses:

  • Argument parsing (simple, intentional)
  • Registry lookup
  • Custom exceptions

🚦 STEP 7 — __main__.py (PHASE 7 MUST-HAVE)

__main__.py

from taskrunner.cli import main

if __name__ == "__main__":
    main()

Now you can run:

python -m taskrunner clean

📜 STEP 8 — LOGGING (PHASE 10)

logging_config.py

import logging

def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s %(message)s"
    )

Call it inside main().

📌 Shows production awareness.


🧪 STEP 9 — TESTING (PHASE 10)

tests/test_registry.py

from taskrunner.registry import register, get_command

def test_register_and_get():
    @register("test")
    def fn(): pass

    assert get_command("test") == fn

Why this test matters:

  • Tests core behavior
  • No mocking needed
  • Fast & deterministic

📦 STEP 10 — PACKAGING (PHASE 12)

pyproject.toml

[project]
name = "taskrunner"
version = "0.1.0"
dependencies = []

[project.scripts]

taskrunner = “taskrunner.cli:main”

Install locally:

pip install -e .

Run:

taskrunner clean

🔥 This is huge — you’ve built a real CLI.


🧠 STEP 11 — SECURITY & ROBUSTNESS

  • No eval
  • No shell calls
  • Controlled imports
  • Clear error handling
  • Safe defaults

You can now talk about:

  • Attack surface
  • Plugin isolation
  • Controlled execution

🏁 WHAT YOU HAVE BUILT (RECAP)

✅ Plugin-based CLI framework
✅ Dynamic imports
✅ Decorator registration
✅ Clean package layout
✅ Custom exceptions
✅ Logging
✅ Tests
✅ Installable CLI

This is not beginner work.


🚀 NEXT STEP (CHOOSE ONE)

Reply with ONLY ONE number:

1 → Extend CLI with config files (JSON/YAML)
2 → Add argument parsing per command
3 → Add retry + timing decorators
4 → Dockerize + CI pipeline
5 → Move to ETL Pipeline project

We’ll keep building this into something you’d proudly show in interviews.

Perfect — this is exactly the right order.
We’ll evolve TaskRunner into a serious, production-grade CLI, then move cleanly to the ETL pipeline.

I’ll proceed step by step, each step building on the previous one, no shortcuts, no phase violations.


🛠️ TASKRUNNER — ADVANCED EXTENSIONS (STEPS 1 → 4)


✅ STEP 1 — CONFIG FILE SUPPORT (JSON / YAML)

🎯 Goal

Allow users to configure TaskRunner without changing code.

Example:

taskrunner build --config config.json

1.1 Config Design (Production Pattern)

Supported formats

  • JSON (mandatory)
  • YAML (optional, if installed)

Config example (config.json)

{
  "logging": {
    "level": "INFO"
  },
  "build": {
    "optimize": true
  }
}

1.2 Config Loader (Phase 9 + 12)

config.py

import json
from pathlib import Path

class ConfigError(Exception):
    pass

def load_config(path: str) -> dict:
    p = Path(path)

    if not p.exists():
        raise ConfigError(f"Config file not found: {path}")

    if p.suffix == ".json":
        return json.loads(p.read_text())
    
    if p.suffix in (".yaml", ".yml"):
        try:
            import yaml
            return yaml.safe_load(p.read_text())
        except ImportError:
            raise ConfigError("YAML support not installed")

    raise ConfigError("Unsupported config format")

📌 Why this is senior-grade

  • Safe parsing
  • Explicit errors
  • Optional dependency handling

1.3 Inject Config into CLI

Modify cli.py:

from taskrunner.config import load_config

def main():
    load_plugins()

    config = {}
    if "--config" in sys.argv:
        idx = sys.argv.index("--config")
        config = load_config(sys.argv[idx + 1])

Now config becomes a first-class citizen.


✅ STEP 2 — ARGUMENT PARSING PER COMMAND

We intentionally avoid argparse magic for clarity.


2.1 Command Signature Upgrade

Change registry to store callables that accept args + config.

registry.py

from typing import Callable, Dict

_COMMANDS: Dict[str, Callable] = {}

def register(name: str):
    def decorator(func: Callable):
        _COMMANDS[name] = func
        return func
    return decorator

def get_command(name: str):
    return _COMMANDS.get(name)

2.2 Modify Plugin Commands

plugins/build.py

from taskrunner.registry import register

@register("build")
def build(args, config):
    optimize = "--optimize" in args or config.get("build", {}).get("optimize")
    print(f"Building project | optimize={optimize}")

2.3 Dispatch Arguments

cli.py

cmd = get_command(cmd_name)
cmd(sys.argv[2:], config)

📌 This design:

  • Avoids global state
  • Makes testing trivial
  • Allows per-command flags

✅ STEP 3 — RETRY & TIMING DECORATORS (Phase 4 + 8 + 11)


3.1 Timing Decorator

decorators.py

import time
from functools import wraps

def timed(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f"[TIME] {func.__name__}: {time.time() - start:.3f}s")
        return result
    return wrapper

3.2 Retry Decorator

def retry(times=3):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(times):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if i == times - 1:
                        raise
                    print(f"Retrying after error: {e}")
        return wrapper
    return decorator

3.3 Apply to Commands

plugins/build.py

from taskrunner.decorators import retry, timed

@register("build")
@retry(2)
@timed
def build(args, config):
    ...

📌 Interview gold

Cross-cutting concerns handled via decorators.


✅ STEP 4 — DOCKER + CI PIPELINE (Phase 12)


4.1 Dockerfile (Minimal & Secure)

FROM python:3.11-slim

WORKDIR /app
COPY pyproject.toml .
RUN pip install .

COPY src/ src/
CMD ["taskrunner"]

4.2 GitHub Actions CI

.github/workflows/ci.yml

name: CI

on: [push]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: "3.11"
      - run: pip install -e .
      - run: pytest

📌 Shows:

  • Automation
  • Reproducibility
  • Production readiness

🏁 TASKRUNNER — FINAL STATE

You now have:

✅ Plugin-based CLI
✅ Config file support
✅ Per-command arguments
✅ Retry & timing decorators
✅ Logging-ready
✅ Tested core
✅ Dockerized
✅ CI-enabled

This is far beyond a toy CLI.


🚀 NEXT — MOVE TO PROJECT 5 (ETL PIPELINE)

We now switch context cleanly.

📌 NEXT PROJECT

PROJECT 5 — PRODUCTION-GRADE ETL PIPELINE

What we’ll build:

  • Extract CSV / JSON (streaming)
  • Transform via pluggable steps
  • Load atomically
  • Retry & error handling
  • Metrics & logging
  • Tested & Dockerized

🔥 NEXT STEP (CONFIRM)

Reply with ONLY ONE word:

START

And we’ll begin ETL Pipeline — Step 0 (Architecture & Design) immediately.