Designing a CLI tool: structure, config and exit contracts

Layered typed config validated once, static vs runtime checks, and plan/apply.

This lesson is about the shape of a Python tool that runs unattended: which module knows what, how settings from a file, the environment and the command line become one checked object, what a type checker does and does not protect you from, and why a saved plan is safer than a --dry-run flag. The example is blsync, a small tool that keeps a firewall block set in line with a reviewed blocklist. It uses only the standard library, runs the same on Ubuntu's Python 3.14.4 and upstream 3.14.7, and every expected failure ends in a one-line message and a documented exit status, never a traceback.

Refresher: "Functions, modules and exceptions" in py-sec owns classes, Exception subclasses, frozen dataclasses and decorators; "Command-line tools with argparse" owns argparse, subcommands and the basic exit codes (0 ok, 1 failure, 2 usage); "JSON, CSV, YAML and TOML" owns tomllib; "Packaging, pinning and auditing your tool" owns pyproject.toml and entry points. Here the question is architecture. Unpack the lesson files (the box at the top of this page) in your home directory and work in ~/scr-py-structure: they hold the package, the feeds in feeds/, the starting state in state/blocked.json and the broken config files used below. The terminals show the lab's deploy account; yours shows your own user.

A layout where each module has one job

blsync is a package: a directory of modules with an __init__.py. python3 -m blsync runs its __main__.py, which only hands over to the command line:

from .cli import main
 
raise SystemExit(main())

Each other module has one job, and the imports only point inward. cli.py is the edge: the only module that reads sys.argv and os.environ, prints to the terminal and picks an exit status from the table in exits.py. loader.py builds the settings, config.py holds them, plan.py is the saved plan, sync.py decides and applies the changes, and store.py reads and writes files. Nothing below cli.py calls sys.exit() or reads the environment itself; it receives values as arguments and raises exceptions. That keeps the core testable with plain function calls. store.py is in the lesson files and not walked through: it parses each blocklist line with ipaddress (so 192.0.2.10 and 192.0.2.10/32 count as the same entry), replaces files atomically with os.replace as "Files, paths and safe writes" in py-sec taught, and holds the lock you will meet in the apply step. The exceptions are defined in one place, one class per thing the caller should do next:

"""The failures blsync reports. cli.py maps each class to one exit status."""
 
 
class ConfigError(Exception):
    """The settings are unusable: fix the config, then run again (exit 78)."""
 
 
class RefusedError(Exception):
    """The blocklist, the state or the plan is malformed, stale or over a limit (exit 1)."""
 
 
class BusyError(Exception):
    """Another apply holds the lock and nothing was changed: try again later (exit 75)."""

Layered configuration, validated once

A tool that runs from cron, systemd and a laptop needs settings from several places: sensible defaults, a TOML file that an operator reviews, environment variables that a unit file or CI job sets, and flags for a one-off run. This is the file the lesson files ship:

The rule is a fixed order in which later layers override earlier ones, and one validation of the merged result:

"""The settings blsync runs with: one frozen object, checked when it is built."""
from dataclasses import dataclass
from pathlib import Path
 
from .errors import ConfigError
 
 
@dataclass(frozen=True)
class Config:
    source: Path = Path("blocklist.txt")
    state: Path = Path("state/blocked.json")
    max_changes: int = 100
    max_remove_fraction: float = 0.2
 
    def __post_init__(self) -> None:
        if self.max_changes < 1:
            raise ConfigError(f"max_changes must be at least 1, got {self.max_changes}")
        if not 0.0 <= self.max_remove_fraction <= 1.0:
            raise ConfigError("max_remove_fraction must be between 0 and 1,"
                              f" got {self.max_remove_fraction}")

Config is a frozen dataclass: after it is built, no code can change a setting halfway through a run. __post_init__ runs right after the generated __init__, so no Config with an impossible value can exist. The loader turns each layer into a dict and merges them with |, where the right-hand dict wins. Two type names in it are new: Mapping (from collections.abc) is any read-only dict-like object, so load_config accepts os.environ in production and a plain dict in a test; Any means "any type, not checked":

"""Build a Config from layers: defaults < TOML file < BLSYNC_* environment < flags."""
import tomllib
from collections.abc import Mapping
from pathlib import Path
from typing import Any
 
from .config import Config
from .errors import ConfigError
 
