Task Automation: Scripting the Boring Parts

The rule is simple: if you've done it manually three times, automate it. Doing it by hand a fourth time wastes your time, and sooner or later you'll forget a step and break something. Python is one of the most common languages for system automation. The stdlib already covers the basics you need (file operations, running other programs, scheduling, locking), and two or three well-known libraries fill the gaps.

This lesson covers the tools: subprocess for running other programs, sched/cron/Task Scheduler for running jobs at set times, watchdog for filesystem events, and file locking to stop two runs of a job from clashing. It also covers idempotency, which makes a scheduled job safe to run again after a failure.

1. The Automation Mindset

Three categories cover almost every automation job you'll write:

• Batch file operations: rename a thousand photos, remove duplicate files from a folder tree, archive last quarter's logs. This is all pathlib work. See fileio for the streaming patterns.
• Scheduled jobs: generate a report every morning, back up a database every hour, email a summary every Monday. These are started by the clock.
• Connecting programs: call ffmpeg, pg_dump, git or any other CLI from Python, parse its output, and pass the result on to something else. Python controls the other programs.

The third category is where you need subprocess.

2. subprocess.run: The One You Should Use

subprocess.run is the modern, blocking API: it runs a command, waits for it to finish, and tells you how it went. The full call:

import subprocess

result = subprocess.run(
    ["git", "status", "--short"],
    capture_output=True,
    text=True,
    check=True,
)

print(result.stdout)        # the captured stdout as a str
print(result.stderr)        # likewise stderr
print(result.returncode)    # 0 on success (anything else raised CalledProcessError)

Four flags that matter:

• capture_output=True: store stdout and stderr in the result object instead of printing them to the terminal.
• text=True: decode the captured bytes as text. Without it you get bytes and have to .decode() everywhere. Always pass it for human-readable commands.
• check=True: raise subprocess.CalledProcessError on a non-zero exit code. Without it, you have to inspect returncode yourself and remember to handle failure. With it, failures look like every other Python exception.
• cwd="/some/path" and env={...}: run in a different working directory or with a custom environment. Both are common in build/deploy scripts.

If you only remember one signature, remember that one.

3. shell=True: The Dangerous Option

# Danger: the shell reads and interprets the whole string
subprocess.run(f"ls {user_supplied_path}", shell=True)

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

subprocess = _AutoMock('subprocess')
user_supplied_path = _AutoMock('user_supplied_path')

If user_supplied_path is "; rm -rf ~", the shell runs both commands without any warning. This is shell injection, the same class of bug as SQL injection, and the fix is the same: don't concatenate user input into a command string.

The list form doesn't use the shell at all:

# Safe: arguments go straight to the program, with no shell in between
subprocess.run(["ls", user_supplied_path])

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

subprocess = _AutoMock('subprocess')
user_supplied_path = _AutoMock('user_supplied_path')

Use shell=True only when you really need shell features (pipes, globs, &&, environment variable expansion), and even then, never with untrusted input. Most subprocess.run calls should use the list form.

4. Streaming Output for Long-Running Commands

subprocess.run is blocking: it waits for the child process to finish before you see anything. For a build that takes ten minutes, that is a long wait with no feedback. subprocess.Popen gives you a live handle:

import subprocess

