Class PickleableFileManager

xarray/backends/file_manager.py:344–442

File manager that supports pickling by reopening a file object. Use PickleableFileManager for wrapping file-like objects that do not natively support pickling (e.g., netCDF4.Dataset and h5netcdf.File) in cases where a global cache is not desirable (e.g., for netCDF files opened from byt

class PickleableFileManager(FileManager[T_File]):
    """File manager that supports pickling by reopening a file object.

    Use PickleableFileManager for wrapping file-like objects that do not natively
    support pickling (e.g., netCDF4.Dataset and h5netcdf.File) in cases where a
    global cache is not desirable (e.g., for netCDF files opened from bytes in
    memory, or from existing file objects).
    """

    def __init__(
        self,
        opener: Callable[..., T_File],
        *args: Any,
        mode: Any = _OMIT_MODE,
        lock: Lock | None | Literal[False] = None,
        kwargs: Mapping[str, Any] | None = None,
    ):
        kwargs = {} if kwargs is None else dict(kwargs)
        self._opener = opener
        self._args = args
        self._mode = "a" if mode == "w" else mode
        self._kwargs = kwargs
        self._lock = lock

        # Note: No need for locking with PickleableFileManager, because all
        # opening of files happens in the constructor.
        if mode != _OMIT_MODE:
            kwargs = kwargs | {"mode": mode}
        self._file: T_File | None = opener(*args, **kwargs)

    @property
    def _closed(self) -> bool:
        # If opener() raised an error in the constructor, _file may not be set
        return getattr(self, "_file", None) is None

    def _get_unclosed_file(self) -> T_File:
        if self._closed:
            raise RuntimeError("file is closed")
        file = self._file
        assert file is not None
        return file

    def acquire(self, needs_lock: bool = True) -> T_File:
        del needs_lock  # unused
        return self._get_unclosed_file()

    @contextmanager
    def acquire_context(self, needs_lock: bool = True) -> Iterator[T_File]:
        del needs_lock  # unused
        yield self._get_unclosed_file()

    def close(self, needs_lock: bool = True) -> None:
        if not self._closed:
            file = self._get_unclosed_file()
            if needs_lock and self._lock:
                with self._lock:
                    file.close()
            else: