import logging
import re
import tarfile
import fsspec
from fsspec.archive import AbstractArchiveFileSystem
from fsspec.compression import compr
from fsspec.utils import infer_compression
typemap = {b"0": "file", b"5": "directory"}
logger = logging.getLogger("tar")
[docs]
class TarFileSystem(AbstractArchiveFileSystem):
"""Compressed Tar archives as a file-system (read-only)
Supports the following formats:
tar.gz, tar.bz2, tar.xz
"""
root_marker = ""
protocol = "tar"
cachable = False
@classmethod
def _strip_protocol(cls, path):
# file paths are always relative to the archive root
return super()._strip_protocol(path).lstrip("/")
[docs]
def __init__(
self,
fo="",
index_store=None,
target_options=None,
target_protocol=None,
compression=None,
**kwargs,
):
super().__init__(**kwargs)
target_options = target_options or {}
if isinstance(fo, str):
self.of = fsspec.open(fo, protocol=target_protocol, **target_options)
fo = self.of.open() # keep the reference
# Try to infer compression.
if compression is None:
name = None
# Try different ways to get hold of the filename. `fo` might either
# be a `fsspec.LocalFileOpener`, an `io.BufferedReader` or an
# `fsspec.AbstractFileSystem` instance.
try:
# Amended io.BufferedReader or similar.
# This uses a "protocol extension" where original filenames are
# propagated to archive-like filesystems in order to let them
# infer the right compression appropriately.
if hasattr(fo, "original"):
name = fo.original
# fsspec.LocalFileOpener
elif hasattr(fo, "path"):
name = fo.path
# io.BufferedReader
elif hasattr(fo, "name"):
name = fo.name
# fsspec.AbstractFileSystem
elif hasattr(fo, "info"):
name = fo.info()["name"]
except Exception as ex:
logger.warning(
f"Unable to determine file name, not inferring compression: {ex}"
)
if name is not None:
compression = infer_compression(name)
logger.info(f"Inferred compression {compression} from file name {name}")
if compression is not None:
# TODO: tarfile already implements compression with modes like "'r:gz'",
# but then would seek to offset in the file work?
fo = compr[compression](fo)
self._fo_ref = fo
self.fo = fo # the whole instance is a context
self.tar = tarfile.TarFile(fileobj=self.fo)
self.dir_cache = None
self.index_store = index_store
self.index = None
self._index()
def _index(self):
# TODO: load and set saved index, if exists
out = {}
for ti in self.tar:
info = ti.get_info()
info["type"] = typemap.get(info["type"], "file")
orig_name = info["name"].rstrip("/")
# Collapse duplicate slashes in the filesystem-facing name.
name = re.sub("/+", "/", orig_name)
info["name"] = name
if ti.islnk() or ti.issym():
# extractfile follows links, so report the size of the target
info["size"] = self._link_target_size(ti)
out[name] = (info, ti.offset_data, orig_name)
self.index = out
# TODO: save index to self.index_store here, if set
def _link_target_size(self, ti):
seen = set()
while ti.islnk() or ti.issym():
if ti.name in seen:
return 0
seen.add(ti.name)
try:
ti = self.tar._find_link_target(ti)
except KeyError:
# dangling link; opening it fails as well
return 0
return ti.size
def _get_dirs(self):
if self.dir_cache is not None:
return
# This enables ls to get directories as children as well as files
self.dir_cache = {
dirname: {"name": dirname, "size": 0, "type": "directory"}
for dirname in self._all_dirnames(self.index)
}
self.dir_cache.update(
{info["name"]: info for info, _, _ in self.index.values()}
)
def _open(self, path, mode="rb", **kwargs):
if mode != "rb":
raise ValueError("Read-only filesystem implementation")
# Accept paths containing the archive's duplicate slashes too.
path = re.sub("/+", "/", path)
try:
details, _, orig_name = self.index[path]
except KeyError as exc:
raise FileNotFoundError(path) from exc
if details["type"] != "file":
raise ValueError("Can only handle regular files")
out = self.tar.extractfile(orig_name)
# cat_file needs the size to resolve negative offsets, as zip provides.
out.size = details["size"]
return out
def close(self):
"""Commits any write changes to the file. Done on ``del`` too."""
self.tar.close()
def __del__(self):
if hasattr(self, "tar"):
self.close()
del self.tar
if hasattr(self, "of") and hasattr(self.of, "__exit__"):
self.of.__exit__(None, None, None)