Check Scheduled Output Freshness with Python: File Age, Empty Files, and Exit Codes
A scheduled job logs “started,” and report.json exists in the output directory, but that file may be yesterday’s result. Checking only a process exit code or file existence can miss a job that never produces a new output.
This beginner-to-intermediate tutorial uses the Python standard library to check an output’s modification time, byte size, and file type, emit JSON, and stop downstream commands with an exit code. Use it for local reports, exports, and an additional check after a static-site build. No packages or scheduler changes are needed.
1. Method: define an acceptable output
We use a small set of rules:
- The path must provide metadata and point to a regular file.
- It must contain at least
--min-bytesbytes, defaulting to 1; empty files trigger an alert. - Its last modification must be no more than
--max-ageseconds ago; the exact limit is accepted. - A modification time more than 5 seconds ahead of the current time also triggers an alert, so a future timestamp cannot keep old output looking healthy.
Python’s stat-result documentation defines st_mtime as the last content-modification time and st_size as the size in bytes. On Linux, st_ctime records metadata changes; it is not a creation timestamp. We use stat.S_ISREG() to identify regular files.
File existence and timely production are separate checks. If a scheduled script looks in the wrong directory, first follow scheduled-job path and environment troubleshooting to fix its working directory and interpreter path.
2. Setup and test files
Use Python 3.9 or newer, with no third-party dependencies. These examples were tested on Linux with Python 3.11.4. Commands work in bash or zsh.
- Create a fresh demo directory and keep subsequent commands in the same terminal. Choose a new directory name when repeating the experiment:
python3 --version
mkdir freshness-demo
cd freshness-demo
freshness_root="$PWD"
python3 - <<'PY_DATA'
from pathlib import Path
import os
import time
now = time.time()
for name, content, modified in [
("fresh.txt", "ready\n", now),
("stale.txt", "ready\n", now - 7200),
("empty.txt", "", now),
("future.txt", "ready\n", now + 3600),
]:
Path(name).write_text(content, encoding="utf-8")
os.utime(name, (modified, modified))
PY_DATA
- Save the next section as
check_freshness.pyin the current directory.fresh.txtis recent,stale.txtis dated two hours ago,empty.txtis empty, andfuture.txtis dated one hour ahead.
os.utime() sets access and modification times. This changes only demo files, without waiting an hour or adjusting the system clock. Complete sections 4 and 5 within an hour of creating the fixtures; otherwise rerun the PY_DATA snippet above.
3. Complete implementation: inspect metadata and emit JSON
"""Check a completed output file's age and size. Python 3.9+, standard library."""
import argparse
import json
from pathlib import Path
import stat
import sys
import time
def nonnegative_int(value):
number = int(value)
if number < 0:
raise argparse.ArgumentTypeError("must be zero or greater")
return number
def inspect_file(path, max_age, min_bytes):
info = path.stat()
if not stat.S_ISREG(info.st_mode):
raise ValueError("not a regular file")
age = time.time() - info.st_mtime
problems = []
if info.st_size < min_bytes:
problems.append("too_small")
if age < -5:
problems.append("future_mtime")
elif age > max_age:
problems.append("too_old")
return {
"file": str(path),
"status": "alert" if problems else "ok",
"age_seconds": round(age, 3),
"size_bytes": info.st_size,
"problems": problems,
}
def main():
parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
parser.add_argument("file", type=Path, help="completed output to inspect")
parser.add_argument("--max-age", type=nonnegative_int, required=True,
help="maximum age in seconds; must be greater than zero")
parser.add_argument("--min-bytes", type=nonnegative_int, default=1,
help="minimum size in bytes (default: 1)")
args = parser.parse_args()
if args.max_age == 0:
parser.error("--max-age must be greater than zero")
try:
result = inspect_file(args.file, args.max_age, args.min_bytes)
except (OSError, ValueError) as exc:
print(f"ERROR: {args.file}: {exc}", file=sys.stderr)
return 1
print(json.dumps(result, ensure_ascii=False))
return 1 if result["problems"] else 0
if __name__ == "__main__":
raise SystemExit(main())
The script reads metadata without reading or modifying the output’s contents. Path.stat() follows symbolic links and checks the target file; a broken link raises an error. Directories, devices, and FIFOs cannot qualify as valid outputs.
File timestamps and time.time() use comparable time references. Do not subtract a file timestamp from time.monotonic(). Wall-clock adjustments affect the age calculation; this example tolerates at most 5 seconds into the future. age_seconds is rounded to three decimal places for display; comparisons use the unrounded value.
4. Usage and exit codes
- Show help, then check the recent and old files:
python3 check_freshness.py --help
python3 check_freshness.py fresh.txt --max-age 3600
python3 check_freshness.py stale.txt --max-age 3600
The recent file produces output like this; age_seconds depends on when you run it:
{"file": "fresh.txt", "status": "ok", "age_seconds": 2.314, "size_bytes": 6, "problems": []}
The old file reports status: "alert" and problems: ["too_old"], with exit code 1. Empty files and future timestamps produce too_small and future_mtime, respectively. One file can have multiple problems.
- Use these exit-code rules in the caller:
| Exit code | Meaning |
|---|---|
| 0 | A regular file passes both age and size checks. |
| 1 | Output alert, missing file, access error, or incorrect file type prevents a successful check. |
| 2 | Invalid command-line arguments, such as --max-age 0 or a negative value. |
When regular-file metadata is available, standard output contains one JSON line, including alert results. Missing paths, access errors, and file-type errors produce only ERROR: on standard error. argparse writes argument errors to standard error; see its official documentation. An absent successful JSON report does not mean there are no problems.
5. Validate failure branches without waiting
Run this self-check to verify both results and exit codes. It explicitly permits subprocess failures, then checks their actual return values:
python3 - <<'PY_CHECK'
import json
import subprocess
import sys
cases = [
(["fresh.txt", "--max-age", "3600"], 0, []),
(["stale.txt", "--max-age", "3600"], 1, ["too_old"]),
(["empty.txt", "--max-age", "3600"], 1, ["too_small"]),
(["future.txt", "--max-age", "3600"], 1, ["future_mtime"]),
(["missing.txt", "--max-age", "3600"], 1, None),
([".", "--max-age", "3600"], 1, None),
(["fresh.txt", "--max-age", "0"], 2, None),
]
for arguments, expected_code, expected_problems in cases:
result = subprocess.run(
[sys.executable, "check_freshness.py", *arguments],
capture_output=True, text=True, check=False,
)
assert result.returncode == expected_code, (arguments, result)
if expected_problems is None:
assert result.stdout == "" and result.stderr, (arguments, result)
else:
report = json.loads(result.stdout)
assert report["problems"] == expected_problems, report
assert report["status"] == ("alert" if expected_problems else "ok"), report
print("OK: fresh, stale, empty, future, missing, directory, bad arguments")
PY_CHECK
Expected output:
OK: fresh, stale, empty, future, missing, directory, bad arguments
The check does not rewrite the fixtures. Missing paths and directories must return nonzero, and invalid arguments must not silently succeed. Handle expected failures with conditionals when inspecting them in a shell; an unhandled exit code 1 can stop a script using set -e.
6. Integrate with an existing workflow and troubleshoot
- Check the actual result with absolute paths. This example still checks only the demo file and changes no scheduler settings:
python3 "$freshness_root/check_freshness.py" \
"$freshness_root/fresh.txt" --max-age 3600 --min-bytes 1 \
&& printf 'Output is recent and nonempty\n'
The command after && runs only if the check succeeds, and a failed check retains its nonzero exit code. To detect a producer that never starts, invoke the checker periodically from an independent existing monitoring entry point. A checker embedded inside the producer cannot run when the producer never starts. For an hourly report, --max-age 5400 is a starting point that allows extra runtime and scheduling delay; adjust it to observed behavior.
After a Hugo build, you can also check the target HTML in that build’s output directory. Its timestamp cannot prove content or links are correct; continue with Hugo internal-link and image checks for content validation.
- Fix the cause indicated by each problem, then check again:
| Symptom | Concrete action |
|---|---|
too_old |
Check the producer’s latest actual success, input, and output paths, then regenerate the output. If a copy tool preserves old modification times, monitor a status file generated after this run succeeds instead. |
too_small |
Inspect producer errors and free disk space. Use --min-bytes 0 only when empty output is valid for the task. A byte threshold is not a record count. |
future_mtime |
Check clocks on the producer and monitoring hosts, plus timestamps preserved during copying. Do not hide the problem by increasing limits indefinitely. |
No such file |
Verify the mount, output name, and producing step with an absolute path. |
Permission denied |
Check parent-directory access as the account running the checker. Being able to stat a file does not prove its contents are readable. |
not a regular file |
Point to the specific file; a directory’s timestamp cannot stand in for a report’s timestamp. |
- Regenerate the old demo output and verify recovery:
python3 - <<'PY_REPAIR'
from pathlib import Path
Path("stale.txt").write_text("ready\n", encoding="utf-8")
PY_REPAIR
python3 check_freshness.py stale.txt --max-age 3600
It should now return status: "ok" and exit code 0. In production, do not use touch merely to clear an alert: it does not generate missing business data.
A recent, nonempty file can still contain incomplete or incorrect data. The producer should validate content and replace the final file only after success; see atomic JSON writes in Python for a saving pattern. Metadata observed during a write cannot prove completion. Copies preserving modification times, filesystem timestamp precision, and network-filesystem caching can also affect the result.
7. Summary
Checking both maximum age and minimum size catches leftover old files and empty outputs while giving automation a clear exit code. Validate failure branches with controlled timestamps before checking production paths. Keep content validation and replacement after success in the producer so recent updates are not mistaken for correct results.
- 原文作者:春江暮客
- 原文链接:https://www.bobobk.com/en/python-output-freshness-check.html
- 版权声明:本作品采用 知识共享署名-非商业性使用-禁止演绎 4.0 国际许可协议 进行许可,非商业转载请注明出处(作者,原文链接),商业转载请联系作者获得授权。
相关文章
- Find Large Files with Python: Rank a Directory with os.walk and heapq
- Python Too Many Open Files: Find File Handle Leaks on Linux
- Stream gzip Logs with Python and Detect Truncation and CRC Errors
- Fix Delayed Python Logs with -u, flush, and Bash pipefail
- Troubleshoot Linux Scheduled Jobs: Python Paths, Working Directories, and Environment