proc = subprocess.Popen(
    ["pytest", "-v"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,         # merge stderr into stdout
    text=True,
    bufsize=1,                        # line-buffered: output arrives one line at a time
)

for line in proc.stdout:
    print(line, end="")               # stream test output as it happens
    # parse, filter, or react here

proc.wait()
if proc.returncode != 0:
    raise RuntimeError(f"pytest failed with exit {proc.returncode}")

bufsize=1 plus iterating proc.stdout gives you a line-at-a-time stream. You can filter the lines or react to them as they arrive. This is useful for CI logs, deploy scripts, and anything that should show progress.

5. Batch File Operations: pathlib + Glob

This is the most common kind of automation. Here are three patterns you'll write again and again:

from pathlib import Path
from datetime import datetime

# Rename: "IMG_1234.jpg" becomes "2026-05-14_1234.jpg"
for img in Path("photos").glob("IMG_*.jpg"):
    stamp = datetime.fromtimestamp(img.stat().st_mtime).strftime("%Y-%m-%d")
    img.rename(img.parent / f"{stamp}_{img.stem[4:]}.jpg")

# Remove duplicates by content hash: keep one copy, delete the rest
import hashlib
seen = {}
for f in Path("downloads").rglob("*"):
    if not f.is_file():
        continue
    digest = hashlib.sha256(f.read_bytes()).hexdigest()
    if digest in seen:
        f.unlink()                    # a duplicate, so delete it
    else:
        seen[digest] = f

# Archive: move anything older than 30 days into archive/
import time
cutoff = time.time() - 30 * 86400
archive = Path("archive")
archive.mkdir(exist_ok=True)
for f in Path("logs").glob("*.log"):
    if f.stat().st_mtime < cutoff:
        f.rename(archive / f.name)

For huge files in a dedupe pass, hash in chunks instead of using read_bytes(). See fileio Section 6. The same pathlib methods work for renaming five files and for processing a dataset of a million files.

6. Scheduling: Pick the Right Level

There are four levels of scheduling, from the lightest to the heaviest:

In-process: sched.scheduler

For "run this every N seconds while the program is alive." It is lightweight and part of the stdlib, with no external dependencies:

import sched, time

s = sched.scheduler(time.time, time.sleep)

def heartbeat():
    print("ping", time.strftime("%H:%M:%S"))
    s.enter(5, 1, heartbeat)          # schedule itself again

s.enter(0, 1, heartbeat)
s.run()                               # blocks; runs scheduled events in order

sched is fine for short-lived daemons and one-off scripts. Don't use it for jobs that must run for months: sooner or later something will restart your process, and the schedule is lost.

OS-level: cron / Task Scheduler

For "run this every day at 03:00" on a server, write a normal Python script and let the OS handle the trigger:

# crontab -e on Linux/macOS
0 3 * * * /usr/bin/python3 /opt/app/daily_report.py >> /var/log/daily_report.log 2>&1

On Windows, Task Scheduler does the same job through a GUI or schtasks.exe. This is the default for production. The OS handles the timing, and the schedule survives reboots. Your Python script only has to do the work.

Library: APScheduler

When you need cron expressions inside a long-running Python process (a web app, a worker) and want persistence (jobs survive restarts), use APScheduler. It supports interval, cron and date triggers, and can store jobs in SQLAlchemy, Redis or MongoDB.

Distributed: Celery beat, Airflow, Prefect

For web apps with worker fleets, Celery beat is the standard replacement for cron: it sends jobs out to the workers. For data pipelines with dependencies between tasks, use Airflow or Prefect. We only mention these here, because each one takes weeks to learn.

7. File Watching: watchdog

watchdog reacts to filesystem events instead of polling. Drop a file into incoming/ and your script processes it within milliseconds, without a loop that checks the folder every 10 seconds and uses CPU time:

# pip install watchdog
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class IncomingHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            return
        print(f"new file: {event.src_path}")
        # process(event.src_path)

observer = Observer()
observer.schedule(IncomingHandler(), path="incoming", recursive=False)
observer.start()
try:
    while True:
        time.sleep(1)
finally:
    observer.stop()
    observer.join()

Use this when every new file must be processed as soon as it arrives: uploads, scanner outputs, exported reports. The alternative, checking os.listdir every few seconds, works, but it wastes CPU time and adds delay.

8. HTTP Polling vs Webhooks

When you're integrating with another system, the question is who initiates contact:

Polling is simpler to start with and harder to scale: you can never react faster than the poll interval, and most requests are wasted when nothing has changed. Webhooks are almost instant, but they need a public URL, signature verification, and idempotent handling because the same event may arrive twice.

9. Idempotency: Design for Retries

If a job is safe to run again, you can retry it after a failure. If it isn't, it breaks as soon as cron starts it twice or you re-run a failed batch.

# Not idempotent: running it again creates duplicate rows
def import_orders(rows):
    for row in rows:
        db.insert(row)

# Idempotent: UPSERT by order_id, so a re-run gives the same final state
def import_orders(rows):
    for row in rows:
        db.upsert(row, key="order_id")

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

db = _AutoMock('db')

Three techniques for idempotency:

• UPSERT instead of INSERT: the same key gives the same result on a re-run.
• Check for a marker first: write a done.flag file at the end. On retry, skip the work if the file exists.
• Compute, then rename: generate report.csv.tmp, then atomically rename it to report.csv. A half-finished run never leaves a half-written final file (see fileio Section 7 for the pattern).

Cron starts your job at exactly 03:00, whether yesterday's run finished or not. Design your job with that in mind.

10. Locking: Stopping Overlapping Runs

Two cron invocations of the same script can overlap if the first runs long. The cleanest cross-platform fix is the filelock library:

# pip install filelock
from filelock import FileLock, Timeout

lock = FileLock("/tmp/daily_backup.lock")
try:
    with lock.acquire(timeout=1):
        run_backup()
except Timeout:
    print("another backup is already running; skipping")

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def run_backup(*_a, **_kw):
    print('-> run_backup() called')
    return _AutoMock('run_backup()')

Stdlib alternatives exist but are platform-specific: fcntl.flock on Unix, msvcrt.locking on Windows. filelock wraps both. The pattern is the same everywhere: take the lock, do the work, release the lock on exit, and don't start if you couldn't get the lock.

11. The try / except / log / notify Pattern

The full skeleton of a maintainable automation:

import logging
import smtplib

logger = logging.getLogger(__name__)

def daily_job():
    try:
        do_the_work()
    except Exception:
        logger.exception("daily_job failed")
        notify_oncall("daily_job failed. See the logs.")
        raise                             # let the runner record a non-zero exit code

# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

def do_the_work(*_a, **_kw):
    print('-> do_the_work() called')
    return _AutoMock('do_the_work()')
def notify_oncall(*_a, **_kw):
    print('-> notify_oncall() called')
    return _AutoMock('notify_oncall()')

Three things happen on failure: a full traceback goes to the logs (logger.exception includes it, see logging), the person on call gets an alert, and the process exits non-zero so cron/systemd marks it as failed. Without the notify step, the error ends up in a log or a local mailbox that nobody reads, and you find out three days later.

A wide try/except like this is fine here because it doesn't hide the error: it logs everything and then re-raises. Compare with exceptions Section 10.

12. Common Mistakes

1. shell=True with user input
Shell injection. Use the list form.

2. Not capturing stderr
If check=False and you only capture stdout, a failing command looks identical to a silent one. Capture both, or pipe stderr=subprocess.STDOUT to merge them.

3. Long-running scripts with no progress logs
Half an hour in, is it still working or stuck on a hung connection? Log progress in your loops, every external call and every state change. It costs very little and saves a lot of guessing.

4. Overlapping cron jobs
The 03:00 job takes 90 minutes one day and is still running when the 04:00 run starts. Both write to the same files, and the output is corrupted. Use a lock (see Section 10).

5. Hardcoded paths

# Bad: breaks as soon as the script moves
config = open("/home/surya/app/config.json").read()

# Good: relative to the script's own folder
from pathlib import Path
HERE = Path(__file__).resolve().parent
config = (HERE / "config.json").read_text(encoding="utf-8")

6. Wide try/except Exception: pass
Hides every bug. Catch what you can handle, log everything else, and re-raise. See exceptions.

7. No failure notification
By default, cron sends a job's output by local email (for example to root@localhost), and nobody reads it. Send failures to a Slack, PagerDuty or email channel that you really watch.

🎯 Your Turn: Daily Backup Script

Write daily_backup(source_dir, archive_dir) that:

1. Creates a timestamped .tar.gz archive of source_dir inside archive_dir/.
2. Deletes any archive in archive_dir/ older than 30 days.
3. Logs each step at INFO level, and logs failures with traceback at ERROR.

The solution should show both approaches: the subprocess form (calling tar) and the pure-stdlib tarfile form.

from pathlib import Path

def daily_backup(source_dir, archive_dir):
    # TODO 1: ensure archive_dir exists
    # TODO 2: build a timestamped archive filename
    # TODO 3: create the .tar.gz of source_dir
    # TODO 4: delete archives older than 30 days
    # TODO 5: log each step; on error, log with traceback and re-raise
    ...

import logging
import subprocess
import tarfile
import time
from datetime import datetime, timezone
from pathlib import Path

logger = logging.getLogger(__name__)


def daily_backup(source_dir, archive_dir, *, use_subprocess=False):
    """Create a timestamped tar.gz of source_dir, then delete archives older than 30 days."""
    source_dir = Path(source_dir).resolve()
    archive_dir = Path(archive_dir).resolve()
    archive_dir.mkdir(parents=True, exist_ok=True)

    stamp = datetime.now(tz=timezone.utc).strftime("%Y%m%d_%H%M%S")
    out_path = archive_dir / f"backup_{stamp}.tar.gz"

    try:
        logger.info("creating archive %s from %s", out_path, source_dir)

        if use_subprocess:
            # Approach A: run the tar program. Useful when the system tar
            # has flags or compression you want and you trust the binary.
            subprocess.run(
                ["tar", "-czf", str(out_path), "-C", str(source_dir.parent), source_dir.name],
                check=True,
                capture_output=True,
                text=True,
            )
        else:
            # Approach B: the stdlib tarfile module. Nothing to install, and it works on Windows.
            with tarfile.open(out_path, "w:gz") as tar:
                tar.add(source_dir, arcname=source_dir.name)

        logger.info("archive created (%.1f KB)", out_path.stat().st_size / 1024)

        # Delete old archives (anything older than 30 days)
        cutoff = time.time() - 30 * 86400
        pruned = 0
        for f in archive_dir.glob("backup_*.tar.gz"):
            if f.stat().st_mtime < cutoff:
                logger.info("pruning old archive %s", f.name)
                f.unlink()
                pruned += 1
        logger.info("done, pruned %d old archive(s)", pruned)

    except subprocess.CalledProcessError as e:
        logger.exception("tar failed: stderr=%s", e.stderr)
        raise
    except Exception:
        logger.exception("daily_backup failed")
        raise


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
    daily_backup("./data", "./archive")

What You Learned

• subprocess.run(["cmd", "arg"], capture_output=True, text=True, check=True) is the modern default. Pass a list of arguments, not a string, and use check=True so failures raise an error.
• shell=True with user input is shell injection. Use the list form.
• Popen + iterate stdout for live streaming of long-running commands.
• Scheduling tiers: sched in-process, cron/Task Scheduler at OS level, APScheduler for cron expressions inside a daemon, Celery/Airflow for distributed.
• watchdog for filesystem-event-driven automation instead of polling.
• Idempotency: design jobs so that running them again is safe (UPSERT, markers, atomic rename).
• File locks (filelock library) to stop overlapping cron runs.
• try/except → logger.exception → notify → re-raise is the maintainable failure pattern.
• Path(__file__).resolve().parent instead of hardcoded paths.

Next: Logging & Debugging. When an automation script fails in production, good logs are how you find out why.