dmemfs.\_fs

March 22, 2026 · View on GitHub

dmemfs._fs

MemoryFileSystem Objects

class MemoryFileSystem()

In-process volatile virtual filesystem with hard quota enforcement.

Provides a POSIX-style hierarchical filesystem entirely within a Python process. All data is stored in RAM; the filesystem ceases to exist when the process exits.

Key properties

Hard quota: every write is rejected before it occurs if it would exceed max_quota. The process is never killed by OOM due to MFS writes.
Zero external dependencies: uses only the Python 3.11+ standard library.
Thread-safe: concurrent reads are permitted; writes are exclusive. Lock acquisition order is always _global_lock → FileNode._rw_lock to prevent deadlocks.
Binary-only modes: open() accepts rb, wb, ab, r+b, and xb only. Text mode raises ValueError.

max_quota : int Maximum total memory budget in bytes that MFS will allocate for file data. Default is 256 MiB. Writes that would exceed this limit raise :exc:MFSQuotaExceededError before any data is written. chunk_overhead_override : int | None Override the per-chunk overhead estimate used for quota accounting in :class:SequentialMemoryFile. None uses the compiled-in default (128 bytes). Increase if quota is unexpectedly exceeded on your platform. promotion_hard_limit : int | None Maximum file size in bytes at which a SequentialMemoryFile may be automatically promoted to RandomAccessMemoryFile. None uses the default of 512 MiB. Set to 0 to disable promotion entirely when combined with default_storage="sequential". max_nodes : int | None Maximum total number of nodes (files + directories) the filesystem may contain. None (default) means unlimited. Exceeding this limit raises :exc:MFSNodeLimitExceededError. default_storage : str Storage backend used when creating new files. One of:

* ``"auto"`` *(default)* — starts as ``SequentialMemoryFile`` and
  promotes to ``RandomAccessMemoryFile`` on the first non-append
  write.
* ``"sequential"`` — always ``SequentialMemoryFile``; promotion is
  disabled and random-access writes raise ``io.UnsupportedOperation``.
* ``"random_access"`` — always ``RandomAccessMemoryFile``.

default_lock_timeout : float | None Default timeout in seconds applied to every open() call that does not provide an explicit lock_timeout. None means block indefinitely (no timeout). Default is 30.0. memory_guard : str Physical-memory guard strategy. One of:

* ``"none"`` *(default)* — no physical-memory checks.
* ``"init"`` — check available RAM once at construction time.
* ``"per_write"`` — re-check available RAM before every write
  (rate-limited by ``memory_guard_interval``).

memory_guard_action : str Action taken when the guard detects low physical memory. One of "warn" (default, emits ResourceWarning ) or "raise" (raises MemoryError ). memory_guard_interval : float Minimum seconds between physical-memory checks for "per_write" mode. Default is 1.0.

Raises

ValueError If default_storage is not one of the accepted values.

Examples

Basic usage::

from dmemfs import MemoryFileSystem

mfs = MemoryFileSystem(max_quota=64 * 1024 * 1024)  # 64 MiB

mfs.mkdir("/work")
with mfs.open("/work/data.bin", "wb") as f:
    f.write(b"hello")

with mfs.open("/work/data.bin", "rb") as f:
    assert f.read() == b"hello"

init

def __init__(max_quota: int = 256 * 1024 * 1024,
             chunk_overhead_override: int | None = None,
             promotion_hard_limit: int | None = None,
             max_nodes: int | None = None,
             default_storage: str = "auto",
             default_lock_timeout: float | None = 30.0,
             memory_guard: str = "none",
             memory_guard_action: str = "warn",
             memory_guard_interval: float = 1.0) -> None

open

def open(path: str,
         mode: str = "rb",
         preallocate: int = 0,
         lock_timeout: float | None = None) -> MemoryFileHandle

Open a file and return a :class:MemoryFileHandle.

Parameters

path : str Absolute path to the file inside MFS. mode : str Binary open mode. Accepted values:

* ``"rb"`` — read-only; raises :exc:`FileNotFoundError` if absent.
  Acquires a **shared** read lock (multiple concurrent readers allowed).
* ``"wb"`` — write-only; creates the file if absent, truncates to
  zero if it exists.  Acquires an **exclusive** write lock.
* ``"ab"`` — append; creates if absent.  Every ``write()`` call
  automatically moves the cursor to EOF before writing.
  Acquires an exclusive write lock.
* ``"r+b"`` — read/write; raises :exc:`FileNotFoundError` if absent.
  Does **not** truncate.  Acquires an exclusive write lock.
