python · barneygale · May 12, 2025 · May 16, 2025 · encukou · May 17, 2025
@@ -28,7 +28,7 @@

 from pathlib._os import (
    PathInfo, DirEntryInfo,
-    magic_open, vfspath,
+    vfsopen, vfspath,
    ensure_different_files, ensure_distinct_paths,
    copyfile2, copyfileobj, copy_info,
 )
@@ -766,6 +766,27 @@ def samefile(self, other_path):
        return (st.st_ino == other_st.st_ino and
                st.st_dev == other_st.st_dev)

+    def __open_reader__(self):
+        """
+        Open the file pointed to by this path for reading in binary mode and
+        return a file object.
+        """
+        return io.open(self, 'rb')
+
+    def __open_writer__(self, mode):
+        """
+        Open the file pointed to by this path for writing in binary mode and
+        return a file object.
+        """
+        return io.open(self, f'{mode}b')
+
+    def __open_updater__(self, mode):
+        """
+        Open the file pointed to by this path for updating in binary mode and
+        return a file object.
+        """
+        return io.open(self, f'{mode}+b')
+
    def open(self, mode='r', buffering=-1, encoding=None,
             errors=None, newline=None):
        """
@@ -1141,7 +1162,7 @@ def _copy_from(self, source, follow_symlinks=True, preserve_metadata=False):

    def _copy_from_file(self, source, preserve_metadata=False):
        ensure_different_files(source, self)
-        with magic_open(source, 'rb') as source_f:
+        with vfsopen(source, 'rb') as source_f:
            with open(self, 'wb') as target_f:
                copyfileobj(source_f, target_f)
        if preserve_metadata:

@@ -166,48 +166,78 @@ def copyfileobj(source_f, target_f):
        write_target(buf)


-def magic_open(path, mode='r', buffering=-1, encoding=None, errors=None,
-               newline=None):
+def _open_reader(obj):
+    cls = type(obj)
+    try:
+        return cls.__open_reader__(obj)
+    except AttributeError:
+        if hasattr(cls, '__open_reader__'):
+            raise
+    raise TypeError(f"{cls.__name__} can't be opened for reading")
-    try:
-        return cls.__open_reader__(obj)
-    except AttributeError:
-        if hasattr(cls, '__open_reader__'):
-            raise
-    raise TypeError(f"{cls.__name__} can't be opened for reading")
+    try:
+        open_reader = cls.__open_reader__
+    except AttributeError:
+        raise TypeError(f"{cls.__name__} can't be opened for reading")
+    else:
+        return open_reader(obj)
-    try:
-        return cls.__open_reader__(obj)
-    except AttributeError:
-        if hasattr(cls, '__open_reader__'):
-            raise
-    raise TypeError(f"{cls.__name__} can't be opened for reading")
+    try:
+        open_reader = cls.__open_reader__
+    except AttributeError:
+        raise TypeError(f"{cls.__name__} can't be opened for reading")
+    else:
+        return open_reader(obj)
+
+
+def _open_writer(obj, mode):
+    cls = type(obj)
+    try:
+        return cls.__open_writer__(obj, mode)
+    except AttributeError:
+        if hasattr(cls, '__open_writer__'):
+            raise
+    raise TypeError(f"{cls.__name__} can't be opened for writing")
+
+
+def _open_updater(obj, mode):
+    cls = type(obj)
+    try:
+        return cls.__open_updater__(obj, mode)
+    except AttributeError:
+        if hasattr(cls, '__open_updater__'):
+            raise
+    raise TypeError(f"{cls.__name__} can't be opened for updating")
+
+
+def vfsopen(obj, mode='r', buffering=-1, encoding=None, errors=None,
+            newline=None):
    """
    Open the file pointed to by this path and return a file object, as
    the built-in open() function does.
