Your First Automated Python Script That Validates and Runs Itself

Most Python automation tutorials skip the part where your script breaks at 3am and you wake up to a process that's been stuck for 6 hours.

This tutorial builds something different: a script that checks its own code before running it, saves its state so it can resume after a crash, and tells you exactly what went wrong.

You need Python 3.10+. No third-party packages — everything here is in the standard library.

---

🎁 Free: AI Publishing Checklist — 7 steps in Python · Full pipeline: germy5.gumroad.com/l/xhxkzz (pay what you want, min $9.99)

---

What we're building

A script that:

1. Reads a list of tasks from a JSON file
2. Checks each task's code for syntax errors before running it
3. Runs each task in an isolated subprocess
4. Saves which tasks succeeded so it can skip them on restart
5. Reports clearly what passed and what failed

By the end you'll have a reusable pattern for any long-running Python automation.

Step 1: The task manifest

Create tasks.json:

{
  "tasks": [
    {
      "id": "task-01",
      "name": "Print hello world",
      "code": "print('Hello, automation!')"
    },
    {
      "id": "task-02",
      "name": "Count to 5",
      "code": "for i in range(1, 6):\n    print(i)"
    },
    {
      "id": "task-03",
      "name": "Calculate square roots",
      "code": "import math\nfor n in [4, 9, 16, 25]:\n    print(f'sqrt({n}) = {math.sqrt(n)}')"
    }
  ]
}

Each task has an id, a name, and code to run.

Step 2: Load tasks and state

import json
import os

TASKS_FILE = "tasks.json"
STATE_FILE = "task_state.json"

def load_tasks() -> list[dict]:
    """Load task definitions from tasks.json."""
    with open(TASKS_FILE) as f:
        data = json.load(f)
    return data["tasks"]

def load_state() -> dict:
    """Load saved task states. Returns empty dict if no state file exists."""
    if not os.path.exists(STATE_FILE):
        return {}
    with open(STATE_FILE) as f:
        return json.load(f)

def save_state(state: dict) -> None:
    """Write state to disk immediately after every change."""
    with open(STATE_FILE, "w") as f:
        json.dump(state, f, indent=2)

Why save after every change? If your script crashes between saving state and completing the next task, you lose progress. Saving immediately means the worst case is one repeated task — not a full restart.

Step 3: Syntax validation with ast

import ast

def validate_syntax(code: str) -> tuple[bool, str]:
    """
    Check if code has valid Python syntax.
    Returns (is_valid, error_message).
    """
    try:
        ast.parse(code)
        return True, ""
    except SyntaxError as e:
        return False, f"Syntax error at line {e.lineno}: {e.msg}"

# Test it:
ok, msg = validate_syntax("print('hello')")
print(ok, msg)   # True, ''

ok, msg = validate_syntax("print('unclosed")
print(ok, msg)   # False, 'Syntax error at line 1: ...'

ast.parse() reads your code without running it. It's fast and safe — no execution risk. Think of it as a spell-checker for Python.

Step 4: Execution with subprocess

import subprocess
import tempfile
import os

def run_code(code: str, timeout: int = 30) -> tuple[bool, str, str]:
    """
    Run code in an isolated temp directory.
    Returns (success, stdout, stderr).
    """
    with tempfile.TemporaryDirectory() as tmpdir:
        # Write code to a temp file
        script_path = os.path.join(tmpdir, "task.py")
        with open(script_path, "w") as f:
            f.write(code)

        # Run it
        try:
            result = subprocess.run(
                ["python3", script_path],
                capture_output=True,
                text=True,
                timeout=timeout,
                cwd=tmpdir,
            )
            success = result.returncode == 0
            return success, result.stdout, result.stderr

        except subprocess.TimeoutExpired:
            return False, "", f"Timed out after {timeout} seconds"

# Test it:
ok, out, err = run_code("print('it works!')")
print(ok, repr(out))  # True, 'it works!\n'

ok, out, err = run_code("raise ValueError('oops')")
print(ok, err[:50])   # False, 'Traceback ... ValueError: oops'

The tempfile.TemporaryDirectory() creates a fresh folder for each task and deletes it automatically. This means one task can't accidentally affect another.

Step 5: The main loop with state tracking

def run_all_tasks() -> None:
    """
    Run all tasks, skipping completed ones.
    Saves state after each task.
    """
    tasks = load_tasks()
    state = load_state()

    print(f"Found {len(tasks)} tasks")
    print(f"Already completed: {sum(1 for t in tasks if state.get(t['id']) == 'done')}")
    print()

    for task in tasks:
        task_id = task["id"]
        name = task["name"]
        code = task["code"]

        # Skip if already done
        if state.get(task_id) == "done":
            print(f"  ⏭  [{task_id}] {name} — already done, skipping")
            continue

        print(f"  ▶  [{task_id}] {name}")

        # Gate 1: syntax check
        syntax_ok, syntax_err = validate_syntax(code)
        if not syntax_ok:
            print(f"     ❌ Syntax error: {syntax_err}")
            state[task_id] = "failed_syntax"
            save_state(state)
            continue

        print(f"     ✅ Syntax valid")

        # Gate 2: run it
        run_ok, stdout, stderr = run_code(code)
        if run_ok:
            print(f"     ✅ Ran successfully")
            if stdout.strip():
                print(f"     Output: {stdout.strip()[:100]}")
            state[task_id] = "done"
        else:
            print(f"     ❌ Failed: {stderr.strip()[:100]}")
            state[task_id] = "failed_execution"

        save_state(state)

    # Summary
    done = sum(1 for v in state.values() if v == "done")
    failed = sum(1 for v in state.values() if v.startswith("failed"))
    print(f"\nResult: {done} done, {failed} failed")

if __name__ == "__main__":
    run_all_tasks()

Running it

# First run
python3 run_tasks.py

# Output:
# Found 3 tasks
# Already completed: 0
#
#   ▶  [task-01] Print hello world
#      ✅ Syntax valid
#      ✅ Ran successfully
#      Output: Hello, automation!
#   ▶  [task-02] Count to 5
#      ✅ Syntax valid
#      ✅ Ran successfully
#      Output: 1
# ...

# If interrupted after task-01, run again:
python3 run_tasks.py

# Output:
# Found 3 tasks
# Already completed: 1
#
#   ⏭  [task-01] Print hello world — already done, skipping
#   ▶  [task-02] Count to 5
# ...

The state file it creates

After a full run, task_state.json looks like:

{
  "task-01": "done",
  "task-02": "done",
  "task-03": "done"
}

Delete it to start fresh. Edit it to re-run a specific task (change "done" to anything else).

What to build next

This pattern scales to anything:

• Replace tasks.json with an API call that returns work items
• Add a "running" state to detect crashed tasks on startup (reset them to pending)
• Add retry logic: if failed_execution, try once more before giving up
• Add logging to a file instead of just stdout

That's exactly what the full AI Publishing Pipeline does for ebook generation — 10 chapters instead of 3 tasks, LLM API calls instead of print statements, but the same core pattern.

---

Further Reading

• Building Crash-Recovery State Machines for Long-Running LLM Pipelines in Python
• Python subprocess.run() Deep Dive: Isolation, Timeouts, and Captured Output
• Python's ast Module: Parse, Inspect, and Validate Code at the Token Level