* ``"xb"`` — exclusive create; raises :exc:`FileExistsError` if the
  file already exists.  Acquires an exclusive write lock.

Text modes (``"r"``, ``"w"``, etc.) are not supported and raise
:exc:`ValueError`.  Use :class:`MFSTextHandle` for text I/O.

preallocate : int If positive, reserve this many bytes of quota immediately and fill the file with null bytes up to that size. Subsequent writes within the pre-allocated region do not consume additional quota. Default is 0 (no pre-allocation). lock_timeout : float | None Seconds to wait for the file lock. Overrides default_lock_timeout for this call only. None inherits the instance default (30.0 by default). 0.0 is non-blocking.

Returns

MemoryFileHandle An open file handle. Must be closed via with statement or an explicit close() call to release the lock.

Raises

ValueError mode is not one of the five accepted binary modes. FileNotFoundError "rb" or "r+b" and path does not exist; or the parent directory does not exist when creating a new file. FileExistsError "xb" and path already exists. IsADirectoryError path refers to a directory. BlockingIOError The file lock could not be acquired within lock_timeout seconds. MFSQuotaExceededError preallocate would exceed the quota.

Notes

The lock is held for the entire lifetime of the handle (from open() to close()). In multi-threaded code keep handle scopes short and always use the with statement.

mkdir

def mkdir(path: str, exist_ok: bool = False) -> None

Create a directory, including all missing intermediate directories.

Behaves like :func:os.makedirs with parents=True: every missing ancestor is created automatically. This differs from :func:os.mkdir, which raises :exc:FileNotFoundError when a parent is absent.

Parameters

path : str Absolute path of the directory to create. exist_ok : bool If True, do nothing when path already exists as a directory. If False (default), raise :exc:FileExistsError in that case.

Raises

FileExistsError path already exists and exist_ok is False, or path already exists as a file (regardless of exist_ok).

rename

def rename(src: str, dst: str) -> None

Rename or move a file or directory.

Updates the parent node's child mapping; no data is copied. Both files and directories complete in O(depth) time.

Parameters

src : str Existing path. dst : str New path. The parent directory of dst must already exist.

Raises

FileNotFoundError src does not exist, or the parent directory of dst does not exist. FileExistsError dst already exists. BlockingIOError src (or any file beneath it if src is a directory) is currently open. ValueError src is the root directory "/".

Notes

Unlike :meth:move, this method does not create missing intermediate directories for dst. Use :meth:move when automatic parent creation is desired.

Timestamps are not updated (metadata-only operation, consistent with POSIX rename(2) semantics for mtime).

move

def move(src: str, dst: str) -> None

Move a file or directory, creating intermediate directories as needed.

Identical to :meth:rename except that missing parent directories of dst are created automatically.

Parameters

src : str Existing path. dst : str Destination path. Missing parent directories are created.

Raises

FileNotFoundError src does not exist. FileExistsError dst already exists. BlockingIOError src (or any file beneath it) is currently open. ValueError src is the root directory "/".

remove

def remove(path: str) -> None

Delete a single file.

Parameters

path : str Absolute path to the file to delete.

Raises

FileNotFoundError path does not exist. IsADirectoryError path is a directory. Use :meth:rmtree for directories. BlockingIOError The file is currently open.

rmtree

def rmtree(path: str) -> None

Recursively delete a directory and all its contents.

All quota consumed by files in the subtree is released atomically.

Parameters

path : str Absolute path to the directory to delete.

Raises

FileNotFoundError path does not exist. NotADirectoryError path is a file. Use :meth:remove for files. BlockingIOError Any file within the subtree is currently open. ValueError path is the root directory "/".

listdir

def listdir(path: str) -> list[str]

Return the names of entries directly inside path.

Returns bare names only (not full paths), matching the behaviour of :func:os.listdir.

Parameters

path : str Absolute path to a directory.

Returns

list[str] Entry names (files and subdirectories) in unspecified order.

Raises

FileNotFoundError path does not exist. NotADirectoryError path is a file.

Examples

mfs.mkdir("/a/b")
with mfs.open("/a/b/f.txt", "wb") as f:
    f.write(b"x")
mfs.listdir("/a/b")   # → ["f.txt"]

exists

def exists(path: str) -> bool

Return True if path exists (file or directory).

Never raises; returns False for invalid paths.

Parameters

path : str Absolute path to check.

Returns

bool

is_dir

def is_dir(path: str) -> bool

Return True if path exists and is a directory.

Never raises; returns False for invalid paths or non-directories.

Parameters

path : str Absolute path to check.

Returns

bool

is_file

