はじめに
「テストのためにファイルシステムをモックしたい」「CI でディスクに書かずに一時ファイルを扱いたい」——そういった場面で真っ先に思い浮かぶのが io.BytesIO だと思います。でも、少し複雑なことをしようとすると、BytesIO はすぐに限界を見せます。
この記事では、BytesIO の限界を整理したうえで、その課題をそのまま解決するために作った純粋 Python のインメモリファイルシステムライブラリ D-MemFS を紹介します。
BytesIO の限界
単一バッファにすぎない
io.BytesIO は単一の可変バイト列に対する read/write インターフェースを提供するものです。ファイルを扱う感覚で使えますが、あくまでも「1 つのバッファ」にすぎません。
from io import BytesIO
buf = BytesIO()
buf.write(b"hello, world")
buf.seek(0)
print(buf.read()) # b'hello, world'
これで困るのは、複数のファイルを扱いたいときです。
dict[str, BytesIO] で代替しようとすると…
複数ファイルを扱いたい場合、よく見かける応急処置が辞書です。
from io import BytesIO
# 簡易インメモリ"ファイルシステム"
vfs: dict[str, BytesIO] = {}
def vfs_write(path: str, data: bytes) -> None:
buf = BytesIO(data)
vfs[path] = buf
def vfs_read(path: str) -> bytes:
buf = vfs[path]
buf.seek(0)
return buf.read()
vfs_write("config/settings.json", b'{"debug": true}')
vfs_write("data/input.csv", b"id,name\n1,Alice\n")
print(vfs_read("config/settings.json"))
一見動くように見えますが、すぐに問題が出てきます。
| 欲しい機能 | dict[str, BytesIO] での状況 |
|---|---|
| ディレクトリ作成・一覧 | キープレフィックスで擬似的に実装するしかない |
| サブツリーの削除 |
{k: v for k, v in vfs.items() if not k.startswith(prefix)} を自前で |
| メモリ上限の設定 | ない。無制限に積み上がる |
| スレッドセーフな読み書き | ない。自前でロックを書く |
| ファイル stat (サイズ・更新日時) | 自前で管理するしかない |
| append モード |
buf.seek(0, 2) を自前で |
ディレクトリ構造を真面目に実装しようとすると、もはや「ファイルシステムを自作している」状態になります。しかも、誰がやっても微妙にバグが入ります。
もう一つの落とし穴：メモリが無制限に膨らむ
大量のデータをインメモリで処理するとき、BytesIO はメモリ上限を設定できないため、プロセスが OOM で落ちるまで気づけません。テスト環境で再現しない、本番だけで落ちる——そういったトラブルの温床になります。
D-MemFS を作った
上記の問題をまるごと解決するために、D-MemFS を作りました。
- 標準ライブラリのみ（ゼロ外部依存）
- 階層ディレクトリ構造
- ハードクォータ（書き込み前に拒否）
- RW ロックによるスレッドセーフ
- 非同期ラッパー (
AsyncMemoryFileSystem)
pip install D-MemFS
基本的な使い方
mkdir / open / write / read
from dmemfs import MemoryFileSystem
mfs = MemoryFileSystem()
# ディレクトリ作成
mfs.mkdir("/work/data")
# ファイルへの書き込み
with mfs.open("/work/data/hello.txt", "wb") as f:
f.write(b"Hello, D-MemFS!\n")
# ファイルの読み込み
with mfs.open("/work/data/hello.txt", "rb") as f:
print(f.read()) # b'Hello, D-MemFS!\n'
# stat 情報
st = mfs.stat("/work/data/hello.txt")
print(st["size"]) # 16
print(st["is_dir"]) # False
print(st["modified_at"]) # Unix タイムスタンプ (float)
ディレクトリ操作
# ディレクトリ一覧
for entry in mfs.listdir("/work/data"):
print(entry) # 'hello.txt'
# サブツリーごと削除
mfs.rmtree("/work")
# 存在確認
print(mfs.exists("/work")) # False
テキストファイルを扱う
バイナリモードのみですが、テキスト I/O には MFSTextHandle を使います。
from dmemfs import MemoryFileSystem, MFSTextHandle
mfs = MemoryFileSystem()
mfs.mkdir("/logs")
# テキスト書き込み
with mfs.open("/logs/app.log", "wb") as f:
th = MFSTextHandle(f, encoding="utf-8")
th.write("起動しました\n")
th.write("処理完了\n")
# テキスト読み込み
with mfs.open("/logs/app.log", "rb") as f:
th = MFSTextHandle(f, encoding="utf-8")
for line in th:
print(line, end="")
クォータで暴走を防ぐ
max_quota パラメータを渡すだけで、書き込み前にメモリ使用量を制限できます。
from dmemfs import MemoryFileSystem, MFSQuotaExceededError
# 1 MiB のクォータを設定
mfs = MemoryFileSystem(max_quota=1 * 1024 * 1024)
mfs.mkdir("/data")
try:
with mfs.open("/data/big.bin", "wb") as f:
# 1 MiB を超えようとするとここで例外
f.write(b"x" * (2 * 1024 * 1024))
except MFSQuotaExceededError as e:
print(f"クォータ超過: {e}")
# → クォータ超過: quota exceeded (limit=1048576, used=0, requested=2097152)
MFSQuotaExceededError は書き込みが実行される前に発生します。中途半端な状態でファイルが汚染されることはありません。
ファイル数の上限も設定できます。
from dmemfs import MemoryFileSystem, MFSNodeLimitExceededError
mfs = MemoryFileSystem(max_nodes=4)
mfs.mkdir("/data")
mfs.open("/data/a.txt", "xb").close()
mfs.open("/data/b.txt", "xb").close()
try:
mfs.open("/data/c.txt", "xb").close()
except MFSNodeLimitExceededError as e:
print(f"ノード数超過: {e}")
import_tree / export_tree で一括操作
ファイルをまとめてインポート・エクスポートする機能も備えています。
import zipfile
import io
from dmemfs import MemoryFileSystem
mfs = MemoryFileSystem()
# dict[str, bytes] 形式でまとめてインポート
mfs.import_tree({
"/snapshot/config.json": b'{"debug": true}',
"/snapshot/data.csv": b"id,name\n1,Alice\n",
})
# 処理...
with mfs.open("/snapshot/config.json", "rb") as f:
config = f.read()
# export_tree でまとめて取得し、ZIP に書き出す
exported = mfs.export_tree("/snapshot")
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
for path, data in exported.items():
zf.writestr(path.lstrip("/"), data)
with open("snapshot.zip", "wb") as f:
f.write(buf.getvalue())
「処理中はすべてメモリ上で完結し、最後だけ書き出す」というパターンが自然に書けます。
比較まとめ
| 機能 | BytesIO |
dict[str, BytesIO] |
D-MemFS |
|---|---|---|---|
| 単一ファイル I/O | ✅ | ✅ | ✅ |
| 階層ディレクトリ | ❌ | △ 擬似的に | ✅ |
| ディレクトリ一覧 | ❌ | △ 自前実装 | ✅ |
| サブツリー削除 | ❌ | △ 自前実装 | ✅ |
| append モード | △ 手動 seek | △ 手動 seek | ✅ |
| ハードクォータ | ❌ | ❌ | ✅ |
| stat (サイズ・日時) | ❌ | ❌ | ✅ |
| スレッドセーフ | ❌ | ❌ | ✅ |
| 外部依存 | なし | なし | なし（標準ライブラリのみ） |
インストール
pip install D-MemFS
Python 3.11 以上が必要です。外部依存はゼロです。
おわりに
BytesIO は「1 つのバッファ」として優秀ですが、ファイルシステムの代わりにはなれません。dict[str, BytesIO] で自前実装すると、毎回同じバグを踏み続けます。
D-MemFS は、インメモリでファイルシステムが必要なあらゆる場面——テスト、CI、一時的なデータ処理パイプライン——で使えるように設計しています。まず pip install D-MemFS して、TemporaryDirectory を置き換えてみてください。
🛡️ D-MemFS 活用ガイド（連載）
- 第1弾：BytesIO じゃダメな理由 — Python インメモリ FS ライブラリを作った話（本記事）
- 第2弾：管理者権限不要でWindowsにRAMディスクを構築する ― Python環境のI/O高速化手法
- 第3弾：PythonのOOMキルを完全防御する：BytesIOの罠とD-MemFS「ハードクォータ」の設計思想
- 第4弾：PythonでSQLiteの共有インメモリDBが消える仕様と、
deserialize()の落とし穴をbackup()で回避した話
📖 設計の裏側をもっと深く
なぜこのライブラリを作ったのか、その原体験や「バイブコーディング消耗からの脱却」としてのSDD（仕様駆動開発）の実践記録については、Zenn で連載しています。あわせてどうぞ。
GitHub で Star ⭐ をいただけると、開発の大きな励みになります！
（↓以下のリンク先がプロジェクトの本体です）