TYPES: dict[str, type] = {"source": Path, "state": Path,
                          "max_changes": int, "max_remove_fraction": float}
IN_TOML: dict[type, tuple[type, ...]] = {Path: (str,), int: (int,), float: (int, float)}
 
 
def from_file(path: Path) -> dict[str, Any]:
    try:
        with path.open("rb") as f:
            data = tomllib.load(f)
    except (OSError, tomllib.TOMLDecodeError) as err:
        raise ConfigError(f"{path}: {err}") from None
    for name, value in data.items():
        if name not in TYPES:
            raise ConfigError(f"{path}: unknown setting {name!r}")
        if isinstance(value, bool) or not isinstance(value, IN_TOML[TYPES[name]]):
            raise ConfigError(f"{path}: {name} must be {TYPES[name].__name__}, got {value!r}")
    typed = {name: TYPES[name](value) for name, value in data.items()}
    # a relative path in the file is relative to the file, not to where the job was started
    return {k: path.parent / v if isinstance(v, Path) else v for k, v in typed.items()}
 
 
def from_env(env: Mapping[str, str]) -> dict[str, Any]:
    out: dict[str, Any] = {}
    for name, kind in TYPES.items():
        var = "BLSYNC_" + name.upper()
        if var in env:
            try:
                out[name] = kind(env[var])  # the environment only holds strings
            except ValueError:
                raise ConfigError(f"{var} must be {kind.__name__}, got {env[var]!r}") from None
    return out
 
 
def load_config(path: Path, env: Mapping[str, str], flags: Mapping[str, Any]) -> Config:
    merged = from_file(path) | from_env(env) | {k: v for k, v in flags.items() if v is not None}
    return Config(**merged)  # validated once, on the merged result

TOML values arrive typed, so a wrong type is an error, not something to convert; bool is rejected explicitly because in Python True is also an int. The environment only holds strings, so those are converted and a failed conversion names the variable. An unknown key is an error, because a typo in a config file would otherwise be ignored silently. A relative path in the file is joined to the file's directory, so state = "state/blocked.json" means the same file whichever directory the job starts in. Now break it in every layer:

Each mistake produced one line that names where it is (the file, the variable, the setting) and exit status 78, the contract's "configuration error: fix it before running again". max_change was caught as a typo, "50" as a string where an integer belongs, the missing quote with a line and column from tomllib. The value 1.5 was a valid float from a flag, and __post_init__ still refused it. The last one is different on purpose: x for a type=float flag is a usage error, so argparse printed the usage line and exited 2 before any configuration was read.

Static types are not runtime validation

The annotations in Config (max_changes: int) document intent, and Python does not check them when the program runs. A first draft of reading one setting from the environment shows what that means:

"""A first draft of reading one setting from the environment. The bug is deliberate."""
import os
from dataclasses import dataclass
 
 
@dataclass(frozen=True)
class Limits:
    max_remove_fraction: float = 0.2
 
 
def limits_from_env() -> Limits:
    raw = os.environ.get("BLSYNC_MAX_REMOVE_FRACTION", "0.2")
    return Limits(max_remove_fraction=raw)
 
 
if __name__ == "__main__":
    limits = limits_from_env()
    print(repr(limits.max_remove_fraction), "* 2 =", repr(limits.max_remove_fraction * 2))

The dataclass accepted the string '0.2' for a float field, and doubling it repeated the text. Nothing failed; a safety limit silently became garbage. A static type checker reads the code without running it and catches this class of bug. Install mypy into a virtual environment, pinned with its dependencies (requirements-dev.txt lists each one with ==). On a fresh Ubuntu server python3 -m venv needs the python3-venv package first (sudo apt install python3-venv, as "Python and virtual environments on Ubuntu 26.04" showed). Ubuntu's archive has mypy 1.19; mypy 2.0 changed several defaults, so the version you pin decides what gets reported, and the same pin belongs in CI:

mypy found the draft's bug before it ran, with the file, the line and the two types. --strict turns on every optional check, and all 10 files of the package passed it. But look back at BLSYNC_MAX_REMOVE_FRACTION=twenty: mypy cannot know what an operator will export, and tomllib.load() returns dict[str, Any], a type that tells mypy to stop checking. Static types check your code against itself; runtime validation checks the outside world against your rules. A production tool needs both, and every boundary where Any enters (a file, the environment, a saved plan) is where the runtime checks go. On 3.14, annotations are evaluated lazily (PEP 649), which changes nothing for dataclasses or mypy here.