def is_file(path: str) -> bool

Return True if path exists and is a regular file.

Never raises; returns False for invalid paths or directories.

Parameters

path : str Absolute path to check.

Returns

bool

stat

def stat(path: str) -> MFSStatResult

Return metadata for the file or directory at path.

Parameters

path : str Absolute path to a file or directory.

Returns

MFSStatResult A :class:~dmemfs.MFSStatResult TypedDict with the following fields:

* ``size`` (*int*) — file size in bytes; ``0`` for directories.
* ``created_at`` (*float*) — creation timestamp
  (:func:`time.time` format).
* ``modified_at`` (*float*) — last-modification timestamp.
* ``generation`` (*int*) — monotonically increasing change
  counter; ``0`` means the file has not been written since
  creation or last ``import_tree``.
* ``is_dir`` (*bool*) — ``True`` for directories.

Raises

FileNotFoundError path does not exist.

stats

def stats() -> MFSStats

Return an aggregate usage snapshot of the entire filesystem.

Returns

MFSStats A :class:~dmemfs.MFSStats TypedDict with the following fields:

* ``used_bytes`` (*int*) — quota bytes currently consumed.
* ``quota_bytes`` (*int*) — configured maximum quota.
* ``free_bytes`` (*int*) — remaining quota
  (``quota_bytes - used_bytes``).
* ``file_count`` (*int*) — total number of files.
* ``dir_count`` (*int*) — total number of directories.
* ``chunk_count`` (*int*) — total chunks held by
  ``SequentialMemoryFile`` instances (promoted files are not
  counted).
* ``overhead_per_chunk_estimate`` (*int*) — per-chunk overhead
  used for quota accounting.

get_size

def get_size(path: str) -> int

Return the size of a file in bytes.

Parameters

path : str Absolute path to a file.

Returns

int File size in bytes.

Raises

FileNotFoundError path does not exist. IsADirectoryError path is a directory.

export_as_bytesio

def export_as_bytesio(path: str, max_size: int | None = None) -> io.BytesIO

Export a file's contents as a :class:io.BytesIO object.

The returned object is a deep copy of the file data at the moment of the call. Subsequent writes to the MFS file do not affect the returned BytesIO.

Parameters

path : str Absolute path to the file to export. max_size : int | None If given, raise :exc:ValueError before allocating memory when the file exceeds this size. Use to guard against unexpectedly large exports consuming heap memory outside quota control.

Returns

io.BytesIO A new BytesIO positioned at offset 0.

Raises

FileNotFoundError path does not exist. IsADirectoryError path is a directory. ValueError The file size exceeds max_size.

Notes

The returned BytesIO is outside quota management. Large files may cause significant process-memory growth beyond max_quota.

export_tree

def export_tree(prefix: str = "/",
                only_dirty: bool = False) -> dict[str, bytes]

Export all files under prefix as a dict.

Convenience wrapper around :meth:iter_export_tree that materialises the entire result into memory at once. For large filesystems prefer :meth:iter_export_tree to avoid peak-memory doubling.

Parameters

prefix : str Only files whose path starts with prefix are exported. Default is "/" (all files). only_dirty : bool If True, only include files whose generation counter is greater than zero (modified since creation or last import_tree).

Returns

dict[str, bytes] Mapping of absolute path → file bytes.

iter_export_tree

def iter_export_tree(prefix: str = "/",
                     only_dirty: bool = False) -> Iterator[tuple[str, bytes]]

Lazily yield (path, data) pairs for all files under prefix.

The set of paths is snapshotted at the start of iteration under the global lock. Individual file data is read under per-file read locks. Files deleted between snapshot time and read time are silently skipped (weak consistency).

Parameters

prefix : str Path prefix filter. Default "/" exports all files. only_dirty : bool If True, skip files with generation == 0.

Yields

tuple[str, bytes] (absolute_path, file_bytes) for each matched file.

import_tree

def import_tree(tree: dict[str, bytes]) -> None

Import a dict of files into MFS atomically (All-or-Nothing).

If any error occurs during import (quota exceeded, memory allocation failure, etc.) the entire operation is rolled back and MFS is left in its pre-call state.

Parameters

tree : dict[str, bytes] Mapping of absolute path → file bytes. Missing parent directories are created automatically. Existing files at the same paths are replaced.

Raises

BlockingIOError A target path has an open handle. MFSQuotaExceededError The net quota increase would exceed the available quota. MFS is rolled back. MemoryError The OS could not allocate memory for one of the files. MFS is rolled back.

Notes

generation is reset to 0 for all imported files (they are treated as a clean baseline).

