# virtual-fs
Powerful Virtual File abstraction api. Works without `FUSE`. Run in unprivileged docker container. Connects to any backend supported by Rclone. Drop in replacement for pathlib.Path. Works with both local and remote files. If you have an `rclone.conf` file in a default path then this api will allow you access to paths like `remote:Bucket/path/file.txt`.
## ENVS
* RCLONE_CONFIG
* path string of the rclone.conf text file
* RCLONE_CONFIG_JSON
* string content of rclone config.json
## Vs others
* fsspec - good alternative, but weakly typed.
* libfuse - this is a mount, virtual-fs is not a mount but an api and therefore can run in docker for unprivileged runtimes.
## Docker Users
This library is built for you. If you are trying to do a `/mount` and having problems because of privileges then this api will give you an escape hatch. Instead of mounting a virtual file system, you use an api in python that will grant you `ls`, `read`, `write` and directory traversal.
To retro fit your code: Swap out `pathlib.Path` for `virtual_fs.FSPath` and apply minor fixes.
```python
from virtual_fs import Vfs
def unit_test():
config = Path("rclone.config") # Or use None to get a default.
cwd = Vfs.begin("remote:bucket/my", config=config)
do_test(cwd)
def unit_test2():
with Vfs.begin("mydir") as cwd: # Closes filesystem when done on cwd.
do_test(cwd)
def do_test(cwd: FSPath):
file = cwd / "info.json"
text = file.read_text()
out = cwd / "out.json"
out.write_text(out)
files, dirs = cwd.ls()
print(f"Found {len(files)} files")
assert 2 == len(files), f"Expected 2 files, but had {len(files)}"
assert 0 == len(dirs), f"Expected 0 dirs, but had {len(dirs)}"
```
This abstraction is made possible thanks to [rclone](https://rclone.org) and my python api bindings called [rclone-api](https://github.com/zackees/rclone-api).
Easily convert your `pathlib.Path` into an `FSPath`, which will either operate on a local file object, or one on a remote.
```python
class FSPath:
def __init__(self, fs: FS, path: str) -> None:
self.fs: FS = fs
self.path: str = path
self.fs_holder: FS | None = None
def set_owner(self) -> None:
self.fs_holder = self.fs
def is_real_fs(self) -> bool:
return isinstance(self.fs, RealFS)
def lspaths(self) -> "tuple[list[FSPath], list[FSPath]]":
filenames, dirnames = self.ls()
fpaths: list[FSPath] = [self / name for name in filenames]
dpaths: list[FSPath] = [self / name for name in dirnames]
return fpaths, dpaths
def ls(self) -> tuple[list[str], list[str]]:
filenames: list[str]
dirnames: list[str]
filenames, dirnames = self.fs.ls(self.path)
return filenames, dirnames
def mkdir(self, parents=True, exist_ok=True) -> None:
self.fs.mkdir(self.path, parents=parents, exist_ok=exist_ok)
def read_text(self) -> str:
data = self.read_bytes()
return data.decode("utf-8")
def read_bytes(self) -> bytes:
data: bytes | None = None
try:
data = self.fs.read_bytes(self.path)
return data
except Exception as e:
raise FileNotFoundError(f"File not found: {self.path}, because of {e}")
def exists(self) -> bool:
return self.fs.exists(self.path)
def __str__(self) -> str:
return self.path
def __repr__(self) -> str:
return f"FSPath({self.path})"
def __enter__(self) -> "FSPath":
if self.fs_holder is not None:
warnings.warn("This operation is reserved for the cwd returned by FS")
return self
def __exit__(self, exc_type, exc_value, traceback) -> None:
if self.fs_holder is not None:
self.fs_holder.dispose()
self.fs_holder = None
def write_text(self, data: str, encoding: str | None = None) -> None:
if encoding is None:
encoding = "utf-8"
self.write_bytes(data.encode(encoding))
def write_bytes(self, data: bytes) -> None:
self.fs.write_binary(self.path, data)
def rmtree(self, ignore_errors=False) -> None:
assert self.exists(), f"Path does not exist: {self.path}"
# check fs is RealFS
assert isinstance(self.fs, RealFS)
shutil.rmtree(self.path, ignore_errors=ignore_errors)
@property
def name(self) -> str:
return Path(self.path).name
@property
def parent(self) -> "FSPath":
parent_path = Path(self.path).parent
parent_str = parent_path.as_posix()
return FSPath(self.fs, parent_str)
def __truediv__(self, other: str) -> "FSPath":
new_path = Path(self.path) / other
return FSPath(self.fs, new_path.as_posix())
# hashable
def __hash__(self) -> int:
return hash(f"{repr(self.fs)}:{self.path}")
```
Raw data
{
"_id": null,
"home_page": "https://github.com/zackees/virtual-fs",
"name": "virtual-fs",
"maintainer": "Zachary Vorhies",
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "template-python-cmd",
"author": null,
"author_email": null,
"download_url": null,
"platform": null,
"description": "# virtual-fs\r\n\r\n\r\n\r\nPowerful Virtual File abstraction api. Works without `FUSE`. Run in unprivileged docker container. Connects to any backend supported by Rclone. Drop in replacement for pathlib.Path. Works with both local and remote files. If you have an `rclone.conf` file in a default path then this api will allow you access to paths like `remote:Bucket/path/file.txt`.\r\n\r\n\r\n## ENVS\r\n\r\n * RCLONE_CONFIG\r\n * path string of the rclone.conf text file\r\n\r\n * RCLONE_CONFIG_JSON\r\n * string content of rclone config.json\r\n \r\n## Vs others\r\n\r\n * fsspec - good alternative, but weakly typed.\r\n * libfuse - this is a mount, virtual-fs is not a mount but an api and therefore can run in docker for unprivileged runtimes.\r\n\r\n## Docker Users\r\n\r\nThis library is built for you. If you are trying to do a `/mount` and having problems because of privileges then this api will give you an escape hatch. Instead of mounting a virtual file system, you use an api in python that will grant you `ls`, `read`, `write` and directory traversal.\r\n\r\nTo retro fit your code: Swap out `pathlib.Path` for `virtual_fs.FSPath` and apply minor fixes.\r\n\r\n\r\n```python\r\n\r\nfrom virtual_fs import Vfs\r\n\r\ndef unit_test():\r\n config = Path(\"rclone.config\") # Or use None to get a default.\r\n cwd = Vfs.begin(\"remote:bucket/my\", config=config)\r\n do_test(cwd)\r\n\r\ndef unit_test2():\r\n with Vfs.begin(\"mydir\") as cwd: # Closes filesystem when done on cwd.\r\n do_test(cwd)\r\n\r\ndef do_test(cwd: FSPath):\r\n file = cwd / \"info.json\"\r\n text = file.read_text()\r\n out = cwd / \"out.json\"\r\n out.write_text(out)\r\n files, dirs = cwd.ls()\r\n print(f\"Found {len(files)} files\")\r\n assert 2 == len(files), f\"Expected 2 files, but had {len(files)}\"\r\n assert 0 == len(dirs), f\"Expected 0 dirs, but had {len(dirs)}\"\r\n\r\n\r\n```\r\n\r\n\r\n\r\nThis abstraction is made possible thanks to [rclone](https://rclone.org) and my python api bindings called [rclone-api](https://github.com/zackees/rclone-api).\r\n\r\nEasily convert your `pathlib.Path` into an `FSPath`, which will either operate on a local file object, or one on a remote.\r\n\r\n\r\n\r\n```python\r\nclass FSPath:\r\n def __init__(self, fs: FS, path: str) -> None:\r\n self.fs: FS = fs\r\n self.path: str = path\r\n self.fs_holder: FS | None = None\r\n\r\n def set_owner(self) -> None:\r\n self.fs_holder = self.fs\r\n\r\n def is_real_fs(self) -> bool:\r\n return isinstance(self.fs, RealFS)\r\n \r\n def lspaths(self) -> \"tuple[list[FSPath], list[FSPath]]\":\r\n filenames, dirnames = self.ls()\r\n fpaths: list[FSPath] = [self / name for name in filenames]\r\n dpaths: list[FSPath] = [self / name for name in dirnames]\r\n return fpaths, dpaths\r\n\r\n def ls(self) -> tuple[list[str], list[str]]:\r\n filenames: list[str]\r\n dirnames: list[str]\r\n filenames, dirnames = self.fs.ls(self.path)\r\n return filenames, dirnames\r\n\r\n def mkdir(self, parents=True, exist_ok=True) -> None:\r\n self.fs.mkdir(self.path, parents=parents, exist_ok=exist_ok)\r\n\r\n def read_text(self) -> str:\r\n data = self.read_bytes()\r\n return data.decode(\"utf-8\")\r\n\r\n def read_bytes(self) -> bytes:\r\n data: bytes | None = None\r\n try:\r\n data = self.fs.read_bytes(self.path)\r\n return data\r\n except Exception as e:\r\n raise FileNotFoundError(f\"File not found: {self.path}, because of {e}\")\r\n\r\n def exists(self) -> bool:\r\n return self.fs.exists(self.path)\r\n\r\n def __str__(self) -> str:\r\n return self.path\r\n\r\n def __repr__(self) -> str:\r\n return f\"FSPath({self.path})\"\r\n\r\n def __enter__(self) -> \"FSPath\":\r\n if self.fs_holder is not None:\r\n warnings.warn(\"This operation is reserved for the cwd returned by FS\")\r\n return self\r\n\r\n def __exit__(self, exc_type, exc_value, traceback) -> None:\r\n if self.fs_holder is not None:\r\n self.fs_holder.dispose()\r\n self.fs_holder = None\r\n\r\n\r\n\r\n def write_text(self, data: str, encoding: str | None = None) -> None:\r\n if encoding is None:\r\n encoding = \"utf-8\"\r\n self.write_bytes(data.encode(encoding))\r\n\r\n def write_bytes(self, data: bytes) -> None:\r\n self.fs.write_binary(self.path, data)\r\n\r\n def rmtree(self, ignore_errors=False) -> None:\r\n assert self.exists(), f\"Path does not exist: {self.path}\"\r\n # check fs is RealFS\r\n assert isinstance(self.fs, RealFS)\r\n shutil.rmtree(self.path, ignore_errors=ignore_errors)\r\n\r\n\r\n\r\n @property\r\n def name(self) -> str:\r\n return Path(self.path).name\r\n\r\n @property\r\n def parent(self) -> \"FSPath\":\r\n parent_path = Path(self.path).parent\r\n parent_str = parent_path.as_posix()\r\n return FSPath(self.fs, parent_str)\r\n\r\n def __truediv__(self, other: str) -> \"FSPath\":\r\n new_path = Path(self.path) / other\r\n return FSPath(self.fs, new_path.as_posix())\r\n\r\n # hashable\r\n def __hash__(self) -> int:\r\n return hash(f\"{repr(self.fs)}:{self.path}\")\r\n```\r\n",
"bugtrack_url": null,
"license": "BSD 3-Clause License",
"summary": "virtual file system for python, api level virtual mounting",
"version": "1.0.19",
"project_urls": {
"Homepage": "https://github.com/zackees/virtual-fs"
},
"split_keywords": [
"template-python-cmd"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "64678ad1e3117cf1e2b3a6e825d378839af2029da8bc356a23bbefba70cb4b4a",
"md5": "2c7b9fc404d423a89c9923204a7fa748",
"sha256": "12b88b0976455ecbdc2570623882229df5534fa912c7b5d2fbeef3c1a46f8a18"
},
"downloads": -1,
"filename": "virtual_fs-1.0.19-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2c7b9fc404d423a89c9923204a7fa748",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 7918,
"upload_time": "2025-03-29T00:50:11",
"upload_time_iso_8601": "2025-03-29T00:50:11.022896Z",
"url": "https://files.pythonhosted.org/packages/64/67/8ad1e3117cf1e2b3a6e825d378839af2029da8bc356a23bbefba70cb4b4a/virtual_fs-1.0.19-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-03-29 00:50:11",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "zackees",
"github_project": "virtual-fs",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"tox": true,
"lcname": "virtual-fs"
}
JSON
