CLI Parsing and Structured Application Logging
CLI Parsing and Structured Application Logging
After reading this chapter, you will master command-line argument parsing with argparse, configure subcommands and flag validation, build production application logging using Python’s logging module, configure multiple handlers (console and rotating files), and implement structured JSON log formatters.
Mental model
Enterprise utilities require two foundational capabilities: 1. Command-Line Interface (argparse): Translates user CLI flags into typed Python namespaces with automated --help generation. 2. Logging Pipeline (logging): Routes messages through loggers, filters, formatters, and handlers:
Logging Architecture:
Logger (logger.info("request"))
│
├── Level Check (DEBUG < INFO < WARN < ERROR < CRITICAL)
│ │
│ ▼ Passed
├── Handler 1 (StreamHandler -> Console / stderr)
│ └── Formatter 1 (Colorized text: "[INFO] 10:00 - message")
│
└── Handler 2 (RotatingFileHandler -> /var/log/app.log)
└── Formatter 2 (JSON Formatter: {"level": "INFO", "msg": ...})
Minimal example
Save as cli_and_logging.py:
# cli_and_logging.py
import argparse
import logging
import sys
def configure_logger(verbose: bool) -> logging.Logger:
logger = logging.getLogger("cluster_manager")
level = logging.DEBUG if verbose else logging.INFO
logger.setLevel(level)
# Console handler
handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter(
fmt="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%dT%H:%M:%SZ"
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
prog="cluster-ctl",
description="Cluster management orchestration CLI utility."
)
parser.add_argument("--verbose", "-v", action="store_true", help="Enable verbose debug logging.")
subparsers = parser.add_subparsers(dest="command", required=True, help="Subcommand to execute.")
# Subcommand: deploy
deploy_parser = subparsers.add_parser("deploy", help="Deploy service instances.")
deploy_parser.add_argument("service", type=str, help="Target service name.")
deploy_parser.add_argument("--replicas", "-r", type=int, default=3, help="Number of replicas (default: 3).")
return parser
def main() -> None:
# Test arguments programmatically (simulating: cluster-ctl deploy auth-svc -r 5 -v)
test_args = ["--verbose", "deploy", "auth-svc", "--replicas", "5"]
parser = build_parser()
args = parser.parse_args(test_args)
logger = configure_logger(args.verbose)
logger.debug("CLI arguments parsed successfully.")
logger.info(f"Initiating deployment: service='{args.service}', replicas={args.replicas}")
if __name__ == "__main__":
main()Run via uv run python cli_and_logging.py:
2026-09-07T...Z [DEBUG] cluster_manager: CLI arguments parsed successfully.
2026-09-07T...Z [INFO] cluster_manager: Initiating deployment: service='auth-svc', replicas=5
Worked examples
Case 1: Structured JSON Logging for Observability Systems
Production observability systems (Datadog, ElasticSearch, CloudWatch) require JSON logs rather than plain text strings:
# json_logger.py
import json
import logging
import sys
from datetime import datetime, timezone
class JSONFormatter(logging.Formatter):
"""Formats log records as single-line JSON objects."""
def format(self, record: logging.LogRecord) -> str:
log_payload = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
"module": record.module,
"line": record.lineno,
}
if record.exc_info:
log_payload["exception"] = self.formatException(record.exc_info)
return json.dumps(log_payload)
def setup_json_logging() -> logging.Logger:
logger = logging.getLogger("api_gateway")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)
return logger
if __name__ == "__main__":
log = setup_json_logging()
log.info("Worker pool initialized successfully.")
try:
1 / 0
except ZeroDivisionError:
log.error("Math processing failure", exc_info=True)Run:
uv run python json_logger.pyOutput:
{"timestamp": "2026-09-07T...", "level": "INFO", "logger": "api_gateway", "message": "Worker pool initialized successfully.", "module": "json_logger", "line": 28}
{"timestamp": "2026-09-07T...", "level": "ERROR", "logger": "api_gateway", "message": "Math processing failure", "module": "json_logger", "line": 32, "exception": "Traceback (most recent call last):\n File \"json_logger.py\", line 30, in <module>\n 1 / 0\nZeroDivisionError: division by zero"}Case 2: Log File Rotation with RotatingFileHandler
Without log rotation, a logging process will eventually fill up the host filesystem. RotatingFileHandler caps file size and maintains backup generations:
# rotating_logs.py
import logging
import tempfile
from logging.handlers import RotatingFileHandler
from pathlib import Path
def test_log_rotation() -> None:
with tempfile.TemporaryDirectory() as tmpdir:
log_file = Path(tmpdir) / "service.log"
# Max size: 500 bytes, keep up to 3 backup files (service.log.1, service.log.2, etc.)
handler = RotatingFileHandler(log_file, maxBytes=500, backupCount=3)
formatter = logging.Formatter("%(asctime)s %(levelname)s: %(message)s")
handler.setFormatter(formatter)
logger = logging.getLogger("rotator")
logger.setLevel(logging.INFO)
logger.addHandler(handler)
# Write enough records to trigger multiple rotations
for i in range(25):
logger.info(f"Log event record #{i:02d} writing payload to disk buffer.")
handler.close()
# Inspect generated files
generated = sorted(Path(tmpdir).glob("service.log*"))
print(f"Generated {len(generated)} rotated files:")
for f in generated:
print(f" - {f.name} ({f.stat().st_size} bytes)")
if __name__ == "__main__":
test_log_rotation()Run:
uv run python rotating_logs.pyOutput:
Generated 4 rotated files:
- service.log (320 bytes)
- service.log.1 (504 bytes)
- service.log.2 (504 bytes)
- service.log.3 (504 bytes)
Pitfalls
Pitfall 1: Adding Duplicate Handlers
Calling logger.addHandler() multiple times (e.g. inside a request handler or helper function) attaches multiple handlers, causing every message to print 2x, 3x, or 4x:
# THE BUG:
def handle_request():
logger = logging.getLogger("app")
logger.addHandler(logging.StreamHandler()) # ADDS NEW HANDLER ON EVERY CALL!
logger.info("Request handled")
# THE FIX:
# Configure handlers ONCE at application startup, or check if handlers exist:
if not logger.handlers:
logger.addHandler(...)Pitfall 2: Using print() Instead of logging in Libraries
Libraries should never use print(). Use logging.getLogger(__name__) so consumers can control verbosity, redirect outputs, or silence messages as needed.
Exercises
- Build an
argparsecommand that accepts an input filename, an optional output filename (defaulting to stdout), and a--dry-runboolean flag. - Configure a custom logging filter (
logging.Filter) that blocks log records containing the string"SECRET_TOKEN". - Implement an
argparsecustom type function that validates that an argument is a valid IPv4 address. - Set up a logger with two handlers: outputting
INFOand above to console, andDEBUGand above to a local file.
Further reading
- Python Standard Library:
argparseandloggingmodules. - Python How-To: Logging HOWTO and Logging Cookbook.