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.tomlwith 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.