Automating the Repetitive Work
This is where Python starts replacing the manual, repetitive work that eats up your day — running shell commands from a script, copying and archiving files automatically, building command-line tools your colleagues can actually use, and scheduling tasks to run unattended. Everything in this module is aimed directly at IT Support, DevOps, and Automation responsibilities.
Up to now, every script you’ve written has been self-contained Python. Real automation work usually means calling out to the operating system — running a shell command, restarting a service, copying files between directories, zipping up logs. This module gives you the tools to do all of that safely and predictably.
subprocess — Running Shell Commands from Python
The subprocess module lets your Python script run any command you’d normally type into a terminal — ping, systemctl, git, a custom shell script — and capture its output, exit code, and any errors. This is the bridge between Python and everything else on the machine.
The modern, recommended function is subprocess.run(). It takes a list of strings — the command followed by its arguments — runs it, waits for it to finish, and returns a result object with everything you need.
Pass check=True to make Python raise a CalledProcessError automatically if the command fails (non-zero exit code) — this plugs neatly into the error handling you learned in M6. Pass a timeout= value (in seconds) to prevent your script from hanging forever if a command never returns.
shutil — File and Directory Operations Beyond os
You met os.path in M5 for inspecting paths. shutil (“shell utilities”) is its companion for actually moving things around — copying files, copying entire directory trees, deleting directories, and creating archives. These are operations the basic os module either can’t do directly or makes painful.
| Function | What it does |
|---|---|
| shutil.copy(src, dst) | Copies a single file (and its permissions) to a new location |
| shutil.copytree(src, dst) | Recursively copies an entire directory and everything inside it |
| shutil.move(src, dst) | Moves a file or directory — works across filesystems, unlike os.rename |
| shutil.rmtree(path) | Deletes a directory and everything inside it — irreversible, use carefully |
| shutil.disk_usage(path) | Returns total, used, and free disk space for a given path |
| shutil.make_archive(name, fmt, dir) | Creates a .zip or .tar.gz archive of a directory |
Building Command-Line Tools with argparse
So far, your scripts have had hard-coded values or used input() for interactivity. Real automation tools are run from the command line with arguments — python backup.py –source /data –days 7. Python’s argparse module turns your script into a proper CLI tool with named arguments, defaults, help text, and automatic validation.
This help text is generated automatically from how you define the arguments — you never write it by hand. This single feature is why argparse is worth learning over manually parsing sys.argv: your tool becomes self-documenting and behaves the way every other professional CLI tool behaves.
Scheduling Patterns — How Automation Actually Runs
A Python script doesn’t run itself — something has to trigger it on a schedule. Understanding the common patterns helps you design scripts that fit cleanly into whichever scheduler your organisation uses.
The key design principle for any scheduled script: it must be able to run unattended, with no human watching. That means thorough error handling (M6), clear logging of what happened, and never relying on interactive input() calls — those will simply hang forever with nobody there to respond.
Putting It Together — The Shape of a Production Automation Script
A well-built automation script follows a consistent shape: parse arguments → validate inputs → do the work with proper error handling → log what happened → exit with an appropriate status code. sys.exit(0) for success, sys.exit(1) (or higher) for failure — schedulers and monitoring systems read this exit code to know whether your script actually succeeded.
subprocess — running and capturing commands
import subprocess
# Basic run — command as a list, never a single string
result = subprocess.run(["ping", "-c", "4", "8.8.8.8"],
capture_output=True, text=True, timeout=10)
result.returncode # 0 = success, non-zero = failure
result.stdout # captured standard output as a string (because text=True)
result.stderr # captured standard error as a string
# check=True raises CalledProcessError automatically on failure
try:
result = subprocess.run(
["systemctl", "restart", "nginx"],
capture_output=True, text=True, check=True, timeout=15
)
print("Service restarted successfully")
except subprocess.CalledProcessError as e:
print(f"Restart failed (exit {e.returncode}): {e.stderr}")
except subprocess.TimeoutExpired:
print("Command timed out after 15 seconds")
# Checking success without raising — manual check
result = subprocess.run(["git", "status"], capture_output=True, text=True)
if result.returncode == 0:
print("Git repo is clean")
else:
print(f"Git error: {result.stderr}")
# Passing variable arguments safely (no shell injection risk)
hostname = "prod-db-01" # could come from user input or a config file
result = subprocess.run(["ping", "-c", "2", hostname], capture_output=True, text=True)
# Safe — hostname is passed as a distinct argument, never interpreted by a shell
shutil — file and directory operations
import shutil
import os
# Copy a single file
shutil.copy("report.csv", "backups/report.csv")
# Copy an entire directory tree (destination must NOT already exist)
shutil.copytree("project_data", "backups/project_data_2026_06_15")
# Move a file or directory (works across filesystems/drives)
shutil.move("temp/output.log", "archive/output.log")
# Delete a directory and everything in it — IRREVERSIBLE
if os.path.exists("old_temp_dir"):
shutil.rmtree("old_temp_dir")
# Check disk usage
usage = shutil.disk_usage("/")
free_gb = usage.free / (1024 ** 3)
print(f"Free space: {free_gb:.1f} GB")
# Create a zip archive of a directory
shutil.make_archive("logs_backup", "zip", "logs")
# creates logs_backup.zip from the contents of the 'logs' directory
# A safe deletion pattern with dry-run support
def remove_directory(path, dry_run=True):
if dry_run:
print(f"[DRY RUN] Would delete: {path}")
else:
shutil.rmtree(path)
print(f"Deleted: {path}")
argparse — building a real CLI tool
import argparse
parser = argparse.ArgumentParser(
description="Clean up old log files in a directory."
)
# Required argument (no default = required)
parser.add_argument("--directory", required=True,
help="Directory to scan for old files")
# Optional argument with a default and a type conversion
parser.add_argument("--days", type=int, default=30,
help="Delete files older than this many days (default: 30)")
# Boolean flag — present means True, absent means False
parser.add_argument("--dry-run", action="store_true",
help="Show what would be deleted without deleting")
# Restrict to a fixed set of choices
parser.add_argument("--log-level", choices=["DEBUG", "INFO", "WARNING"], default="INFO")
args = parser.parse_args()
# Access parsed values as attributes — note --dry-run becomes args.dry_run
print(args.directory)
print(args.days)
print(args.dry_run)
# Run it:
# python cleanup.py --directory /var/log --days 7 --dry-run
# python cleanup.py --directory /var/log (uses default days=30)
# python cleanup.py --help (auto-generated help text)
Exit codes for scheduled scripts
import sys
def main():
try:
run_backup_job()
except Exception as e:
print(f"FATAL: {e}", file=sys.stderr)
sys.exit(1) # non-zero — scheduler/monitor sees this as a failure
sys.exit(0) # zero — explicit success signal
if __name__ == "__main__":
main()
These examples are written the way real automation scripts are structured in production — proper argument parsing, error handling around every external command, and clear logging output. This is the level of polish expected of automation code that other people will run.
Pings a list of hosts using subprocess, capturing success/failure for each without letting one unreachable host crash the whole sweep. The classic “is anything down right now” health check.
import subprocess
from datetime import datetime
def ping_host(host, count=2, timeout=5):
"""
Ping a host and return (success: bool, detail: str).
Never raises — failures are reported, not propagated.
"""
try:
result = subprocess.run(
["ping", "-c", str(count), "-W", str(timeout), host],
capture_output=True, text=True, timeout=timeout + 3
)
if result.returncode == 0:
return True, "reachable"
return False, "no response"
except subprocess.TimeoutExpired:
return False, "timed out"
except FileNotFoundError:
return False, "ping command not available"
def sweep(hosts):
"""Ping every host, isolating failures, and return results list."""
results = []
for host in hosts:
ok, detail = ping_host(host)
results.append({"host": host, "ok": ok, "detail": detail})
return results
# ── Run ──
hosts = ["8.8.8.8", "1.1.1.1", "10.255.255.1", "127.0.0.1"]
print(f"Ping sweep — {datetime.now().strftime('%H:%M:%S')}\n")
results = sweep(hosts)
up = [r for r in results if r["ok"]]
down = [r for r in results if not r["ok"]]
for r in results:
icon = "✓" if r["ok"] else "✗"
print(f" {icon} {r['host']:16} {r['detail']}")
print(f"\n{len(up)}/{len(hosts)} reachable")
✓ 8.8.8.8 reachable
✓ 1.1.1.1 reachable
✗ 10.255.255.1 timed out
✓ 127.0.0.1 reachable
3/4 reachable
A complete argparse-driven CLI tool that finds log files older than N days, zips them into an archive, and optionally deletes the originals — with a –dry-run flag that’s the default-safe behaviour. This is the canonical shape of a real ops utility script.
import argparse
import os
import shutil
import sys
from datetime import datetime, timedelta
def find_old_logs(directory, days):
"""Return paths of .log files older than `days` days."""
cutoff = datetime.now() - timedelta(days=days)
old_files = []
for filename in os.listdir(directory):
if not filename.endswith(".log"):
continue
filepath = os.path.join(directory, filename)
modified = datetime.fromtimestamp(os.path.getmtime(filepath))
if modified < cutoff:
old_files.append(filepath)
return old_files
def archive_logs(files, archive_name, dry_run):
"""Zip the given files into archive_name.zip, then optionally delete originals."""
if dry_run:
print(f"[DRY RUN] Would archive {len(files)} file(s) into {archive_name}.zip")
for f in files:
print(f" - {f}")
return
staging_dir = "_archive_staging"
os.makedirs(staging_dir, exist_ok=True)
for f in files:
shutil.copy(f, staging_dir)
shutil.make_archive(archive_name, "zip", staging_dir)
shutil.rmtree(staging_dir)
for f in files:
os.remove(f)
print(f"Archived {len(files)} file(s) → {archive_name}.zip (originals removed)")
def main():
parser = argparse.ArgumentParser(description="Archive old log files into a zip.")
parser.add_argument("--directory", required=True, help="Directory containing .log files")
parser.add_argument("--days", type=int, default=30, help="Archive files older than this many days")
parser.add_argument("--dry-run", action="store_true", help="Preview without making changes")
args = parser.parse_args()
if not os.path.isdir(args.directory):
print(f"Error: {args.directory} is not a valid directory", file=sys.stderr)
sys.exit(1)
old_logs = find_old_logs(args.directory, args.days)
if not old_logs:
print(f"No .log files older than {args.days} days found in {args.directory}")
sys.exit(0)
archive_name = f"logs_archive_{datetime.now().strftime('%Y%m%d')}"
archive_logs(old_logs, archive_name, args.dry_run)
if __name__ == "__main__":
main()
# Run examples:
# python log_archiver.py --directory /var/log/app --days 14 --dry-run
# python log_archiver.py --directory /var/log/app --days 14
– ./logs/app_2026-05-20.log
– ./logs/app_2026-05-21.log
– ./logs/worker_2026-05-22.log
Checks whether a system service is running using subprocess, and restarts it automatically if it’s down — the kind of self-healing script that runs on a schedule to reduce manual intervention. Demonstrates check=True combined with proper exception handling.
import subprocess
import sys
from datetime import datetime
class ServiceCheckError(Exception):
"""Raised when a service status cannot be determined or restarted."""
pass
def is_service_active(service_name):
"""Return True if the service is active, using systemctl is-active."""
result = subprocess.run(
["systemctl", "is-active", service_name],
capture_output=True, text=True
)
# systemctl is-active prints "active" and returns 0 only when running
return result.returncode == 0 and result.stdout.strip() == "active"
def restart_service(service_name):
"""Attempt to restart a service. Raises ServiceCheckError on failure."""
try:
subprocess.run(
["systemctl", "restart", service_name],
capture_output=True, text=True, check=True, timeout=20
)
except subprocess.CalledProcessError as e:
raise ServiceCheckError(f"Failed to restart {service_name}: {e.stderr.strip()}") from e
except subprocess.TimeoutExpired:
raise ServiceCheckError(f"Restart of {service_name} timed out")
def check_and_heal(services):
"""Check each service; restart and log any that are down."""
log = []
for svc in services:
if is_service_active(svc):
log.append(f"✓ {svc}: already running")
continue
log.append(f"✗ {svc}: down — attempting restart")
try:
restart_service(svc)
log.append(f" ↻ {svc}: restarted successfully")
except ServiceCheckError as e:
log.append(f" ⚠ {svc}: {e}")
return log
# ── Run ──
services = ["nginx", "postgresql", "redis"]
print(f"Service health check — {datetime.now().strftime('%H:%M:%S')}\n")
for line in check_and_heal(services):
print(line)
✓ nginx: already running
✗ postgresql: down — attempting restart
↻ postgresql: restarted successfully
✓ redis: already running
A CLI tool that sorts files in a messy downloads-style folder into subfolders by file extension, using shutil.move(). Built with argparse so it can be run by anyone on the team without editing the script.
import argparse
import os
import shutil
import sys
EXTENSION_MAP = {
".pdf": "Documents", ".docx": "Documents", ".txt": "Documents",
".jpg": "Images", ".png": "Images", ".gif": "Images",
".csv": "Spreadsheets", ".xlsx": "Spreadsheets",
".zip": "Archives", ".tar": "Archives",
}
def organise(directory, dry_run):
"""Move each file into a subfolder based on its extension."""
moved, skipped = 0, 0
for filename in os.listdir(directory):
filepath = os.path.join(directory, filename)
if not os.path.isfile(filepath):
continue
_, ext = os.path.splitext(filename)
folder_name = EXTENSION_MAP.get(ext.lower())
if folder_name is None:
skipped += 1
continue
target_dir = os.path.join(directory, folder_name)
target_path = os.path.join(target_dir, filename)
if dry_run:
print(f"[DRY RUN] {filename} → {folder_name}/")
else:
os.makedirs(target_dir, exist_ok=True)
shutil.move(filepath, target_path)
print(f"Moved {filename} → {folder_name}/")
moved += 1
return moved, skipped
def main():
parser = argparse.ArgumentParser(description="Organise files into subfolders by type.")
parser.add_argument("--directory", required=True, help="Directory to organise")
parser.add_argument("--dry-run", action="store_true", help="Preview without moving files")
args = parser.parse_args()
if not os.path.isdir(args.directory):
print(f"Error: {args.directory} is not a directory", file=sys.stderr)
sys.exit(1)
moved, skipped = organise(args.directory, args.dry_run)
print(f"\n{moved} file(s) organised, {skipped} file(s) skipped (unknown type)")
if __name__ == "__main__":
main()
[DRY RUN] team_photo.jpg → Images/
[DRY RUN] q2_budget.xlsx → Spreadsheets/
[DRY RUN] backup.zip → Archives/
4 file(s) organised, 1 file(s) skipped (unknown type)
A scheduler-ready script that checks disk usage with shutil.disk_usage() and exits with a non-zero status code if free space drops below a threshold — designed to be run by cron and hooked into alerting (any monitoring system watching exit codes will fire an alert automatically).
import argparse
import shutil
import sys
from datetime import datetime
def check_disk(path, min_free_pct):
"""
Return (ok: bool, free_pct: float, message: str).
ok is False if free space percentage is below min_free_pct.
"""
usage = shutil.disk_usage(path)
free_pct = (usage.free / usage.total) * 100
ok = free_pct >= min_free_pct
message = (f"{path}: {free_pct:.1f}% free "
f"({usage.free // (1024**3)} GB / {usage.total // (1024**3)} GB)")
return ok, free_pct, message
def main():
parser = argparse.ArgumentParser(description="Monitor disk space and alert if low.")
parser.add_argument("--path", default="/", help="Path/mount point to check")
parser.add_argument("--min-free-pct", type=float, default=15.0,
help="Minimum free space percentage before alerting")
args = parser.parse_args()
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
ok, free_pct, message = check_disk(args.path, args.min_free_pct)
if ok:
print(f"[{ts}] OK — {message}")
sys.exit(0)
else:
print(f"[{ts}] ALERT — {message} (threshold: {args.min_free_pct}%)", file=sys.stderr)
sys.exit(2) # non-zero — a monitoring system watching this exit code will alert
if __name__ == "__main__":
main()
# Scheduled via cron, e.g.:
# */15 * * * * /usr/bin/python3 /opt/scripts/disk_monitor.py --path /var --min-free-pct 10
exit code: 2
Run these on your own machine — subprocess and shutil exercises only make sense when you can see them actually touch the filesystem or call a real command. Always test destructive operations (move, delete, archive) on throwaway test files first.
📋 Backup & Retention CLI Tool
Build a production-style command-line tool called backup_tool.py that backs up a directory and enforces a retention policy on old backups — a real pattern used in countless ops environments. This brings together argparse, shutil, error handling, and exit code discipline from this entire module.
- Argument parsing — required –source (directory to back up) and –backup-dir (where backups are stored); optional –keep (integer, default 5 — how many recent backups to retain) and –dry-run (flag).
- create_backup(source, backup_dir, dry_run) — uses shutil.make_archive() to zip the source directory into backup_dir, named with a timestamp (e.g. mydata_20260615_0941.zip). In dry-run mode, print what would happen without creating anything.
- enforce_retention(backup_dir, keep, dry_run) — lists all .zip files in backup_dir matching the backup naming pattern, sorts them by modification time, and deletes all but the keep most recent ones. In dry-run mode, print which files would be deleted without deleting them.
- Validation — if –source doesn’t exist or isn’t a directory, print a clear error to sys.stderr and exit with code 1 before attempting anything else.
- Error handling — wrap the backup and retention steps so that any unexpected failure (disk full, permission denied, etc.) is caught, logged with a clear message, and results in sys.exit(1) rather than an unhandled traceback.
- Summary output — on success, print the backup file created, how many old backups were removed (if any), and exit with code 0.
These questions focus on the judgment calls that separate a script that works on your laptop from one that’s safe to run unattended in production.
