"""
This provides abstractions around a number of different file and
stream types available to Python so that they are always used in the
most efficient way.
The classes in this module should not be instantiated directly, but
instead, one should use the factory function `get_file`.
"""
from __future__ import annotations
import io
import mmap
import os
import pathlib
import re
import sys
import typing
import warnings
from os import SEEK_CUR, SEEK_END, SEEK_SET
from typing import TYPE_CHECKING, overload
from urllib.request import url2pathname
import numpy as np
from typing_extensions import Reader, Writer, deprecated
from ._extern import atomicfile
from .exceptions import DelimiterNotFoundError
from .util import _patched_urllib_parse
if TYPE_CHECKING:
from collections.abc import Iterator
from fsspec.core import OpenFile
from asdf.typing import ByteArray1D, FileLike, FileMode, PathLike, Reader
__all__ = ["GenericFile", "get_file", "get_uri", "relative_uri", "resolve_uri"]
_FILE_PERMISSIONS_DEFAULT_UMASK = 0o22
_FILE_PERMISSIONS_ALL = 0o777
_FILE_PERMISSIONS_NO_EXECUTE = 0o666
_local_file_schemes = ["", "file"]
if sys.platform.startswith("win"): # pragma: no cover
import string
_local_file_schemes.extend(string.ascii_letters)
def _check_bytes(fd, mode):
"""
Checks whether a given file-like object is opened in binary mode.
"""
# On Python 3, doing fd.read(0) on an HTTPResponse object causes
# it to not be able to read any further, so we do this different
# kind of check, which, unfortunately, is not as robust.
if isinstance(fd, io.IOBase):
if isinstance(fd, io.TextIOBase):
return False
return True
if "r" in mode:
x = fd.read(0)
if not isinstance(x, bytes):
return False
elif "w" in mode:
try:
fd.write(b"")
except TypeError:
return False
return True
[docs]
def resolve_uri(base: str | None, uri: str) -> str:
"""
Resolve a URI against a base URI.
"""
if base is None:
base = ""
resolved = _patched_urllib_parse.urljoin(base, uri)
parsed = _patched_urllib_parse.urlparse(resolved)
if parsed.path != "" and not parsed.path.startswith("/"):
msg = "Resolved to relative URL"
raise ValueError(msg)
return resolved
[docs]
def relative_uri(source: str, target: str) -> str:
"""
Make a relative URI from source to target.
"""
su = _patched_urllib_parse.urlparse(source)
tu = _patched_urllib_parse.urlparse(target)
extra = list(tu[3:])
relative = None
if tu[0] == "" and tu[1] == "":
if tu[2] == su[2]:
relative = ""
elif not tu[2].startswith("/"):
relative = tu[2]
elif su[0:2] != tu[0:2]:
return target
if relative is None:
relative = "" if tu[2] == su[2] else os.path.relpath(tu[2], os.path.dirname(su[2]))
if relative == ".":
relative = ""
return _patched_urllib_parse.urlunparse(["", "", relative, *extra])
class _TruncatedReader:
"""
Reads until a given delimiter is found. Only works with
RandomAccessFile and InputStream, though as this is a private
class, this is not explicitly enforced.
"""
def __init__(
self,
fd,
delimiter,
readahead_bytes,
delimiter_name=None,
include=False,
initial_content=b"",
exception=True,
):
self._fd = fd
self._delimiter = delimiter
self._readahead_bytes = readahead_bytes
if delimiter_name is None:
delimiter_name = delimiter
self._delimiter_name = delimiter_name
self._include = include
self._initial_content = initial_content
self._trailing_content = b""
self._exception = exception
self._past_end = False
def read(self, nbytes=None):
if self._past_end:
content = self._trailing_content[:nbytes]
if nbytes is None:
self._trailing_content = b""
else:
self._trailing_content = self._trailing_content[nbytes:]
return content
if nbytes is None:
content = self._fd.peek()
elif nbytes > len(self._initial_content):
content = self._fd.peek(nbytes - len(self._initial_content) + self._readahead_bytes)
else:
content = self._initial_content[:nbytes]
self._initial_content = self._initial_content[nbytes:]
return content
if content == b"":
if self._exception:
msg = f"{self._delimiter_name} not found"
raise DelimiterNotFoundError(msg)
self._past_end = True
return content
index = re.search(self._delimiter, content)
if index is not None:
index = index.end() if self._include else index.start()
content = content[:index]
self._past_end = True
elif not (nbytes is None and self._exception):
if nbytes:
content = content[: nbytes - len(self._initial_content)]
else:
# Read the whole file and didn't find the delimiter
msg = f"{self._delimiter_name} not found"
raise DelimiterNotFoundError(msg)
self._fd.fast_forward(len(content))
if self._initial_content:
content = self._initial_content + content
self._initial_content = b""
if self._past_end and nbytes:
self._trailing_content = content[nbytes:]
content = content[:nbytes]
return content
[docs]
class GenericFile:
"""
Base class for an abstraction layer around a number of different
file-like types. Each of its subclasses handles a particular kind
of file in the most efficient way possible.
This class should not be instantiated directly, but instead the
factory function `get_file` should be used to get the correct
subclass for the given file-like object.
"""
def __init__(self, fd, mode: FileMode, close: bool = False, uri: str | None = None):
"""
Parameters
----------
fd : file-like object
The particular kind of file-like object must match the
subclass of `GenericFile` being instantiated.
mode : str
Must be ``"r"`` (read), ``"w"`` (write), or ``"rw"``
(read/write).
close : bool, optional
When ``True``, close the given ``fd`` in the ``__exit__``
method, i.e. at the end of the with block. Should be set
to ``True`` when this object "owns" the file object.
Default: ``False``.
uri : str, optional
The file path or URI used to open the file. This is used
to resolve relative URIs when the file refers to external
sources.
"""
if not _check_bytes(fd, mode):
msg = "File-like object must be opened in binary mode."
raise ValueError(msg)
# can't import at the top level due to circular import
from .config import get_config
self._asdf_get_config = get_config
self._fd = fd
self._mode = mode
self._close = close
self._uri = uri
self.block_size = get_config().io_block_size
def __enter__(self) -> GenericFile:
return self
def __exit__(self, type_, value, traceback) -> None:
if self._close:
if hasattr(self._fd, "__exit__"):
self._fd.__exit__(type_, value, traceback)
else:
self._fd.close()
@property
def block_size(self):
return self._blksize
@block_size.setter
def block_size(self, block_size):
if block_size == -1:
try:
block_size = os.fstat(self._fd.fileno()).st_blksize
except Exception:
block_size = io.DEFAULT_BUFFER_SIZE
if block_size <= 0:
msg = f"block_size ({block_size}) must be > 0"
raise ValueError(msg)
self._blksize = block_size
@property
def mode(self) -> FileMode:
"""
The mode of the file. Will be ``'r'``, ``'w'`` or ``'rw'``.
"""
return self._mode
@property
def uri(self) -> str | None:
"""
The base uri of the file.
"""
return self._uri
[docs]
def read(self, size: int = -1) -> bytes:
"""
Read at most size bytes from the file (less if the read hits
EOF before obtaining size bytes). If the size argument is
negative or omitted, read all data until EOF is reached. The
bytes are returned as a `bytes` object. An empty `bytes`
object is returned when EOF is encountered immediately.
Only available if `readable` returns `True`.
"""
# On Python 3, reading 0 bytes from a socket causes it to stop
# working, so avoid doing that at all costs.
if size == 0:
return b""
return self._fd.read(size)
[docs]
def read_block(self) -> bytes:
"""
Read a "block" from the file. For real filesystem files, the
block is the size of a native filesystem block.
"""
return self.read(self._blksize)
[docs]
def read_blocks(self, size: int) -> Iterator[bytes]:
"""
Read ``size`` bytes of data from the file, one block at a
time. The result is a generator where each value is a bytes
object.
"""
for i in range(0, size, self._blksize):
thissize = min(self._blksize, size - i)
yield self.read(thissize)
[docs]
def write(self, content: bytes | bytearray | memoryview) -> None:
"""
Write a string to the file. There is no return value. Due to
buffering, the string may not actually show up in the file
until the flush() or close() method is called.
Only available if `writable` returns `True`.
"""
self._fd.write(content)
[docs]
def write_array(self, array: ByteArray1D) -> None:
"""
Write array content to the file. Array must be 1D contiguous
so that this method can avoid making assumptions about the
intended memory layout. Endianness is preserved.
Parameters
----------
array : np.ndarray
Must be 1D contiguous.
"""
if len(array.shape) != 1 or not array.flags.contiguous:
msg = "Requires 1D contiguous array."
raise ValueError(msg)
self.write(array.data)
[docs]
def peek(self, size: int = -1) -> bytes:
"""
Read bytes of the file without consuming them. This method
must be implemented by all GenericFile implementations that
provide ASDF input (those that aren't seekable should use a
buffer to store peeked bytes).
Parameters
----------
size : int
Number of bytes to peek, or -1 to peek all remaining bytes.
"""
if self.seekable():
cursor = self.tell()
content = self.read(size)
self.seek(cursor, SEEK_SET)
return content
msg = "Non-seekable file"
raise RuntimeError(msg)
[docs]
def seek(self, offset: int, whence: int = 0) -> int:
"""
Set the file's current position. Only available if `seekable`
returns `True`.
Parameters
----------
offset : integer
Offset, in bytes.
whence : integer, optional
The ``whence`` argument is optional and defaults to
SEEK_SET or 0 (absolute file positioning); other values
are SEEK_CUR or 1 (seek relative to the current
position) and SEEK_END or 2 (seek relative to the
file`s end).
"""
result = self._fd.seek(offset, whence)
return result
[docs]
def tell(self) -> int:
"""
Return the file's current position, in bytes. Only available
in `seekable` returns `True`.
"""
return self._fd.tell()
[docs]
def flush(self) -> None:
"""
Flush the internal buffer.
"""
self._fd.flush()
[docs]
def close(self) -> None:
"""
Close the file. The underlying file-object will only be
closed if ``close=True`` was passed to the constructor.
"""
if self._close:
self._fd.close()
if hasattr(self, "_fix_permissions"):
self._fix_permissions()
[docs]
def truncate(self, size: int | None = None) -> None:
"""
Truncate the file to the given size.
"""
raise NotImplementedError
[docs]
def writable(self) -> bool:
"""
Returns `True` if the file can be written to.
"""
return "w" in self.mode
[docs]
def readable(self) -> bool:
"""
Returns `True` if the file can be read from.
"""
return "r" in self.mode
[docs]
def seekable(self) -> bool:
"""
Returns `True` if the file supports random access (`seek` and
`tell`).
"""
return False
[docs]
def can_memmap(self) -> bool:
"""
Returns `True` if the file supports memmapping.
"""
return False
[docs]
def is_closed(self) -> bool:
"""
Returns `True` if the underlying file object is closed.
"""
return self._fd.closed
[docs]
def read_until(
self,
delimiter: str | bytes,
readahead_bytes: int,
delimiter_name: str | None = None,
include: bool = True,
initial_content: bytes = b"",
exception: bool = True,
) -> bytes:
"""
Reads until a match for a given regular expression is found.
Parameters
----------
delimiter : str
A regular expression.
readahead_bytes : int
The number of bytes to read ahead to make sure the
delimiter isn't on a block boundary.
delimiter_name : str, optional
The name of the delimiter. Used in error messages if the
delimiter is not found. If not provided, the raw content
of ``delimiter`` will be used.
include : bool, optional
When ``True``, include the delimiter in the result.
initial_content : bytes, optional
Additional content to include at the beginning of the
first read.
exception : bool, optional
If ``True`` (default), raise an exception if the end
marker isn't found.
Returns
-------
content : bytes
The content from the current position in the file, up to
the delimiter. Includes the delimiter if ``include`` is
``True``.
Raises
------
DelimiterNotFoundError :
If the delimiter is not found before the end of the file.
"""
buff = io.BytesIO()
reader = self.reader_until(
delimiter,
readahead_bytes,
delimiter_name=delimiter_name,
include=include,
initial_content=initial_content,
exception=exception,
)
while True:
content = reader.read(self.block_size)
buff.write(content)
if len(content) < self.block_size:
break
return buff.getvalue()
[docs]
def reader_until(
self,
delimiter: str | bytes,
readahead_bytes: int,
delimiter_name: str | None = None,
include: bool = True,
initial_content: bytes = b"",
exception: bool = True,
) -> Reader:
"""
Returns a readable file-like object that treats the given
delimiter as the end-of-file.
Parameters
----------
delimiter : str
A regular expression.
readahead_bytes : int
The number of bytes to read ahead to make sure the
delimiter isn't on a block boundary.
delimiter_name : str, optional
The name of the delimiter. Used in error messages if the
delimiter is not found. If not provided, the raw content
of ``delimiter`` will be used.
include : bool, optional
When ``True``, include the delimiter in the result.
initial_content : bytes, optional
Additional content to include at the beginning of the
first read.
exception : bool, optional
If ``True`` (default), raise an exception if the end
marker isn't found.
Raises
------
DelimiterNotFoundError :
If the delimiter is not found before the end of the file.
"""
raise NotImplementedError
[docs]
def seek_until(
self,
delimiter: str | bytes,
readahead_bytes: int,
delimiter_name: str | None = None,
include: bool = True,
initial_content: bytes = b"",
exception: bool = True,
) -> bool:
"""
Seeks in the file until a match for a given regular expression
is found. This is similar to ``read_until``, except the
intervening content is not retained.
Parameters
----------
delimiter : str
A regular expression.
readahead_bytes : int
The number of bytes to read ahead to make sure the
delimiter isn't on a block boundary.
delimiter_name : str, optional
The name of the delimiter. Used in error messages if the
delimiter is not found. If not provided, the raw content
of ``delimiter`` will be used.
include : bool, optional
When ``True``, include the delimiter in the result.
initial_content : bytes, optional
Additional content to include at the beginning of the
first read.
exception : bool, optional
If ``True`` (default), raise an exception if the end
marker isn't found.
Returns
-------
bool
``True`` if the delimiter was found.
Raises
------
DelimiterNotFoundError :
If ``exception`` is enabled and the delimiter is not found
before the end of the file.
"""
reader = self.reader_until(
delimiter,
readahead_bytes,
delimiter_name=delimiter_name,
include=include,
initial_content=initial_content,
exception=True,
)
try:
while reader.read(self.block_size) != b"":
pass
except DelimiterNotFoundError:
if exception:
raise
return False
return True
[docs]
def fast_forward(self, size: int) -> None:
"""
Move the file position forward by ``size``.
"""
raise NotImplementedError
[docs]
def clear(self, nbytes: int) -> None:
"""
Write nbytes of zeros.
"""
blank_data = b"\0" * self.block_size
for i in range(0, nbytes, self.block_size):
length = min(nbytes - i, self.block_size)
self.write(blank_data[:length])
[docs]
def memmap_array(self, offset: int, size: int) -> np.memmap:
"""
Memmap a chunk of the file into a ``np.memmap`` object.
Parameters
----------
offset : integer
The offset, in bytes, in the file.
size : integer
The size of the data to memmap.
Returns
-------
array : np.memmap
"""
msg = f"memmapping is not implemented for {self.__class__.__name__}"
raise NotImplementedError(msg)
[docs]
def close_memmap(self) -> None:
"""
Close the memmapped file (if one was mapped with memmap_array)
"""
msg = f"memmapping is not implemented for {self.__class__.__name__}"
raise NotImplementedError(msg)
[docs]
def flush_memmap(self) -> None:
"""
Flush any pending writes to the memmapped file (if one was mapped with
memmap_array)
"""
msg = f"memmapping is not implemented for {self.__class__.__name__}"
raise NotImplementedError(msg)
[docs]
def read_into_array(self, size: int) -> ByteArray1D:
"""
Read a chunk of the file into a uint8 array.
Parameters
----------
size : integer
The size of the data.
Returns
-------
array : np.memmap
"""
buff = self.read(size)
# Need cast because numpy doesn't type frombuffer as a 1D array
return typing.cast("ByteArray1D", np.frombuffer(buff, np.uint8, size, 0))
class GenericWrapper:
"""
A wrapper around a `GenericFile` object so that closing only
happens in the very outer layer.
"""
def __init__(self, fd):
self._fd = fd
def __enter__(self):
return self
def __exit__(self, type_, value, traceback):
pass
def __getattr__(self, attr):
return getattr(self._fd, attr)
class RandomAccessFile(GenericFile):
"""
The base class of file types that support random access.
"""
def __init__(self, fd, mode: FileMode, close: bool = False, uri: str | None = None, inner_fd=None):
super().__init__(fd, mode, close, uri)
self._inner_fd = inner_fd
def seekable(self) -> bool:
return True
def reader_until(
self,
delimiter,
readahead_bytes,
delimiter_name=None,
include=True,
initial_content=b"",
exception=True,
):
return _TruncatedReader(
self,
delimiter,
readahead_bytes,
delimiter_name=delimiter_name,
include=include,
initial_content=initial_content,
exception=exception,
)
def fast_forward(self, size: int) -> None:
if size < 0:
self.seek(0, SEEK_END)
self.seek(size, SEEK_CUR)
def truncate(self, size: int | None = None) -> None:
if size is None:
self._fd.truncate()
else:
self._fd.truncate(size)
self.seek(size, SEEK_SET)
class RealFile(RandomAccessFile):
"""
Handles "real" files on a filesystem.
"""
def __init__(self, fd, mode: FileMode, close: bool = False, uri: str | None = None, inner_fd=None):
super().__init__(fd, mode, close=close, uri=uri, inner_fd=inner_fd)
if uri is None and hasattr(fd, "name") and isinstance(fd.name, str):
self._uri = pathlib.Path(fd.name).expanduser().absolute().as_uri()
def write_array(self, arr):
if isinstance(arr, np.memmap) and getattr(arr, "fd", None) is self:
arr.flush()
self.fast_forward(len(arr.data))
else:
if len(arr.shape) != 1 or not arr.flags.contiguous:
msg = "Requires 1D contiguous array."
raise ValueError(msg)
self._fd.write(arr.data)
def can_memmap(self):
return True
def memmap_array(self, offset, size):
if not hasattr(self, "_mmap"):
loc = self._fd.tell()
acc = mmap.ACCESS_WRITE if "w" in self._mode else mmap.ACCESS_READ
self._fd.seek(0, 2)
nbytes = self._fd.tell()
self._mmap = mmap.mmap(self._fd.fileno(), nbytes, access=acc)
# on windows mmap seeks to the start of the file so return the file
# pointer to this previous location
self._fd.seek(loc, 0)
return np.ndarray.__new__(np.memmap, shape=size, offset=offset, dtype="uint8", buffer=self._mmap)
def close_memmap(self):
if hasattr(self, "_mmap"):
# we no longer close the _mmap here. This does mean that views of arrays
# that are backed by _mmap will keep the _mmap alive (and open). This is
# the cost of avoiding segfaults as np.memmap does not check if mmap is
# closed.
del self._mmap
def flush_memmap(self):
if hasattr(self, "_mmap"):
self._mmap.flush()
def read_into_array(self, size):
return np.fromfile(self._fd, dtype=np.uint8, count=size)
def _fix_permissions(self):
"""
atomicfile internally uses tempfile.NamedTemporaryFile
which uses tempfile.mkstemp which makes a file that is
"readable and writable only by the creating user ID."
this creates files with mode 0o600 regardless of umask
Rather than modify atomicfile, this will use
the umask to determine the file permissions and modify
the resulting file permission bits
"""
if isinstance(self._fd, atomicfile._AtomicWFile):
fn = self._fd._filename
if not os.path.exists(fn):
return
# there is no way to read the umask without setting it
# so set it to the typical default 0o22
umask = os.umask(_FILE_PERMISSIONS_DEFAULT_UMASK)
if umask != _FILE_PERMISSIONS_DEFAULT_UMASK:
# restore the read value if it differs from the default
os.umask(umask)
permissions = _FILE_PERMISSIONS_ALL if os.path.isdir(fn) else _FILE_PERMISSIONS_NO_EXECUTE
os.chmod(self._fd._filename, permissions & ~umask)
def __exit__(self, type_, value, traceback):
super().__exit__(type_, value, traceback)
if self._close:
self._fix_permissions()
def close(self):
self.flush_memmap()
super().close()
self.close_memmap()
if self._close:
self._fix_permissions()
def truncate(self, size=None):
# windows supports truncating as long as the file not opened
# more than once. So this must be called after closing all
# memmaps
if sys.platform.startswith("win") and hasattr(self, "_mmap"):
self._mmap.close()
self.close_memmap()
super().truncate(size=size)
class MemoryIO(RandomAccessFile):
"""
Handles random-access memory buffers, mainly `io.BytesIO` and
`StringIO.StringIO`.
"""
def __init__(self, fd, mode: FileMode, uri: str | None = None, inner_fd=None):
super().__init__(fd, mode, uri=uri, inner_fd=inner_fd)
def read_into_array(self, size):
buf = self._fd.getvalue()
offset = self._fd.tell()
result = np.frombuffer(buf, np.uint8, size, offset)
# Copy the buffer so the original memory can be released.
result = result.copy()
self.seek(size, SEEK_CUR)
return result
class InputStream(GenericFile):
"""
Handles an input stream, such as stdin.
"""
def __init__(self, fd, mode: FileMode = "r", close: bool = False, uri: str | None = None):
super().__init__(fd, mode, close=close, uri=uri)
self._fd = fd
self._buffer = b""
def peek(self, size=-1):
if size < 0:
self._buffer += self._fd.read()
else:
len_buffer = len(self._buffer)
if len_buffer < size:
self._buffer += self._fd.read(size - len_buffer)
return self._buffer
def read(self, size=-1):
# On Python 3, reading 0 bytes from a socket causes it to stop
# working, so avoid doing that at all costs.
if size == 0:
return b""
len_buffer = len(self._buffer)
if len_buffer == 0:
return self._fd.read(size)
if size < 0:
self._buffer += self._fd.read()
buffer = self._buffer
self._buffer = b""
return buffer
if len_buffer < size:
if len_buffer < size:
self._buffer += self._fd.read(size - len(self._buffer))
buffer = self._buffer
self._buffer = b""
return buffer
buffer = self._buffer[:size]
self._buffer = self._buffer[size:]
return buffer
def reader_until(
self,
delimiter,
readahead_bytes,
delimiter_name=None,
include=True,
initial_content=b"",
exception=True,
):
return _TruncatedReader(
self,
delimiter,
readahead_bytes,
delimiter_name=delimiter_name,
include=include,
initial_content=initial_content,
exception=exception,
)
def fast_forward(self, size):
if size >= 0 and len(self.read(size)) != size:
msg = "Read past end of file"
raise OSError(msg)
def read_into_array(self, size):
try:
# See if Numpy can handle this as a real file first...
return np.fromfile(self._fd, np.uint8, size)
except (OSError, AttributeError):
# Else, fall back to reading into memory and then
# returning the Numpy array.
data = self.read(size)
# We need to copy the array, so it is writable
result = np.frombuffer(data, np.uint8, size)
# When creating an array from a buffer, it is read-only.
# If we need a read/write array, we have to copy it.
if "w" in self._mode:
result = result.copy()
return result
class OutputStream(GenericFile):
"""
Handles an output stream, such as stdout.
"""
def __init__(self, fd, close: bool = False, uri: str | None = None):
super().__init__(fd, "w", close=close, uri=uri)
self._fd = fd
def fast_forward(self, size):
if size < 0:
return
self.clear(size)
[docs]
def get_uri(file_obj):
"""
Returns the uri of the given file object
Parameters
----------
uri : object
"""
if isinstance(file_obj, str):
return file_obj
if isinstance(file_obj, GenericFile):
return file_obj.uri
# A catch-all for types from Python's io module that have names
return getattr(file_obj, "name", "")
@overload
def get_file(
init: PathLike | io.IOBase | GenericFile,
mode: FileMode = ...,
uri: str | None = ...,
close: bool = ...,
) -> GenericFile: ...
@overload
@deprecated("Duck-typed file objects are deprecated. Use an instance of IOBase instead.")
def get_file(
init: Reader | Writer,
mode: FileMode = ...,
uri: str | None = ...,
close: bool = ...,
) -> GenericFile: ...
[docs]
def get_file(
init: FileLike,
mode: FileMode = "r",
uri: str | None = None,
close: bool = False,
) -> GenericFile:
"""
Returns a `GenericFile` instance suitable for wrapping the given
object ``init``.
If passed an already open file-like object, it must be opened for
reading/writing in binary mode. It is the caller's responsibility
to close it.
Parameters
----------
init : object
``init`` may be:
- A ``bytes`` or ``unicode`` file path or ``file:`` or ``http:``
url.
- A Python 2 ``file`` object.
- An `io.IOBase` object (the default file object on Python 3).
- A ducktyped object that looks like a file object. If ``mode``
is ``"r"``, it must have a ``read`` method. If ``mode`` is
``"w"``, it must have a ``write`` method. If ``mode`` is
``"rw"`` it must have the ``read``, ``write``, ``tell`` and
``seek`` methods.
- A `GenericFile` instance, in which case it is wrapped in a
``GenericWrapper`` instance, so that the file is closed when
only when the final layer is unwrapped.
mode : str
Must be one of ``"r"``, ``"w"`` or ``"rw"``.
uri : str
Sets the base URI of the file object. This will be used to
resolve any relative URIs contained in the file. This is
redundant if ``init`` is a ``bytes`` or ``unicode`` object (since it
will be the uri), and it may be determined automatically if
``init`` refers to a regular filesystem file. It is not required
if URI resolution is not used in the file.
close : bool
If ``True``, closes the underlying file handle when this object is
closed. Defaults to ``False``.
Returns
-------
fd : GenericFile
Raises
------
ValueError, TypeError, IOError
"""
if mode not in ("r", "w", "rw"):
msg = "mode must be 'r', 'w' or 'rw'"
raise ValueError(msg)
if isinstance(init, io.TextIOWrapper) and init in (sys.__stdout__, sys.__stdin__, sys.__stderr__):
init = os.fdopen(init.fileno(), init.mode + "b")
if isinstance(init, (GenericFile, GenericWrapper)):
if mode not in init.mode:
msg = f"File is opened as '{init.mode}', but '{mode}' was requested"
raise ValueError(msg)
# From a user perspective GenericWrapper is a GenericFile
# But we can't make GenericWrapper a subclass of GenericFile because there are places internally
# that depend on it not being one
return typing.cast("GenericFile", GenericWrapper(init))
if isinstance(init, (str, pathlib.Path)):
parsed = _patched_urllib_parse.urlparse(str(init))
realmode = "r+b" if mode == "rw" else mode + "b"
if parsed.scheme in _local_file_schemes:
# if paths have an extra leading '/' urlparse will
# parse them even though they violate rfc8089. This will
# lead to errors or writing files to unexpected locations
# on non-windows systems
if not sys.platform.startswith("win") and parsed.scheme == "" and parsed.netloc != "":
msg = f"Invalid path {init}"
raise ValueError(msg)
# Windows paths are not URIs, and so they should not be parsed as
# such. Otherwise, the drive component of the path can get lost.
# This is not an ideal solution, but we can't use pathlib here
# because it doesn't handle URIs properly.
realpath = (
str(init)
if sys.platform.startswith("win") and parsed.scheme in string.ascii_letters
else url2pathname(parsed.path)
)
try:
fd = atomicfile.atomic_open(realpath, realmode) if mode == "w" else open(realpath, realmode)
fd = fd.__enter__()
except FileNotFoundError as e:
# atomic_open will create an Exception with an odd looking path
# overwrite the error message to make it more informative
e.filename = realpath
raise e
return RealFile(fd, mode, close=True, uri=uri)
# this is not a local file, import fsspec (if available)
try:
import fsspec
except ImportError:
fsspec = None
if fsspec:
# Cast because fsspec doesn't have type hints
fd = typing.cast("OpenFile", fsspec.open(init, realmode))
try:
fd = fd.__enter__()
except NotImplementedError as err:
msg = f"Unable to open {init} with mode {mode}"
raise ValueError(msg) from err
return get_file(fd, uri=str(uri or init), close=True)
if isinstance(init, io.BytesIO):
return MemoryIO(init, mode, uri=uri)
if isinstance(init, io.StringIO):
msg = "io.StringIO objects are not supported. Use io.BytesIO instead."
raise TypeError(msg)
if not isinstance(init, io.IOBase):
# Only generate a warning if its an actual duck-typed file.
# If its just a random object it will raise a ValueError below anyway.
if hasattr(init, "read") or hasattr(init, "write"):
warnings.warn(
"Duck-typed file objects are deprecated. Use an instance of IOBase instead.",
DeprecationWarning,
)
elif not init.seekable() and mode != "w":
warnings.warn(
(
"Reading from non-seekable files is deprecated. "
"Consider implementing seek() or reading the file into a BytesIO instance."
),
DeprecationWarning,
)
if isinstance(init, io.IOBase):
if ("r" in mode and not init.readable()) or ("w" in mode and not init.writable()):
msg = f"File is opened as '{init.mode}', but '{mode}' was requested"
raise ValueError(msg)
if init.seekable():
init2 = init.raw if hasattr(init, "raw") else init
if hasattr(init2, "getvalue"):
return MemoryIO(init2, mode, uri=uri, inner_fd=init)
# can we call 'fileno'? if not, we can't memmap so don't
# make a RealFile
try:
init2.fileno()
return RealFile(init2, mode, uri=uri, close=close, inner_fd=init)
except io.UnsupportedOperation:
return RandomAccessFile(init2, mode, uri=uri, close=close, inner_fd=init)
if mode == "w":
return OutputStream(init, uri=uri, close=close)
if mode == "r":
return InputStream(init, mode, uri=uri, close=close)
msg = f"File '{init}' could not be opened in 'rw' mode"
raise ValueError(msg)
if mode == "w" and (hasattr(init, "write") and hasattr(init, "seek") and hasattr(init, "tell")):
return MemoryIO(init, mode, uri=uri)
if mode == "r" and (hasattr(init, "read") and hasattr(init, "seek") and hasattr(init, "tell")):
return MemoryIO(init, mode, uri=uri)
if mode == "rw" and (
hasattr(init, "read") and hasattr(init, "write") and hasattr(init, "seek") and hasattr(init, "tell")
):
return MemoryIO(init, mode, uri=uri)
if mode == "w" and hasattr(init, "write"):
return OutputStream(init, uri=uri, close=close)
if mode == "r" and hasattr(init, "read"):
return InputStream(init, mode, uri=uri, close=close)
msg = f"Can't handle '{init}' as a file for mode '{mode}'"
raise ValueError(msg)