Wait for a File in Python: Timeouts, Atomic Handoffs, and Exit Codes
On this page
An export is still writing JSON when the next script starts: sometimes the file is missing, sometimes it contains half a document. Increasing sleep(10) is a guess that also delays successful runs.
This beginner-to-intermediate tutorial uses Python’s standard library to wait for a file with a timeout, while the producer publishes a complete result through atomic replacement. Use it for small local reports, batch summaries, and post-build helper tasks, without installing packages or changing a scheduler.
1. Method: agree on a handoff before polling
- Use a new output directory for each run so an old result cannot pass as a new one.
- The producer writes a sibling temporary file, closes it, and publishes the final path with
os.replace(). It does not modify that result afterward. - The consumer waits only for a missing path. Invalid JSON, directories, and permission errors fail immediately.
- Calculate the remaining wait with
time.monotonic()and cap each sleep at that remaining budget.
Python’s monotonic documentation says this clock is unaffected by system-clock updates, making it suitable for elapsed durations. The os.replace documentation describes successful renaming as atomic on POSIX, while cross-filesystem operations may fail. Keep the temporary file beside the destination.
This handoff is safer than checking that file size stayed unchanged twice: a writer can pause without finishing. See atomic JSON writes for a fuller writing workflow; this tutorial focuses on consumer waiting and failure boundaries.
2. Setup: use a separate demo directory
Use Python 3.9 or newer, with no third-party dependencies. The examples were tested on Linux with Python 3.11.4; shell commands work in bash and zsh. Run these in the same terminal:
python3 --version
mkdir wait-file-demo
cd wait-file-demo
mkdir run-001
Choose a new directory name when repeating the experiment. Do not reuse an existing run-001/result.json. Save the next section as wait_json.py in the current directory.
3. Implementation: retry only missing files
"""Wait for a small, trusted local JSON output. Python 3.9+, no dependencies."""
import argparse
import json
import math
from pathlib import Path
import sys
import time
def seconds(value):
number = float(value)
if not math.isfinite(number) or not 0 <= number <= 86400:
raise argparse.ArgumentTypeError("must be finite and between 0 and 86400")
return number
def wait_json(path, timeout, interval, clock=time.monotonic, sleep=time.sleep):
deadline = clock() + timeout
first = True
while first or clock() < deadline:
first = False
try:
with path.open(encoding="utf-8") as stream:
return json.load(stream)
except FileNotFoundError:
remaining = deadline - clock()
if remaining <= 0:
break
sleep(min(interval, remaining))
raise TimeoutError(f"no file before deadline: {path}")
def main():
parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
parser.add_argument("file", type=Path)
parser.add_argument("--timeout", type=seconds, default=30.0,
help="wait budget in seconds; 0 means one immediate attempt")
parser.add_argument("--interval", type=seconds, default=0.2,
help="poll interval in seconds; minimum 0.001")
args = parser.parse_args()
if args.interval < 0.001:
parser.error("--interval must be at least 0.001 seconds")
try:
data = wait_json(args.file, args.timeout, args.interval)
except TimeoutError as exc:
print(f"TIMEOUT: {exc}", file=sys.stderr)
return 1
except (OSError, ValueError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 2
except KeyboardInterrupt:
return 130
print(json.dumps({"status": "ready", "data": data}, ensure_ascii=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
--timeout 0 still attempts one immediate read. Otherwise, the first attempt starts immediately and later attempts start only before the deadline; the remaining time limits the final sleep. Scheduling can extend a sleep, so this is not a real-time guarantee.
Read only small, trusted, regular JSON files in a local directory. Do not use FIFOs, devices, attacker-controlled paths, or network mounts that might hang. The timeout limits polling waits; it cannot interrupt an open, read, or JSON parse already in progress. json.load() does not validate business fields; see the official JSON documentation.
4. Usage: simulate delayed publication
- Check the help first:
python3 wait_json.py --help
- Start a background producer while the consumer waits. Run this in an interactive shell without
set -e, so both process exit codes can be saved:
python3 - <<'PY_PRODUCER' &
import json
import os
from pathlib import Path
import tempfile
import time
output = Path("run-001/result.json")
pending = None
try:
time.sleep(0.3) # Simulate work before publishing.
with tempfile.NamedTemporaryFile(
mode="w", encoding="utf-8", dir=output.parent,
prefix=".pending-", delete=False,
) as stream:
pending = Path(stream.name)
json.dump({"rows": 3}, stream, allow_nan=False)
stream.write("\n")
os.replace(pending, output)
finally:
if pending is not None:
pending.unlink(missing_ok=True)
PY_PRODUCER
producer_pid=$!
python3 wait_json.py run-001/result.json --timeout 5 --interval 0.1
consumer_exit=$?
wait "$producer_pid"
producer_exit=$?
printf 'consumer=%s producer=%s\n' "$consumer_exit" "$producer_exit"
NamedTemporaryFile() generates a unique temporary name. delete=False keeps the closed file available for renaming; finally cleans up this temporary file. Normal output is:
{"status": "ready", "data": {"rows": 3}}
consumer=0 producer=0
If you run the commands slowly, publication might finish before the consumer starts; immediate success is still correct. The example does not promise power-loss durability or coordinate multiple producers writing one path. It uses one producer and a new directory per run.
| Exit code | Meaning |
|---|---|
| 0 | JSON was read and parsed; stdout contains one result line. |
| 1 | No file was obtained before the deadline; stderr starts with TIMEOUT:. |
| 2 | Content, access, or argument error; only stderr is written. |
| 130 | Ctrl-C during waiting or reading. |
Chain a next command conditionally with python3 wait_json.py ... && next-command. Do not hide failures with || true. For retained outputs, use output freshness and empty-file checks, but modification time is not a substitute for a run identifier either.
5. Validation: success, timeout, and error exits
This self-check creates and removes only its own fixtures, allows child processes to fail, and then checks their actual exit codes:
python3 - <<'PY_CHECK'
from pathlib import Path
import subprocess
import sys
import tempfile
with tempfile.TemporaryDirectory(prefix="checks-", dir=".") as folder:
root = Path(folder)
(root / "good.json").write_text('{"rows": 3}\n', encoding="utf-8")
(root / "bad.json").write_text('{"rows":', encoding="utf-8")
cases = [
("ready", "good.json", "0", 0, "--interval", "0.05"),
("missing", "missing.json", "0.1", 1, "--interval", "0.05"),
("invalid", "bad.json", "0", 2, "--interval", "0.05"),
("bad-argument", "good.json", "0", 2, "--interval", "0"),
]
for label, name, timeout, expected, flag, value in cases:
result = subprocess.run(
[sys.executable, "wait_json.py", str(root / name),
"--timeout", timeout, flag, value],
capture_output=True, text=True, timeout=5,
)
assert result.returncode == expected, (label, result.stderr)
if expected == 0:
assert '"status": "ready"' in result.stdout
assert not result.stderr
else:
assert not result.stdout and result.stderr
print(f"PASS {label} exit={result.returncode}")
PY_CHECK
Expected output:
PASS ready exit=0
PASS missing exit=1
PASS invalid exit=2
PASS bad-argument exit=2
Section 4 covers delayed publication. To extend polling into recurring monitoring, see fixed-interval tasks with monotonic, but an endless loop is not a timeout policy.
6. Troubleshooting: do not hide problems with a longer sleep
| Symptom | Cause and direct fix |
|---|---|
Persistent TIMEOUT: |
Confirm both paths and working directories match; inspect the saved producer_exit. Diagnose with absolute paths before selecting --timeout from normal execution times. |
ERROR: Expecting ... |
The final file contains incomplete JSON. Close the producer’s temporary file before replacement; do not retry parse errors forever as though the path were missing. |
| Immediate success with old data | A previous path was reused. Create a new output directory for each run; do not blindly delete old files in a shared directory. |
Permission denied or a directory path |
Use a regular file readable by the current user and check parent-directory traversal permissions. Do not immediately retry as root. |
| Still running after the timeout | Blocking I/O is outside this polling budget. Inspect the mount or file type; if needed, run the entire consumer in a managed subprocess with an external time limit. |
7. Summary
Define how a complete file is published, then wait using a monotonic deadline and bounded retries. Retry missing paths only, fail promptly on content or access errors, and use a new directory for each run to avoid stale results. This fits small local file handoffs; cross-machine notifications, enforced I/O deadlines, and multiple producers need an appropriate task-management mechanism.
- Author: yanzhi
- Original article: https://www.bobobk.com/en/python-wait-for-file-timeout.html
- License: This work is licensed under CC BY-NC-ND 4.0. See the license for attribution, noncommercial use and no-derivatives terms.
Related articles
- Test Directory Writes with Python: Permissions, Read-Only Storage, and Path Errors
- Python File Permissions on Linux: umask, Mode 600, and No Overwrites
- Merge Sorted Logs by Timestamp in Python with heapq.merge
- Read the Last N Log Lines in Python with deque
- Missing Hugo Posts: Diagnose Drafts, Publication Dates, Expiry, and Output Paths