copy

def copy(src: str, dst: str) -> None

Copy a single file (deep copy of byte content).

The destination file is a completely independent node with new timestamps. Equivalent to POSIX cp src dst.

Parameters

src : str Source file path. dst : str Destination file path. Must not already exist. The parent directory must exist.

Raises

FileNotFoundError src does not exist, or the parent directory of dst does not exist. FileExistsError dst already exists. IsADirectoryError src is a directory. MFSQuotaExceededError Insufficient quota to store the copied data.

copy_tree

def copy_tree(src: str, dst: str) -> None

Deep-copy an entire directory subtree.

All files are copied as independent byte buffers; no reference sharing (copy-on-write) is performed. Quota for the entire subtree is pre-checked before any node is created; the operation is all-or-nothing with respect to quota.

Parameters

src : str Source directory path. dst : str Destination directory path. Must not already exist. The parent directory must exist.

Raises

FileNotFoundError src does not exist, or the parent of dst does not exist. NotADirectoryError src is not a directory. FileExistsError dst already exists. MFSQuotaExceededError Combined size of all files in the subtree exceeds available quota.

walk

def walk(path: str = "/") -> Iterator[tuple[str, list[str], list[str]]]

Recursively walk the directory tree top-down.

Mirrors the behaviour of :func:os.walk for top-down traversal.

Parameters

path : str Root of the walk. Default "/" walks the entire filesystem.

Yields

tuple[str, list[str], list[str]] (dirpath, dirnames, filenames) for each directory visited, where dirnames are the names of immediate subdirectories and filenames are the names of immediate files.

Raises

FileNotFoundError path does not exist. NotADirectoryError path is a file.

Warnings

Weak consistency: the global lock is not held between yield statements. Concurrent structural changes may cause inconsistencies (e.g. a yielded subdirectory name may no longer exist by the time it is visited). Deleted entries are skipped silently; no exception is raised. For a fully consistent snapshot use :meth:export_tree.

glob

def glob(pattern: str) -> list[str]

Return a sorted list of paths matching pattern.

Parameters

pattern : str Glob pattern with POSIX-style separators. Supported wildcards:

* ``*``  — matches any sequence of characters within a single
  path component (does **not** match ``/``).
* ``**`` — matches zero or more path components (recursive).
* ``?``  — matches exactly one character (not ``/``).
* ``[seq]``, ``[!seq]`` — character classes.

Returns

list[str] Sorted list of absolute paths that match pattern. May be empty.

Examples

mfs.glob("/src/**/*.py")   # all .py files under /src
mfs.glob("/src/*.py")      # .py files directly under /src only

Notes

Applies weak consistency (same as :meth:walk): concurrent structural changes during matching may cause incomplete results but will not raise exceptions.

dmemfs._handle

MemoryFileHandle Objects

class MemoryFileHandle(io.RawIOBase)

Binary file handle returned by :meth:MemoryFileSystem.open.

Wraps a :class:FileNode and provides standard stream semantics over MFS in-memory storage. Inherits :class:io.RawIOBase.

The underlying file lock (read or write) is held for the entire lifetime of the handle — from :meth:MemoryFileSystem.open until :meth:close. Always use the with statement to ensure timely release::

with mfs.open("/data.bin", "rb") as f:
    data = f.read()

Do not instantiate this class directly; always obtain it through :meth:MemoryFileSystem.open.

init

def __init__(mfs: MemoryFileSystem,
             fnode: FileNode,
             path: str,
             mode: str,
             is_append: bool = False) -> None

read

def read(size: int = -1) -> bytes

Read and return up to size bytes.

Parameters

size : int Maximum bytes to read. -1 (default) reads to EOF.

Returns

bytes The data read. Returns b"" when the cursor is already at or past EOF.

Raises

ValueError The handle is closed. io.UnsupportedOperation The handle was opened in a write-only mode ("wb", "ab", or "xb").

write

def write(data: Any) -> int

Write data to the file and return the number of bytes written.

For "ab" mode, the cursor is automatically moved to EOF before each write (POSIX append semantics).

Quota is charged only for the net increase in file size. Overwriting existing bytes (cursor + len(data) ≤ current size) does not consume additional quota.

Parameters

data : bytes | bytearray | memoryview Data to write.

Returns

int Number of bytes written (always len(data) unless the storage raises).

Raises

TypeError data is not a bytes-like object. ValueError The handle is closed. io.UnsupportedOperation The handle was opened in read-only mode ("rb"). MFSQuotaExceededError Writing data would exceed the filesystem quota.

readinto