+
+    Unlike the built-in open() function, this function accepts 'openable'
+    objects, which are objects with any of these magic methods:
-    objects, which are objects with any of these magic methods:
+    objects, which are objects with any of these special methods:
-    objects, which are objects with any of these magic methods:
+    objects, which are objects with any of these special methods:
+
+        __open_reader__()
+        __open_writer__(mode)
+        __open_updater__(mode)
+
+    '__open_reader__' is called for 'r' mode; '__open_writer__' for 'a', 'w'
+    and 'x' modes; and '__open_updater__' for 'r+' and 'w+' modes. If text
+    mode is requested, the result is wrapped in an io.TextIOWrapper object.
    """
    text = 'b' not in mode
-    if text:
+    if buffering != -1:
+        raise ValueError("buffer size can't be customized")
+    elif text:
        # Call io.text_encoding() here to ensure any warning is raised at an
        # appropriate stack level.
        encoding = text_encoding(encoding)
-    try:
-        return open(path, mode, buffering, encoding, errors, newline)
-    except TypeError:
-        pass
-    cls = type(path)
-    mode = ''.join(sorted(c for c in mode if c not in 'bt'))
-    if text:
-        try:
-            attr = getattr(cls, f'__open_{mode}__')
-        except AttributeError:
-            pass
-        else:
-            return attr(path, buffering, encoding, errors, newline)
    elif encoding is not None:
        raise ValueError("binary mode doesn't take an encoding argument")
    elif errors is not None:
        raise ValueError("binary mode doesn't take an errors argument")
    elif newline is not None:
        raise ValueError("binary mode doesn't take a newline argument")
-
-    try:
-        attr = getattr(cls, f'__open_{mode}b__')
-    except AttributeError:
-        pass
+    mode = ''.join(sorted(c for c in mode if c not in 'bt'))
+    if mode == 'r':
+        stream = _open_reader(obj)
+    elif mode in ('a', 'w', 'x'):
+        stream = _open_writer(obj, mode)
+    elif mode in ('+r', '+w'):
+        stream = _open_updater(obj, mode[1])
    else:
-        stream = attr(path, buffering)
-        if text:
-            stream = TextIOWrapper(stream, encoding, errors, newline)
-        return stream
-
-    raise TypeError(f"{cls.__name__} can't be opened with mode {mode!r}")
+        raise ValueError(f'invalid mode: {mode}')
+    if text:
+        stream = TextIOWrapper(stream, encoding, errors, newline)
+    return stream


 def vfspath(path):

@@ -13,7 +13,7 @@
 from abc import ABC, abstractmethod
 from glob import _GlobberBase
 from io import text_encoding
-from pathlib._os import (magic_open, vfspath, ensure_distinct_paths,
+from pathlib._os import (vfsopen, vfspath, ensure_distinct_paths,
                         ensure_different_files, copyfileobj)
 from pathlib import PurePath, Path
 from typing import Optional, Protocol, runtime_checkable
@@ -264,18 +264,18 @@ def info(self):
        raise NotImplementedError

    @abstractmethod
-    def __open_rb__(self, buffering=-1):
+    def __open_reader__(self):
        """
        Open the file pointed to by this path for reading in binary mode and
-        return a file object, like open(mode='rb').
+        return a file object.
        """
        raise NotImplementedError

    def read_bytes(self):
        """
        Open the file in bytes mode, read it, and close the file.
        """
-        with magic_open(self, mode='rb', buffering=0) as f:
+        with vfsopen(self, mode='rb') as f:
            return f.read()

    def read_text(self, encoding=None, errors=None, newline=None):
@@ -285,7 +285,7 @@ def read_text(self, encoding=None, errors=None, newline=None):
        # Call io.text_encoding() here to ensure any warning is raised at an
        # appropriate stack level.
        encoding = text_encoding(encoding)
-        with magic_open(self, mode='r', encoding=encoding, errors=errors, newline=newline) as f:
+        with vfsopen(self, mode='r', encoding=encoding, errors=errors, newline=newline) as f:
            return f.read()

    @abstractmethod
@@ -394,10 +394,10 @@ def mkdir(self):
        raise NotImplementedError

    @abstractmethod
-    def __open_wb__(self, buffering=-1):
+    def __open_writer__(self, mode):
        """
        Open the file pointed to by this path for writing in binary mode and
-        return a file object, like open(mode='wb').
+        return a file object.
        """
        raise NotImplementedError

@@ -407,7 +407,7 @@ def write_bytes(self, data):
        """
        # type-check for the buffer interface before truncating the file
        view = memoryview(data)
-        with magic_open(self, mode='wb') as f:
+        with vfsopen(self, mode='wb') as f:
            return f.write(view)

    def write_text(self, data, encoding=None, errors=None, newline=None):
@@ -420,7 +420,7 @@ def write_text(self, data, encoding=None, errors=None, newline=None):
        if not isinstance(data, str):
            raise TypeError('data must be str, not %s' %
                            data.__class__.__name__)
-        with magic_open(self, mode='w', encoding=encoding, errors=errors, newline=newline) as f:
+        with vfsopen(self, mode='w', encoding=encoding, errors=errors, newline=newline) as f:
            return f.write(data)

    def _copy_from(self, source, follow_symlinks=True):
@@ -439,8 +439,8 @@ def _copy_from(self, source, follow_symlinks=True):
                    stack.append((child, dst.joinpath(child.name)))
            else:
                ensure_different_files(src, dst)
-                with magic_open(src, 'rb') as source_f:
-                    with magic_open(dst, 'wb') as target_f:
+                with vfsopen(src, 'rb') as source_f:
+                    with vfsopen(dst, 'wb') as target_f:
                        copyfileobj(source_f, target_f)



@@ -145,7 +145,7 @@ def __init__(self, *pathsegments):
        super().__init__(*pathsegments)
        self.info = LocalPathInfo(self)

-    def __open_rb__(self, buffering=-1):
+    def __open_reader__(self):
        return open(self, 'rb')

    def iterdir(self):
@@ -163,8 +163,8 @@ class WritableLocalPath(_WritablePath, LexicalPath):
    __slots__ = ()
    __fspath__ = LexicalPath.__vfspath__

-    def __open_wb__(self, buffering=-1):
-        return open(self, 'wb')
+    def __open_writer__(self, mode):
+        return open(self, f'{mode}b')

    def mkdir(self, mode=0o777):
        os.mkdir(self, mode)

