春江暮客

春江暮客的个人学习分享网站

用 Python 做磁盘空间预检:预算不足时阻止批处理启动

2026-10-17 技术
用 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 上测试。

  1. 创建新的演示目录;重复练习时换一个名字,避免覆盖旧文件:
python3 --version
mkdir disk-preflight-demo
cd disk-preflight-demo
mkdir output
  1. 把下一节保存为当前目录的 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. 运行与验证

  1. 查看帮助,再检查输出目录:
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. 模拟恰好够用和少 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。这是模拟接口返回值的逻辑测试,不能代表真实配额或磁盘写入测试。

  1. 再用真实文件系统验证退出码。请求超过整个文件系统容量的预算,仅用于计算比较,不会写入大文件:
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. 小结

在任务启动前,把预计峰值增量和保留空间一起与输出目录的空闲量比较,并用退出码阻止不满足预算的任务启动。先测试通过、空间不足和配置错误三条路径,再接入定时任务;运行过程中仍要处理真实写入失败。

标签

1024 12306 ablang adsense agents.md ai ai-agent ai-agents ai-seo algorithm amp antibodies apparmor argparse automation backup bash batch-processing binarycif bioinformatics biopython blockchain boltz bootstrapping boxes bubblewrap c-index cca cdn chatgpt checkpoint cli cloudflare codex cofoldarena copy cpu监控 cron csv cuda curl data-leakage data-processing data-quality data-validation datascience datavisualization datetime debugging deployment desktop-app devtools disk-space disown docker dovecot download electron encoding esm esm2 esm3 esmc esmfold2 faceswap fasta fastmcp faulthandler ffmpeg file-descriptors file-io flashppi flask folium frontend game generator git github-actions google google-research grep gzip harness hls html http hugo indexnow javascript jev json jsonl just k-means kaggle langfuse leecode linux list litellm llm llms.txt logging logs lollipop lsof m3u8 machine-learning macos manacher matplotlib mcp mirror mmcif model-evaluation monitoring mp3 mp4 mpnn multiomics mutation mysql nanobert nanobodies networkx nginx normalize numpy ollama omegatherm pandas password pdb pep-723 phaser pillow pip postfix preprocessing print protein-alignment protein-design protein-embeddings protein-interactions protein-language-models protein-stability protein-structure proxy pydantic pyecharts pyqt python python3 r raincloud reproducibility requests reservoir-sampling rfoptimization rg ripgrep rosettafold3 roundcube rrsi rsi rsync s-tui sampling scale scikit-learn scrapy screen seaborn security selenium seo sequence-identity sha256 signals sklearn solana somaticsignatures spl sqlite ssh standardize static-site subprocess symlink sysadmin system-one tensorflow tkinter toml tron tronpy troubleshooting turtle typesafe-ai ubuntu usdt uv vhhbert vite webp wordcloud wordpress workflow yaml zoneinfo 后台 寓言 概率 经济 贸易 迅雷解析 钱包

友情链接

其它