def readinto(buffer: Any) -> int

seek

def seek(offset: int, whence: int = 0) -> int

Set the stream position and return the new absolute position.

Parameters

offset : int Byte offset relative to the position indicated by whence. whence : int Positioning anchor:

* ``0`` / ``io.SEEK_SET`` — relative to file start;
  *offset* must be ≥ 0.
* ``1`` / ``io.SEEK_CUR`` — relative to current position.
* ``2`` / ``io.SEEK_END`` — relative to EOF;
  *offset* must be ≤ 0 (seeking *past* EOF is not supported).

Returns

int New absolute stream position.

Raises

ValueError whence is not 0, 1, or 2; or the resulting position is negative; or a positive offset is used with SEEK_END. ValueError The handle is closed.

tell

def tell() -> int

Return the current stream position.

Returns

int Current byte offset from the beginning of the file.

Raises

ValueError The handle is closed.

truncate

def truncate(size: int | None = None) -> int

Resize the file to exactly size bytes and return the new size.

If size is larger than the current file size, the file is extended with null bytes (POSIX semantics). If size is smaller, excess data is discarded and the freed quota is released immediately.

Parameters

size : int | None Target size in bytes. None uses the current cursor position. Must be ≥ 0.

Returns

int The new file size.

Raises

ValueError The handle is closed, or size is negative. io.UnsupportedOperation The handle was opened in read-only mode ("rb"). MFSQuotaExceededError Extending the file would exceed the filesystem quota.

flush

def flush() -> None

readable

def readable() -> bool

writable

def writable() -> bool

seekable

def seekable() -> bool

close

def close() -> None

Flush and release the file lock.

Subsequent operations on a closed handle raise :exc:ValueError. Calling close() on an already-closed handle is a no-op.

Notes

Prefer the with statement over explicit close() calls to guarantee cleanup even when exceptions occur.

enter

def __enter__() -> MemoryFileHandle

exit

def __exit__(*args: object) -> None

del

def __del__() -> None

Fallback cleanup: emit ResourceWarning and close if not already closed.

This is a last-resort safety net. It is invoked by the garbage collector and its timing is not guaranteed (especially outside CPython or when circular references exist).

Do not rely on __del__ for deterministic lock release. Always use the with statement.

dmemfs._async

Async wrapper around MemoryFileSystem.

All I/O is delegated to :func:asyncio.to_thread, so the underlying synchronous locks are never held on the event-loop thread.

MemoryFileSystem

MFSStatResult

MFSStats

AsyncMemoryFileHandle Objects

class AsyncMemoryFileHandle()

Async wrapper for a single open-file handle.

init

def __init__(_sync_handle) -> None

read

async def read(size: int = -1) -> bytes

Read bytes. See :meth:MemoryFileHandle.read.

write

async def write(data: bytes) -> int

Write bytes. See :meth:MemoryFileHandle.write.

seek

async def seek(offset: int, whence: int = 0) -> int

Set stream position. See :meth:MemoryFileHandle.seek.

tell

async def tell() -> int

Return current stream position. See :meth:MemoryFileHandle.tell.

truncate

async def truncate(size: int | None = None) -> int

Resize the file. See :meth:MemoryFileHandle.truncate.

flush

async def flush() -> None

Flush the write buffer (no-op for MFS). See :meth:MemoryFileHandle.flush.

readable

async def readable() -> bool

Return True if the handle supports reading.

writable

async def writable() -> bool

Return True if the handle supports writing.

seekable

async def seekable() -> bool

Return True (MFS handles are always seekable).

close

async def close() -> None

Release the file lock. See :meth:MemoryFileHandle.close.

aenter

async def __aenter__() -> AsyncMemoryFileHandle

aexit

async def __aexit__(*args) -> None

AsyncMemoryFileSystem Objects

class AsyncMemoryFileSystem()

Thin async facade over :class:MemoryFileSystem.

Every method delegates to the synchronous implementation via asyncio.to_thread, so the event-loop is never blocked.

init

def __init__(max_quota: int = 256 * 1024 * 1024,
             chunk_overhead_override: int | None = None,
             promotion_hard_limit: int | None = None,
             max_nodes: int | None = None,
             default_storage: str = "auto",
             default_lock_timeout: float | None = 30.0,
             memory_guard: str = "none",
             memory_guard_action: str = "warn",
             memory_guard_interval: float = 1.0) -> None

Parameters

