CLI and Automation

What we're doing: Build a two-subcommand CLI (migrate, status) where each subcommand owns its own arguments and dispatches to its own function.

import argparse

def cmd_migrate(args):
    return f"migrating to {args.target}"

def cmd_status(args):
    return "status: ok"

parser = argparse.ArgumentParser(prog="dbtool")
sub = parser.add_subparsers(dest="command", required=True)

migrate_p = sub.add_parser("migrate", help="apply pending migrations")
migrate_p.add_argument("--target", default="latest")
migrate_p.set_defaults(func=cmd_migrate)

status_p = sub.add_parser("status", help="show migration status")
status_p.set_defaults(func=cmd_status)

args = parser.parse_args(["migrate", "--target", "0007"])
print(args.func(args))

args2 = parser.parse_args(["status"])
print(args2.func(args2))

9

add_subparsers(dest="command", required=True) turns dbtool into a program that needs a subcommand — dbtool migrate, dbtool status.

10

sub.add_parser("migrate") creates a full ArgumentParser scoped to migrate; --target added here does not exist under status.

13

set_defaults(func=cmd_migrate) attaches the handler directly to the parsed args — no if args.command == "migrate": chain needed.

18

args.func(args) calls whichever handler matched the subcommand that was actually given.

migrating to 0007
status: ok

Why this works: Each subcommand gets its own ArgumentParser scoped under add_subparsers(), so migrate can require --target while status needs no arguments at all — and set_defaults(func=...) lets parse_args() itself carry the correct handler, replacing a manual if/elif dispatch on args.command.

What we're doing: Read a required setting from the environment with a fallback, and exit with a distinct code depending on how a subprocess run finishes.

import os
import sys
import subprocess

os.environ["APP_ENV"] = "production"
env = os.environ.get("APP_ENV", "development")
print("env:", env)

missing = os.environ.get("MISSING_VAR", "default-value")
print("missing:", missing)

ok = subprocess.run([sys.executable, "-c", "import sys; sys.exit(0)"])
print("ok exit code:", ok.returncode)

failed = subprocess.run([sys.executable, "-c", "import sys; sys.exit(1)"])
print("fail exit code:", failed.returncode)

6

os.environ.get("APP_ENV", "development") returns the real value when set — "production" here, since line 5 set it.

9

MISSING_VAR was never set, so .get() returns "default-value" instead of raising KeyError.

12

A subprocess that exits cleanly (sys.exit(0)) reports returncode 0 back to the parent — the standard success signal.

15

A subprocess that calls sys.exit(1) reports returncode 1 — the parent can branch on this without reading any output.

env: production
missing: default-value
ok exit code: 0
fail exit code: 1

Why this works: os.environ.get() with a default is the safe read for optional configuration; every subprocess or script that finishes runs sys.exit(code) implicitly or explicitly, and that integer — not anything printed — is what a shell, CI pipeline, or cron wrapper actually inspects to decide what happened.

What we're doing: Capture one subprocess's stdout, filter it in Python (replacing a shell pipeline), and use check=True to catch a failing step.

import subprocess
import sys

listing = subprocess.run(
    [sys.executable, "-c", "print('apple\nbanana\navocado\ncherry')"],
    capture_output=True, text=True,
)
lines = listing.stdout.splitlines()
filtered = [line for line in lines if line.startswith("a")]
print(filtered)

try:
    subprocess.run([sys.executable, "-c", "import sys; sys.exit(2)"], check=True)
except subprocess.CalledProcessError as e:
    print("caught:", e.returncode)

4

The first subprocess prints four lines; capture_output=True, text=True captures them as one string, not to the terminal.

8

listing.stdout.splitlines() turns the captured output into a list of lines, exactly like piping into a filter command would.

9

The filtering itself happens in Python, not a shell | grep — this is the same result with no shell dependency.

12

check=True makes this call raise instead of returning a CompletedProcess with returncode=2 that nothing inspects.

['apple', 'avocado']
caught: 2

Why this works: Capturing stdout as text and processing it with normal Python (splitlines, a list comprehension) replaces a shell pipeline's | without spawning a shell — and check=True converts a silently-ignored nonzero exit code into a Python exception the script is forced to handle or crash on, matching how a well-written automation script should fail.

What we're doing: Write a script designed to be invoked by cron: it logs instead of printing, is safe to re-run, and reports failure through its exit code.

import logging
import sys
from io import StringIO

log_stream = StringIO()
logger = logging.getLogger("cleanup_script")
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler(log_stream))