Plan, review, apply

A --dry-run flag computes the changes, prints them and throws them away; the real run computes them again, from inputs that may have changed since the review. blsync splits the work into two commands. plan saves the exact changes together with the sha256 of the state they were computed against; apply executes that saved plan only if the state still has the same digest. The plan is a frozen dataclass. Its load is a @classmethod: you call it on the class (Plan.load(path)), it receives the class itself as cls, and cls(...) builds the instance, which is the usual way to give a class a second constructor:

"""A plan: the exact changes, and the sha256 of the state they were computed against."""
import hashlib
import json
from dataclasses import asdict, dataclass
from pathlib import Path
 
from .errors import RefusedError
from .store import entry_list, write_json
 
 
@dataclass(frozen=True)
class Plan:
    add: list[str]
    remove: list[str]
    state_sha256: str
 
    def save(self, path: Path) -> None:
        write_json(path, asdict(self))  # atomic, like the state file
 
    @classmethod
    def load(cls, path: Path, approved_sha256: str | None = None) -> "Plan":
        """plan.json is input: whoever could write it after the review chose what it says."""
        raw = path.read_bytes()
        if approved_sha256 and hashlib.sha256(raw).hexdigest() != approved_sha256.lower():
            raise RefusedError(f"{path} is not the reviewed plan: its sha256 differs")
        try:
            data = json.loads(raw)
        except ValueError as err:
            raise RefusedError(f"{path}: not a plan: {err}") from None
        if not isinstance(data, dict) or sorted(data) != ["add", "remove", "state_sha256"]:
            raise RefusedError(f"{path}: not a plan: expected add, remove and state_sha256")
        # the same rule the blocklist and the state pass: canonical addresses, nothing else
        return cls(entry_list(data["add"], f"{path}: add"),
                   entry_list(data["remove"], f"{path}: remove"),
                   str(data["state_sha256"]))

sync.py decides and applies. The safety limits live in one function, check_limits, and both steps call it:

"""Decide the changes, and apply a saved plan. The same limits guard both steps."""
from .config import Config
from .errors import RefusedError
from .plan import Plan
from .store import locked, read_blocklist, read_state, write_json
 
 
def check_limits(add: list[str], remove: list[str], current: set[str], cfg: Config) -> None:
    changes = len(add) + len(remove)
    if changes > cfg.max_changes:
        raise RefusedError(f"{changes} changes, max_changes is {cfg.max_changes}")
    if current and len(remove) / len(current) > cfg.max_remove_fraction:
        raise RefusedError(f"would remove {len(remove)} of {len(current)} entries,"
                           f" max_remove_fraction is {cfg.max_remove_fraction}")
 
 
def make_plan(cfg: Config) -> Plan:
    wanted = read_blocklist(cfg.source)
    current, digest = read_state(cfg.state)
    add, remove = sorted(wanted - current), sorted(current - wanted)
    check_limits(add, remove, current, cfg)
    return Plan(add, remove, digest)
 
 
def apply_plan(plan: Plan, cfg: Config) -> None:
    with locked(cfg.state):  # no other writer between the digest check and the replace
        current, digest = read_state(cfg.state)
        if digest != plan.state_sha256:
            raise RefusedError(f"{cfg.state} changed after the plan was made; run plan again")
        check_limits(plan.add, plan.remove, current, cfg)  # plan.json may have been edited
        write_json(cfg.state, sorted((current - set(plan.remove)) | set(plan.add)))

The state file is a JSON list of what is blocked now, a stand-in for the firewall set a real tool would read. Make a plan and review it:

The state holds 10 entries, and the plan adds two and removes one. plan.json is what a reviewer approves, and sha256sum fingerprints exactly what they read. The reviewer records that digest in the approval itself (the ticket, the merge request, the pipeline's approval step), not in a file next to the plan, where whoever can change one can change both. make_plan also refuses changes that look like an accident. A truncated feed download is the classic one:

BLSYNC_SOURCE pointed at a feed that had lost most of its lines, and the plan would have unblocked 7 of 10 entries; the max_remove_fraction of 0.2 from blsync.toml refused it. In the second command the environment said 0.9 and the flag said 0.5, and the message shows 0.5: the flag won. Both variables were set on the command line in front of python3, so they applied to that one command and do not stay in your shell. A malformed address stopped the plan with the file and the line. All three exited 1 (a human has to look at the feed before anything changes), and none of them touched the reviewed plan.json: a refused plan is never saved.

A saved plan is input too

Between review and apply, plan.json sits in a directory or travels as a CI artifact from the plan job to the apply job. Whoever can write it there decides what apply does, so apply must treat it like any other input. tamper.py edits a plan the way a careless hand or a compromised pipeline step could:

"""Edit plan.json after the review, the way a careless hand or a compromised CI step could."""
import json
import sys
from pathlib import Path
 
plan_file = Path("plan.json")
plan = json.loads(plan_file.read_text())
how = sys.argv[1] if len(sys.argv) > 1 else ""
if how == "string":          # a bare string where a list belongs
    plan["add"] = "203.0.113.250"
elif how == "unblock-all":   # well-formed: remove every entry the state holds
    plan["remove"] = json.loads(Path("state/blocked.json").read_text())
elif how == "one-more":      # well-formed and inside every limit
    plan["add"].append("203.0.113.250")
else:
    sys.exit("usage: tamper.py string|unblock-all|one-more")
plan_file.write_text(json.dumps(plan, indent=1) + "\n")
print(f"plan.json edited: {how}")

A bare string in add was refused by Plan.load: without the type check, set("203.0.113.250") is a set of characters, and those would have been written into the state as entries. The unblock-all edit is well-formed JSON with real addresses, so the type checks pass; apply then ran the same check_limits as plan and refused to remove 10 of 10. The state's sha256 is still the state_sha256 from plan.json above: nothing was written. The third edit adds one valid address and stays inside every limit, so validation has nothing to object to, and the lab confirmed that a plain apply accepts it. Only the digest the reviewer recorded catches it: with --plan-sha256, apply refuses any file that is not byte for byte the reviewed one. Treat CI plan artifacts like code: keep them in storage the apply job trusts, and bind the apply to the reviewed digest or a signature.

apply also reads the state, compares its digest and writes the new state. If two applies ran at once, both could pass the check before either wrote. sync.py therefore holds an exclusive lock (fcntl.flock on state/blocked.json.lock, the same call as flock(1)) from the read to the os.replace. The first command below makes a fresh plan (byte for byte the reviewed one, so the recorded digest still matches) and runs apply while flock holds that lock, the way a second copy of the job would:

The locked apply changed nothing and exited 75: the job should simply run again later. With the lock free, the reviewed plan applied. The second apply was refused with exit status 1: the state no longer matched the digest, so the same plan cannot be applied twice, or over changes someone else made. The lock file stays in state/ after the run, for the reason "Temp files, locks and timeouts" in bash-ops gave: deleting it while a job holds it would let a second job lock a new file.

Paths in blsync.toml follow the file, but everything else is relative to the working directory: the default blsync.toml and plan.json, the defaults in config.py, and any path in a flag or a BLSYNC_* variable. That directory depends on who starts the job: / for a systemd system service unless the unit sets WorkingDirectory=, the home directory for cron. For a service, set StateDirectory=blsync (systemd creates /var/lib/blsync for the service's user and passes it as $STATE_DIRECTORY), point WorkingDirectory= there, and use absolute paths in blsync.toml.

The exit contract

Every outcome above ended in one of a few statuses, defined once. Exit is an IntEnum: an enumeration whose members are also plain integers, so the code says Exit.CONFIG and the shell receives 78:

"""The exit statuses blsync documents: the course contract, and what the caller does next."""
from enum import IntEnum
 
 
class Exit(IntEnum):
    OK = 0          # plan saved, or plan applied
    FAILED = 1      # a human must look: refused, malformed input, a file that cannot be read
    USAGE = 2       # argparse: unknown option, missing argument
    SOFTWARE = 70   # a bug: an exception nobody classified; the traceback goes to stderr
    TEMPFAIL = 75   # another apply holds the lock; nothing changed, run again later
    CONFIG = 78     # unusable settings in the file, the environment or the flags

These are the course's contract, which "Observable jobs" later in this course owns and explains for a whole pipeline: 0, 1 and 2 as py-sec set them, and 70, 75 and 78 from sysexits.h. The rule for choosing is what the caller does next. A cron wrapper retries 75, pages nobody for it and pages someone for 1; 78 means "fix the config", and 70 means a bug. A refused truncated feed and a malformed address both get 1, because in both cases a human looks at the input before anything changes; the one-line message says which. Each class from errors.py maps to exactly one status in main():

"""The command line: the only module that reads argv and os.environ and picks the exit status."""
import argparse
import os
import sys
import traceback
from pathlib import Path
 
from .errors import BusyError, ConfigError, RefusedError
from .exits import Exit
from .loader import load_config
from .plan import Plan
from .sync import apply_plan, make_plan
 
PLAN_FILE = Path("plan.json")  # relative to the working directory, like every path given here
 
 
def parse_args(argv: list[str] | None) -> argparse.Namespace:
    p = argparse.ArgumentParser(prog="blsync", allow_abbrev=False, epilog="exit status: 0 ok,"
                                " 1 failed, 2 usage, 70 software, 75 tempfail, 78 config")
    p.add_argument("--config", type=Path, default=Path("blsync.toml"))
    p.add_argument("--max-remove-fraction", type=float, metavar="F")
    sub = p.add_subparsers(dest="command", required=True)
    sub.add_parser("plan", help=f"compute the changes and save them to {PLAN_FILE}")
    apply = sub.add_parser("apply", help=f"apply {PLAN_FILE} if the state has not changed")
    apply.add_argument("--plan-sha256", metavar="HEX",
                       help=f"refuse unless {PLAN_FILE} has this sha256 (the reviewed plan)")
    return p.parse_args(argv)
 
 
def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    try:
        cfg = load_config(args.config, os.environ,
                          {"max_remove_fraction": args.max_remove_fraction})
        if args.command == "plan":
            plan = make_plan(cfg)
            plan.save(PLAN_FILE)
            print(f"plan: add {plan.add}, remove {plan.remove}; saved to {PLAN_FILE}")
        else:
            plan = Plan.load(PLAN_FILE, args.plan_sha256)
            apply_plan(plan, cfg)
            print(f"applied: added {len(plan.add)}, removed {len(plan.remove)}")
    except ConfigError as err:
        return fail(Exit.CONFIG, "config", err)
    except RefusedError as err:
        return fail(Exit.FAILED, "refused", err)
    except BusyError as err:
        return fail(Exit.TEMPFAIL, "busy", err)
    except OSError as err:
        return fail(Exit.FAILED, "failed", err)
    except Exception:
        traceback.print_exc()  # unclassified means a bug: keep every detail for the fix
        return Exit.SOFTWARE
    return Exit.OK
 
 
def fail(status: Exit, kind: str, err: Exception) -> int:
    print(f"blsync: {kind}: {err}", file=sys.stderr)  # one line, no traceback
    return status

The contract is in the help text, where an operator finds it. The final except Exception is the one place a traceback is printed: an exception no layer classified is a bug, and the lab confirmed that main() then prints the traceback and returns 70. main() takes an argument list and returns the status instead of exiting, so a pytest case can call main(["--config", "typo.toml", "plan"]) and assert that it returns 78, with no subprocess.

One case sits between 1 and 2: an input file that the run was told to use (here through BLSYNC_SOURCE) but that cannot be read. except OSError gives it 1. The command line was valid and the file is not, so a human has to fix or replace the file before a rerun helps; 2 stays for what argparse rejects, as the run without a command showed:

Try this

Add a setting min_entries: int (default 0) that refuses any plan leaving fewer entries than that, and set min_entries = 8 in blsync.toml. It needs a field and a check in config.py, one entry in TYPES in loader.py (which gives you the TOML key and BLSYNC_MIN_ENTRIES for free) and three lines in check_limits, so apply enforces it too. Verify: with the truncated feed and --max-remove-fraction 1, the fraction check no longer protects you, and the lab's solution printed blsync: refused: would leave 3 entries, min_entries is 8 with exit status 1; BLSYNC_MIN_ENTRIES=eight gave blsync: config: BLSYNC_MIN_ENTRIES must be int, got 'eight' and exit status 78; the normal plan still passes, and mypy --strict blsync still reports no issues.

Takeaway

Keep argv, the environment and exit statuses in one edge module, merge settings in a fixed order into one frozen object validated once, and treat a saved plan as untrusted input: at apply, check its types, rerun the same limits under a lock, and require the digest the reviewer approved.

Related