max_quota : int See :class:MemoryFileSystem. chunk_overhead_override : int | None See :class:MemoryFileSystem. promotion_hard_limit : int | None See :class:MemoryFileSystem. max_nodes : int | None See :class:MemoryFileSystem. default_storage : str See :class:MemoryFileSystem. default_lock_timeout : float | None See :class:MemoryFileSystem. Default for async is 30.0. memory_guard : str See :class:MemoryFileSystem. memory_guard_action : str See :class:MemoryFileSystem. memory_guard_interval : float See :class:MemoryFileSystem.

open

async def open(path: str,
               mode: str = "rb",
               preallocate: int = 0,
               lock_timeout: float | None = None) -> AsyncMemoryFileHandle

Open a file and return an :class:AsyncMemoryFileHandle.

All parameters and exceptions are identical to :meth:MemoryFileSystem.open; the operation is offloaded to a thread via :func:asyncio.to_thread.

Returns

AsyncMemoryFileHandle Use with async with to ensure the handle is closed::

    async with await mfs.open("/f.bin", "wb") as f:
        await f.write(b"data")

mkdir

async def mkdir(path: str, exist_ok: bool = False) -> None

Create a directory. See :meth:MemoryFileSystem.mkdir.

rename

async def rename(src: str, dst: str) -> None

Rename or move a file or directory. See :meth:MemoryFileSystem.rename.

move

async def move(src: str, dst: str) -> None

Move a file or directory, creating parents as needed. See :meth:MemoryFileSystem.move.

remove

async def remove(path: str) -> None

Delete a single file. See :meth:MemoryFileSystem.remove.

rmtree

async def rmtree(path: str) -> None

Recursively delete a directory. See :meth:MemoryFileSystem.rmtree.

listdir

async def listdir(path: str) -> list[str]

Return entry names in a directory. See :meth:MemoryFileSystem.listdir.

exists

async def exists(path: str) -> bool

Return True if path exists. See :meth:MemoryFileSystem.exists.

is_dir

async def is_dir(path: str) -> bool

Return True if path is a directory. See :meth:MemoryFileSystem.is_dir.

is_file

async def is_file(path: str) -> bool

Return True if path is a regular file. See :meth:MemoryFileSystem.is_file.

stat

async def stat(path: str) -> MFSStatResult

Return metadata for path. See :meth:MemoryFileSystem.stat.

stats

async def stats() -> MFSStats

Return aggregate filesystem statistics. See :meth:MemoryFileSystem.stats.

get_size

async def get_size(path: str) -> int

Return the size of a file in bytes. See :meth:MemoryFileSystem.get_size.

export_as_bytesio

async def export_as_bytesio(path: str,
                            max_size: int | None = None) -> io.BytesIO

Export file contents as BytesIO. See :meth:MemoryFileSystem.export_as_bytesio.

export_tree

async def export_tree(prefix: str = "/",
                      only_dirty: bool = False) -> dict[str, bytes]

Export files as a dict. See :meth:MemoryFileSystem.export_tree.

import_tree

async def import_tree(tree: dict[str, bytes]) -> None

Import a dict of files atomically. See :meth:MemoryFileSystem.import_tree.

copy

async def copy(src: str, dst: str) -> None

Copy a single file. See :meth:MemoryFileSystem.copy.

copy_tree

async def copy_tree(src: str, dst: str) -> None

Deep-copy a directory subtree. See :meth:MemoryFileSystem.copy_tree.

walk

async def walk(path: str = "/") -> list[tuple[str, list[str], list[str]]]

Walk the directory tree and return results as a list.

Unlike the synchronous :meth:MemoryFileSystem.walk generator, this method collects all results into a list before returning (the generator cannot be safely iterated across thread boundaries).

Returns

list[tuple[str, list[str], list[str]]] Same structure as :func:os.walk top-down results.

glob

async def glob(pattern: str) -> list[str]

Return paths matching pattern. See :meth:MemoryFileSystem.glob.

dmemfs._archive

dmemfs/_archive.py — Pluggable archive extraction adapters.

v15 addition: provides ArchiveAdapter base class, ZipAdapter / TarAdapter standard adapters, and the expand_archive() / expand_archive_streaming() public functions for extracting archives directly into a MemoryFileSystem.

References

spec_v15.md §8 (第8部)
DetailedDesignSpec_v3.md §27
DetailedDesignSpec_test_v3.md §23–§27

ArchiveAdapter Objects

class ArchiveAdapter(ABC)

Base class for archive format adapters.

Subclasses implement members() and _can_handle_impl(). The public can_handle() classmethod handles BytesIO seek restoration and exception suppression automatically — subclasses must not override it; implement _can_handle_impl() instead.

members

@abstractmethod
def members() -> Iterator[tuple[str, bytes]]

