用 Python 做磁盘空间预检:预算不足时阻止批处理启动
批处理跑到最后才报 No space left on device,前面的计算已经花了很久。可以在启动前检查真正输出目录所在文件系统的剩余空间,不够就返回失败,让后面的任务不再启动。
本文面向初学者到中级读者,用 Python 标准库实现空间预算检查,并在不填满磁盘的情况下测试失败分支。适合 CSV 导出、备份和 Hugo 构建前的检查;已经写入失败时,先看 Linux 磁盘空间与 inode 排查。
1. 方法:预计峰值增量加保留空间
判断规则是 free_bytes >= (need_mib + reserve_mib) × 1048576。--need-mib 是任务运行期间同时存在的额外数据峰值,--reserve-mib 是希望任务完成后仍留出的余量,默认 1024 MiB。这里的 MiB 是 1024 × 1024 字节。
例如,导出过程中新增 256 MiB,希望保留 1024 MiB,就要求至少 1280 MiB 空闲。若旧产物、新产物和中间文件同时存在,预算要包含新增的各份数据;压缩包大小不能直接当作解压后的空间需求。下面的预算只是示例,应按自己的任务调整。
Python 官方 shutil.disk_usage() 文档说明,该接口返回给定路径的总量、已用量和空闲量,单位是字节。脚本要求输出目录已存在,并解析符号链接;不要只检查 /,因为输出目录可能位于另一个挂载点。
2. 准备环境
需要 Linux、Python 3.9 或更新版本,以及后续入口示例使用的 Bash,无第三方 Python 依赖。本文示例已在 Linux、Python 3.11.4 上测试。
- 创建新的演示目录;重复练习时换一个名字,避免覆盖旧文件:
python3 --version
mkdir disk-preflight-demo
cd disk-preflight-demo
mkdir output
- 把下一节保存为当前目录的
disk_preflight.py,也可下载完整脚本。所有后续命令都在这个演示目录执行。
3. 完整实现
"""Check an existing output directory's free-space budget without writing files."""
import argparse
import json
from pathlib import Path
import shutil
import sys
MIB = 1024 * 1024
def nonnegative_int(text):
value = int(text)
if value < 0:
raise argparse.ArgumentTypeError("must be a nonnegative integer")
return value
def budget_report(directory, need_mib, reserve_mib):
directory = Path(directory).resolve(strict=True)
if not directory.is_dir():
raise NotADirectoryError(f"not a directory: {directory}")
free = shutil.disk_usage(directory).free
required = (need_mib + reserve_mib) * MIB
return {
"path": str(directory),
"free_bytes": free,
"required_bytes": required,
"shortfall_bytes": max(0, required - free),
"ok": free >= required,
}
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
parser.add_argument("directory", type=Path, help="existing output directory")
parser.add_argument("--need-mib", type=nonnegative_int, required=True,
help="estimated peak additional space in MiB")
parser.add_argument("--reserve-mib", type=nonnegative_int, default=1024,
help="space to leave free in MiB (default: 1024)")
args = parser.parse_args(argv)
try:
report = budget_report(args.directory, args.need_mib, args.reserve_mib)
except (OSError, RuntimeError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 2
print(json.dumps(report, ensure_ascii=False))
return 0 if report["ok"] else 1
if __name__ == "__main__":
raise SystemExit(main())
argparse 的自定义类型转换负责拒绝负数、非整数和拼写错误的选项。空间不足是一次有效检查,因此仍会向标准输出写 JSON;目录不存在、读取失败或参数错误则返回 2,不输出 JSON。
| 退出码 | 含义 | 自动化处理 |
|---|---|---|
| 0 | 达到空间预算 | 可以尝试启动任务 |
| 1 | 空闲空间低于预算 | 停止,等待释放空间或调整任务 |
| 2 | 参数或路径检查失败 | 修正配置后再运行 |
脚本只查询信息,不创建目录、不分配空间,也不删除文件。shortfall_bytes 表示距离这次预算还差多少字节。
4. 运行与验证
- 查看帮助,再检查输出目录:
python3 disk_preflight.py --help
python3 disk_preflight.py output --need-mib 256 --reserve-mib 1024
输出包含 path、free_bytes、required_bytes、shortfall_bytes 和 ok。本例的 required_bytes 固定为 1342177280;实际空闲空间和 ok 取决于你的机器。查看路径,确认它确实指向要写入的目录。
- 模拟恰好够用和少 1 MiB 的情况,不消耗磁盘空间:
python3 - <<'PY_CHECK'
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
from disk_preflight import MIB, budget_report
for free_mib, expected in [(1280, True), (1279, False)]:
with patch("disk_preflight.shutil.disk_usage",
return_value=SimpleNamespace(free=free_mib * MIB)):
report = budget_report(Path("output"), 256, 1024)
assert report["ok"] is expected
assert report["shortfall_bytes"] == (0 if expected else MIB)
print("PASS: exact budget accepted; one MiB short rejected")
PY_CHECK
应看到 PASS: exact budget accepted; one MiB short rejected。这是模拟接口返回值的逻辑测试,不能代表真实配额或磁盘写入测试。
- 再用真实文件系统验证退出码。请求超过整个文件系统容量的预算,仅用于计算比较,不会写入大文件:
python3 - <<'PY_FAIL'
import json
import shutil
import subprocess
import sys
need = shutil.disk_usage("output").total // (1024 * 1024) + 1
result = subprocess.run(
[sys.executable, "disk_preflight.py", "output",
"--need-mib", str(need), "--reserve-mib", "0"],
capture_output=True, text=True, check=False,
)
assert result.returncode == 1
assert json.loads(result.stdout)["ok"] is False
print("PASS: insufficient budget returns 1 without allocating disk space")
PY_FAIL
python3 disk_preflight.py missing-output --need-mib 1
python3 disk_preflight.py output --need-mib -1
第一段应打印 PASS: insufficient budget returns 1 without allocating disk space。后两条分别测试目录不存在和负数参数,都会返回 2;测试时可分别用 echo $? 查看。文章示例还检查了普通文件路径、目录符号链接和读取错误。
5. 接入任务入口
把下面内容保存为演示目录的 run_demo.sh。它在预检通过后写一个小型标记文件,便于确认后续命令是否运行;它本身不需要 256 MiB,预算数字用于演示。
#!/usr/bin/env bash
set -eu
cd -- "$(dirname -- "$0")"
mkdir -p output
if python3 disk_preflight.py output --need-mib 256 --reserve-mib 1024; then
python3 -c 'from pathlib import Path; Path("output/completed.txt").write_text("done\n", encoding="utf-8")'
else
result=$?
printf 'Preflight stopped the task (exit %s)\n' "$result" >&2
exit "$result"
fi
运行入口:
bash run_demo.sh
通过时可检查 output/completed.txt;失败时入口保留预检的退出码。重复演示前注意旧标记文件可能仍在,不能仅凭文件存在认定本次成功。接入实际任务时,把标记写入命令替换成自己的导出或构建命令,并使用实际输出路径与预算。
定时运行时使用确定的解释器和工作目录,参考 Python 定时任务路径排错。如果任务同时使用输出目录、临时目录和缓存目录所在的多个文件系统,应分别检查各目录和各自预算;同一文件系统上的并行写入需求则要合并估算。
6. 排错与边界
用同一个任务用户,在同一个容器或挂载环境里检查目标路径:
df -h -- output
df -i -- output
GNU df 官方文档说明路径参数用于查看相关文件系统,-h 显示易读容量,-i 查看 inode。空间够但 inode 不足时,仍可能无法创建新文件。
| 现象 | 具体处理 |
|---|---|
FileNotFoundError |
确认输出路径,先创建真实目录,再执行预检;不要用不存在的目录猜测未来挂载位置。 |
not a directory |
传入输出目录,而不是某个输出文件。 |
Permission denied |
检查任务用户对目录及父目录的访问权限。 |
ok: false |
按只读查找大文件教程定位占用,确认用途后另行处理,或降低任务峰值。 |
| 预检通过但写入失败 | 继续检查用户配额、inode、写权限、只读挂载,以及其他进程是否抢占空间。 |
预检不会预留或锁住磁盘空间。 检查结果随时可能变化;它也不验证写权限或用户配额,不能代替实际写入时的错误处理。应用仍应处理 OSError,并避免把未完成的产物当作成功结果。代码中的 reserve 是计算阈值,不会修改文件系统保留块配置。
7. 小结
在任务启动前,把预计峰值增量和保留空间一起与输出目录的空闲量比较,并用退出码阻止不满足预算的任务启动。先测试通过、空间不足和配置错误三条路径,再接入定时任务;运行过程中仍要处理真实写入失败。
- 原文作者:春江暮客
- 原文链接:https://www.bobobk.com/python-disk-space-preflight.html
- 版权声明:本作品采用 知识共享署名-非商业性使用-禁止演绎 4.0 国际许可协议 进行许可,非商业转载请注明出处(作者,原文链接),商业转载请联系作者获得授权。