Building Practical Automation Workflows

Most automation starts the same way: a task that takes ten minutes, done often enough to be annoying. A script appears. Six months later the script is a small, undocumented system that fails at night and nobody remembers how it works.

Useful automation is small, observable and safe to run twice. Four habits get you most of the way there.

1. Write the manual process down first

Before automating anything, write the steps as a checklist and run it by hand a few times. You will find the steps that need judgement, the edge cases, and the parts that are not worth automating at all.

Automate the boring, predictable majority. Leave the judgement calls to a person, and make the automation hand them over clearly instead of guessing.

2. Make every run safe to repeat

Jobs get retried. Schedulers overlap. Someone runs the script by hand while the timer is also running it. A job that creates a duplicate invoice the second time it runs is a liability.

Idempotency means running the job twice has the same effect as running it once. The usual tools:

• Natural keys. Identify each item by something stable, such as an email's message ID or an order number, and refuse to process it twice.
• Check before write. Look for the record you are about to create.
• Idempotency keys. When an API supports them, send one with every request that creates something.

A dry-run flag that prints what would happen, without doing it, makes every one of these easier to trust.

import argparse
import logging

log = logging.getLogger("inbox-sync")


def main() -> int:
    parser = argparse.ArgumentParser(description="Turn labelled emails into tasks.")
    parser.add_argument("--dry-run", action="store_true", help="show what would change")
    args = parser.parse_args()

    created = skipped = 0

    for message in fetch_labelled_messages():
        if task_exists(message.id):
            skipped += 1
            continue

        if args.dry_run:
            log.info("would create task for %s", message.subject)
        else:
            create_task(message, idempotency_key=f"email-{message.id}")

        created += 1

    log.info("done: %d created, %d already existed", created, skipped)
    return 0

The helper functions are left out here. The structure is the point: skip what is already done, say what you are about to do, and count the results.

3. Retry the right failures, the right way

Networks fail briefly and constantly. Retrying makes sense for timeouts and connection errors, and for rate limits once you have waited. It does not make sense for a validation error, which will fail the same way forever.

Exponential backoff with jitter spreads retries out so that many jobs failing at once do not all retry at the same instant:

import random
import time
from functools import wraps


def retry(attempts=4, base=0.5, cap=8.0, retry_on=(TimeoutError, ConnectionError)):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(1, attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except retry_on:
                    if attempt == attempts:
                        raise
                    delay = min(cap, base * 2 ** (attempt - 1))
                    time.sleep(random.uniform(0, delay))
        return wrapper
    return decorator

When the attempts run out, fail loudly and keep enough detail to replay the item later.

4. Make it observable

Every run should leave a line somewhere: when it started, how long it took, what it changed and whether it failed. A small table or a log file is enough. Alert on failures, not successes, or the alerts become noise you learn to ignore.

Schedule it without overlaps

A systemd timer is a good home for a recurring job on a Linux server. It logs to the journal, catches up after downtime, and never starts a second copy while the first is still running:

[Unit]
Description=Inbox sync

[Service]
Type=oneshot
User=automation
ExecStart=/opt/inbox-sync/.venv/bin/python /opt/inbox-sync/run.py
TimeoutStartSec=10min

[Unit]
Description=Run inbox sync every 15 minutes

[Timer]
OnCalendar=*:0/15
Persistent=true
RandomizedDelaySec=60

[Install]
WantedBy=timers.target

sudo systemctl daemon-reload
sudo systemctl enable --now inbox-sync.timer
systemctl list-timers inbox-sync.timer
journalctl -u inbox-sync.service -n 50

If you prefer cron, wrap the command in flock -n so a slow run is never joined by a second one:

*/15 * * * * flock -n /tmp/inbox-sync.lock /opt/inbox-sync/.venv/bin/python /opt/inbox-sync/run.py

The test of a good automation

Ask three questions. Can it run twice without harm? Will you know when it fails? Could someone else read it and understand what it does in five minutes? If the answer to all three is yes, it will save time instead of quietly costing it.