Yield (raw_path_in_archive, file_bytes) for each file entry.

Directory entries must not be yielded. Path normalization is the caller's responsibility.

Notes

This is an :func:~abc.abstractmethod. Every concrete adapter subclass must implement this method. Do not yield directory entries; only yield file members.

can_handle

@classmethod
def can_handle(cls, source: str | io.BytesIO) -> bool

Return True if this adapter can process source.

For BytesIO sources the seek position is always restored to its value before this call. Any exception raised by _can_handle_impl is caught and False is returned (fail-safe).

Do not override in subclasses — implement _can_handle_impl.

close

def close() -> None

Release any held resources (default: no-op).

enter

def __enter__() -> Self

exit

def __exit__(*exc: object) -> None

ZipAdapter Objects

class ZipAdapter(ArchiveAdapter)

Adapter for ZIP archives (uses stdlib zipfile).

init

def __init__(source: str | io.BytesIO) -> None

members

def members() -> Iterator[tuple[str, bytes]]

TarAdapter Objects

class TarAdapter(ArchiveAdapter)

Adapter for TAR archives (uses stdlib tarfile).

Supports .tar, .tar.gz, .tar.bz2, .tar.xz. Uses sequential for member in tf: iteration to avoid loading all member metadata into memory at once (lower peak memory than getmembers()).

init

def __init__(source: str | io.BytesIO) -> None

members

def members() -> Iterator[tuple[str, bytes]]

expand_archive

def expand_archive(mfs: "MemoryFileSystem",
                   source: str | io.BytesIO,
                   dest: str = "/",
                   *,
                   on_conflict: Literal["raise", "overwrite"] = "raise",
                   adapter: ArchiveAdapter | None = None,
                   adapters: list[type[ArchiveAdapter]] | None = None) -> None

Extract an archive into mfs atomically (All-or-Nothing).

The extraction delegates to :meth:MemoryFileSystem.import_tree, which provides full rollback on quota exceeded or any other error.

Parameters

mfs: Target MemoryFileSystem instance. source: File path (str) or io.BytesIO containing the archive. dest: Destination prefix inside MFS. Defaults to root "/" (archive paths are appended directly). on_conflict: How to handle collisions:

"raise" (default) — raise FileExistsError / IsADirectoryError and leave MFS unchanged.
"overwrite" — overwrite existing files (emit UserWarning). Directory collisions always raise IsADirectoryError.

Note: "skip" is only available in :func:expand_archive_streaming. adapter: Provide a pre-instantiated :class:ArchiveAdapter directly. Bypasses automatic format detection. Takes priority over adapters. adapters: List of adapter classes for auto-detection. None uses the default [ZipAdapter, TarAdapter].

Raises

ValueError If on_conflict is not "raise" or "overwrite", or if no adapter can handle source. FileExistsError on_conflict="raise" and a file already exists at the destination. IsADirectoryError A directory already exists at the destination path. MFSQuotaExceededError Combined archive data exceeds MFS quota. MFS is rolled back.

expand_archive_streaming

def expand_archive_streaming(
        mfs: "MemoryFileSystem",
        source: str | io.BytesIO,
        dest: str = "/",
        *,
        on_conflict: Literal["raise", "overwrite", "skip"] = "raise",
        adapter: ArchiveAdapter | None = None,
        adapters: list[type[ArchiveAdapter]] | None = None) -> int

Extract an archive into mfs entry-by-entry (streaming).

Unlike :func:expand_archive, this function does not provide All-or-Nothing atomicity. If an error occurs mid-extraction, already extracted files remain in MFS.

Parameters

mfs: Target MemoryFileSystem instance. source: File path (str) or io.BytesIO containing the archive. dest: Destination prefix inside MFS. on_conflict: How to handle collisions:

* ``"raise"`` *(default)* — raise on first conflict, leaving partial
  extraction in MFS.
* ``"overwrite"`` — overwrite existing files (emit ``UserWarning``).
* ``"skip"`` — silently skip both archive-duplicate and MFS-existing
  entries.  Directory collisions always raise ``IsADirectoryError``.

adapter: Pre-instantiated adapter; bypasses auto-detection (takes priority). adapters: Adapter class list for auto-detection.

Returns

int Number of successful write operations performed. When on_conflict is "overwrite" and a duplicate archive entry is encountered, each write is counted individually (so the return value may exceed the number of unique destination paths). Skipped entries are not counted.

Raises

ValueError Invalid on_conflict value or no adapter found for source. FileExistsError on_conflict="raise" and a collision is detected. IsADirectoryError A directory already exists at the destination path (regardless of on_conflict). MFSQuotaExceededError Quota exceeded during write. Partial extraction may remain in MFS.