processed = 0
errors = 0
items = ["file1.tmp", "file2.tmp", "bad.lock"]
for item in items:
    if item.endswith(".tmp"):
        processed += 1
        logger.info("removed %s", item)
    else:
        errors += 1
        logger.warning("skipped %s", item)

print(f"processed={processed} errors={errors}")
exit_code = 0 if errors == 0 else 1
print("exit_code:", exit_code)

6

A named logger, not print(), so a cron-invoked run leaves a record even with no terminal attached — cron typically redirects stdout to /dev/null or a mail digest.

15

Deleting only .tmp files, keyed off a stable condition, is what makes a second identical run safe — it does not re-delete or double-count anything.

21

The script computes its own exit code from what actually happened, rather than always exiting 0 regardless of outcome.

processed=2 errors=1
exit_code: 1

Why this works: cron reads only the exit code to decide whether the job succeeded — exit_code = 0 if errors == 0 else 1 makes that decision reflect what actually happened, instead of a script that always exits 0 because nothing computed otherwise. The logger records what a human reviewing later needs, since nothing was watching the terminal at run time.

What we're doing: Write a maintenance script that logs each action, tracks a processed/errors count, and computes its exit code from that count.

import logging
from io import StringIO

log_stream = StringIO()
logger = logging.getLogger("cleanup_script")
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler(log_stream))

processed = 0
errors = 0
items = ["file1.tmp", "file2.tmp", "bad.lock"]
for item in items:
    if item.endswith(".tmp"):
        processed += 1
        logger.info("removed %s", item)
    else:
        errors += 1
        logger.warning("skipped %s", item)

print(f"processed={processed} errors={errors}")
exit_code = 0 if errors == 0 else 1
print("exit_code:", exit_code)

12

Only files ending in .tmp are removed — a narrow, explicit condition rather than "everything in this directory".

14

Every removal is logged individually, so a later review can see exactly which files were touched.

16

An item that does not match is counted as skipped and logged as a warning, not silently ignored.

19

The exit code is derived from the errors count computed during the run, not hardcoded.

processed=2 errors=1

Why this works: Tracking processed/errors counters while iterating, rather than just performing the actions, is what turns "the script ran" into "the script did N things and skipped M" — the difference a human reviewing a log actually needs to trust an unattended run.

What we're doing: Split ten records into fixed-size batches of three and process each batch, tracking a running total.

def chunked(iterable, size):
    chunk = []
    for item in iterable:
        chunk.append(item)
        if len(chunk) == size:
            yield chunk
            chunk = []
    if chunk:
        yield chunk

records = list(range(1, 11))
batches = list(chunked(records, 3))
total_processed = 0
for i, batch in enumerate(batches, start=1):
    total_processed += len(batch)
    print(f"batch {i}: {batch} -> processed {len(batch)}")
print("total_processed:", total_processed)

1

chunked() is a generator — it yields one batch at a time instead of building the full list of batches in memory first.

12

10 records split into batches of 3 gives four batches: three full ones and one final partial batch of 1.

13

Each batch is processed and logged independently — a failure on batch 3 would still leave batches 1 and 2 done.

batch 1: [1, 2, 3] -> processed 3
batch 2: [4, 5, 6] -> processed 3
batch 3: [7, 8, 9] -> processed 3
batch 4: [10] -> processed 1
total_processed: 10

Why this works: chunked() as a generator means only one batch (at most `size` items) is ever in memory at a time, regardless of how large `records` grows — and processing each batch as its own unit of work means the running total, and any failure, is scoped to one batch instead of the entire input.

What we're doing: Check a required tool's version, then gate the next deployment step on a prior check's exit code.

import subprocess
import sys

result = subprocess.run([sys.executable, "--version"], capture_output=True, text=True)
print(result.stdout.strip() or result.stderr.strip())

# simulate a pre-deploy check: gate on its exit code, not its text
check = subprocess.run([sys.executable, "-c", "import sys; sys.exit(0)"])
if check.returncode == 0:
    print("check passed, proceeding")
else:
    print("check failed, aborting")

4

Confirming the tool's version before running steps that assume it — the same pattern as checking a required Python or build-tool version in CI.

8

The pre-deploy check runs as its own subprocess; only check.returncode is inspected, not any text it printed.

9

The deployment logic branches directly on the exit code — 0 means proceed, anything else means abort.

Python 3.14.3
check passed, proceeding

Why this works: Gating on check.returncode == 0 rather than scanning stdout for a string like "PASSED" is what keeps the script correct if the wrapped tool ever changes its wording — the exit code is the one part of a CLI tool's interface meant to be depended on by callers.