@@ -264,13 +264,13 @@ def info(self):
        tree = self.zip_file.filelist.tree
        return tree.resolve(vfspath(self), follow_symlinks=False)

-    def __open_rb__(self, buffering=-1):
+    def __open_reader__(self):
        info = self.info.resolve()
        if not info.exists():
            raise FileNotFoundError(errno.ENOENT, "File not found", self)
        elif info.is_dir():
            raise IsADirectoryError(errno.EISDIR, "Is a directory", self)
-        return self.zip_file.open(info.zip_info, 'r')
+        return self.zip_file.open(info.zip_info)

    def iterdir(self):
        info = self.info.resolve()
@@ -320,8 +320,8 @@ def __repr__(self):
    def with_segments(self, *pathsegments):
        return type(self)(*pathsegments, zip_file=self.zip_file)

-    def __open_wb__(self, buffering=-1):
-        return self.zip_file.open(vfspath(self), 'w')
+    def __open_writer__(self, mode):
+        return self.zip_file.open(vfspath(self), mode)

    def mkdir(self, mode=0o777):
        zinfo = zipfile.ZipInfo(vfspath(self) + '/')

@@ -13,10 +13,10 @@

 if is_pypi:
    from pathlib_abc import PathInfo, _ReadablePath
-    from pathlib_abc._os import magic_open
+    from pathlib_abc._os import vfsopen
 else:
    from pathlib.types import PathInfo, _ReadablePath
-    from pathlib._os import magic_open
+    from pathlib._os import vfsopen


 class ReadTestBase:
@@ -32,7 +32,7 @@ def test_is_readable(self):

    def test_open_r(self):
        p = self.root / 'fileA'
-        with magic_open(p, 'r', encoding='utf-8') as f:
+        with vfsopen(p, 'r', encoding='utf-8') as f:
            self.assertIsInstance(f, io.TextIOBase)
            self.assertEqual(f.read(), 'this is file A\n')

@@ -43,17 +43,17 @@ def test_open_r(self):
    def test_open_r_encoding_warning(self):
        p = self.root / 'fileA'
        with self.assertWarns(EncodingWarning) as wc:
-            with magic_open(p, 'r'):
+            with vfsopen(p, 'r'):
                pass
        self.assertEqual(wc.filename, __file__)

    def test_open_rb(self):
        p = self.root / 'fileA'
-        with magic_open(p, 'rb') as f:
+        with vfsopen(p, 'rb') as f:
            self.assertEqual(f.read(), b'this is file A\n')
-        self.assertRaises(ValueError, magic_open, p, 'rb', encoding='utf8')
-        self.assertRaises(ValueError, magic_open, p, 'rb', errors='strict')
-        self.assertRaises(ValueError, magic_open, p, 'rb', newline='')
+        self.assertRaises(ValueError, vfsopen, p, 'rb', encoding='utf8')
+        self.assertRaises(ValueError, vfsopen, p, 'rb', errors='strict')
+        self.assertRaises(ValueError, vfsopen, p, 'rb', newline='')

    def test_read_bytes(self):
        p = self.root / 'fileA'

@@ -13,10 +13,10 @@

 if is_pypi:
    from pathlib_abc import _WritablePath
-    from pathlib_abc._os import magic_open
+    from pathlib_abc._os import vfsopen
 else:
    from pathlib.types import _WritablePath
-    from pathlib._os import magic_open
+    from pathlib._os import vfsopen


 class WriteTestBase:
@@ -31,7 +31,7 @@ def test_is_writable(self):

    def test_open_w(self):
        p = self.root / 'fileA'
-        with magic_open(p, 'w', encoding='utf-8') as f:
+        with vfsopen(p, 'w', encoding='utf-8') as f:
            self.assertIsInstance(f, io.TextIOBase)
            f.write('this is file A\n')
        self.assertEqual(self.ground.readtext(p), 'this is file A\n')
@@ -43,19 +43,19 @@ def test_open_w(self):
    def test_open_w_encoding_warning(self):
        p = self.root / 'fileA'
        with self.assertWarns(EncodingWarning) as wc:
-            with magic_open(p, 'w'):
+            with vfsopen(p, 'w'):
                pass
        self.assertEqual(wc.filename, __file__)

    def test_open_wb(self):
        p = self.root / 'fileA'
-        with magic_open(p, 'wb') as f:
+        with vfsopen(p, 'wb') as f:
            #self.assertIsInstance(f, io.BufferedWriter)
            f.write(b'this is file A\n')
        self.assertEqual(self.ground.readbytes(p), b'this is file A\n')
-        self.assertRaises(ValueError, magic_open, p, 'wb', encoding='utf8')
-        self.assertRaises(ValueError, magic_open, p, 'wb', errors='strict')
-        self.assertRaises(ValueError, magic_open, p, 'wb', newline='')
+        self.assertRaises(ValueError, vfsopen, p, 'wb', encoding='utf8')
+        self.assertRaises(ValueError, vfsopen, p, 'wb', errors='strict')
+        self.assertRaises(ValueError, vfsopen, p, 'wb', newline='')

    def test_write_bytes(self):
        p = self.root / 'fileA'