dmemfs._text

MFSTextHandle: bufferless text I/O helper.

MFS-specific text wrapper used instead of io.TextIOWrapper. Immediate quota checking, no readinto() required, no cookie seek issues.

MFSTextHandle Objects

class MFSTextHandle()

Bufferless text I/O helper that wraps MemoryFileHandle.

Parameters

handle: Binary handle obtained from MemoryFileSystem.open(). encoding: Text encoding (default "utf-8"). errors: Decode error handling (default "strict").

Example

with mfs.open("/data/hello.bin", "wb") as f: ... th = MFSTextHandle(f, encoding="utf-8") ... th.write("こんにちは世界\n")

init

def __init__(handle: MemoryFileHandle,
             encoding: str = "utf-8",
             errors: str = "strict") -> None

encoding

@property
def encoding() -> str

Text encoding.

errors

@property
def errors() -> str

Decode error handling.

write

def write(text: str) -> int

Encode text and write it to the handle.

Parameters

text: The string to write.

Returns

int Number of characters written (not bytes).

read

def read(size: int = -1) -> str

Read bytes and decode them.

Parameters

size: Maximum number of characters to read. -1 reads everything.

Returns

str Decoded text. Returns "" at EOF.

readline

def readline(limit: int = -1) -> str

Read one line.

Recognizes \n, \r\n, and bare \r as line endings.

Parameters

limit: Maximum number of characters to read (-1 means unlimited).

Returns

str One line including the terminating newline, or "" at EOF.

iter

def __iter__() -> Iterator[str]

Line iterator.

def __next__() -> str

enter

def __enter__() -> MFSTextHandle

Return self to support with MFSTextHandle(...) as th: usage.

exit

def __exit__(*args: object) -> None

Exit the context. Does not close the underlying binary handle.

The binary handle's lifecycle is managed by the enclosing with mfs.open(...) block.

dmemfs._exceptions

MFSQuotaExceededError Objects

class MFSQuotaExceededError(OSError)

Raised when the quota limit is exceeded. Subclass of OSError.

init

def __init__(requested: int, available: int) -> None

Parameters

requested : int Number of bytes that were requested. available : int Number of bytes that were available at the time of the request.

MFSNodeLimitExceededError Objects

class MFSNodeLimitExceededError(MFSQuotaExceededError)

Raised when the node count limit is exceeded. Subclass of MFSQuotaExceededError.

init

def __init__(current: int, limit: int) -> None

Parameters

current : int Number of nodes present when the limit was hit. limit : int Configured max_nodes value.

dmemfs._typing

MFSStats Objects

class MFSStats(TypedDict)

Aggregate filesystem usage statistics returned by :meth:MemoryFileSystem.stats.

Fields

used_bytes : int Total quota bytes currently consumed by file data and chunk overhead. quota_bytes : int Configured maximum quota (max_quota passed to the constructor). free_bytes : int Remaining quota (quota_bytes - used_bytes). file_count : int Number of file nodes in the filesystem. dir_count : int Number of directory nodes (including the root). chunk_count : int Total number of chunks held by :class:SequentialMemoryFile instances. Files that have been promoted to :class:RandomAccessMemoryFile are not counted. overhead_per_chunk_estimate : int Per-chunk overhead value used for quota accounting.

used_bytes

quota_bytes

free_bytes

file_count

dir_count

chunk_count

overhead_per_chunk_estimate

MFSStatResult Objects

class MFSStatResult(TypedDict)

Per-path metadata returned by :meth:MemoryFileSystem.stat.

Fields

size : int File size in bytes. Always 0 for directories. created_at : float Creation timestamp in the same format as :func:time.time. modified_at : float Timestamp of the last data modification (write() or truncate()). Equal to created_at for files that have not been written to. generation : int Monotonically increasing change counter. 0 means the file has not been written since creation or the last :meth:~MemoryFileSystem.import_tree call. is_dir : bool True for directories, False for files.

dmemfs._pytest_plugin

pytest fixture plugin.

Usage::

# conftest.py
pytest_plugins = ["dmemfs._pytest_plugin"]

This makes the mfs fixture automatically available::

def test_something(mfs):
    with mfs.open("/a.txt", "wb") as f:
        f.write(b"hello")

MemoryFileSystem

mfs

@pytest.fixture
def mfs() -> MemoryFileSystem

A :class:MemoryFileSystem fixture with default quota (1 MiB).

Provides an independent instance per test (function scope).