diff --git a/typesafety/test_upath_signatures.yml b/typesafety/test_upath_signatures.yml index 36d9f035..e3904e1e 100644 --- a/typesafety/test_upath_signatures.yml +++ b/typesafety/test_upath_signatures.yml @@ -90,6 +90,8 @@ cls: SMBPath - module: upath.implementations.webdav cls: WebdavPath + - module: upath.implementations.zip + cls: ZipPath - module: upath.extensions cls: ProxyUPath main: | @@ -271,6 +273,8 @@ cls: SMBPath - module: upath.implementations.webdav cls: WebdavPath + - module: upath.implementations.zip + cls: ZipPath - module: upath.extensions cls: ProxyUPath main: | @@ -582,6 +586,8 @@ cls: SMBPath - module: upath.implementations.webdav cls: WebdavPath + - module: upath.implementations.zip + cls: ZipPath - module: upath.extensions cls: ProxyUPath main: | @@ -972,6 +978,8 @@ cls: SMBPath - module: upath.implementations.webdav cls: WebdavPath + - module: upath.implementations.zip + cls: ZipPath - module: upath.extensions cls: ProxyUPath main: | diff --git a/typesafety/test_upath_types.yml b/typesafety/test_upath_types.yml index 333fdc20..1bd56815 100644 --- a/typesafety/test_upath_types.yml +++ b/typesafety/test_upath_types.yml @@ -49,6 +49,8 @@ protocol: smb - cls_fqn: upath.implementations.webdav.WebdavPath protocol: webdav + - cls_fqn: upath.implementations.zip.ZipPath + protocol: zip main: | from upath.registry import get_upath_class @@ -120,6 +122,8 @@ protocol: smb - cls_fqn: upath.implementations.webdav.WebdavPath protocol: webdav + - cls_fqn: upath.implementations.zip.ZipPath + protocol: zip - cls_fqn: upath.core.UPath protocol: unknown-protocol main: | @@ -201,6 +205,10 @@ cls: WebdavPath supported_example_name: base_url supported_example_value: '"https://webdav.example.com"' + - module: upath.implementations.zip + cls: ZipPath + supported_example_name: compression + supported_example_value: 9 main: | from {{ module }} import {{ cls }} @@ -262,6 +270,10 @@ cls: WebdavPath supported_example_name: base_url unsupported_example_value: '123' + - module: upath.implementations.zip + cls: ZipPath + supported_example_name: compression + unsupported_example_value: '"hello"' main: | from {{ module }} import {{ cls }} diff --git a/upath/core.py b/upath/core.py index adc1f270..02bfa8b4 100644 --- a/upath/core.py +++ b/upath/core.py @@ -561,6 +561,14 @@ def __new__( chain_parser: FSSpecChainParser = ..., **storage_options: Any, ) -> _uimpl.webdav.WebdavPath: ... + @overload # noqa: E301 + def __new__( + cls, + *args: JoinablePathLike, + protocol: Literal["zip"], + chain_parser: FSSpecChainParser = ..., + **storage_options: Any, + ) -> _uimpl.zip.ZipPath: ... if sys.platform == "win32": diff --git a/upath/implementations/zip.py b/upath/implementations/zip.py new file mode 100644 index 00000000..73500657 --- /dev/null +++ b/upath/implementations/zip.py @@ -0,0 +1,83 @@ +from __future__ import annotations + +import sys +from typing import TYPE_CHECKING +from zipfile import ZipInfo + +from upath.core import UPath +from upath.types import JoinablePathLike + +if TYPE_CHECKING: + from collections.abc import Iterator + from typing import Literal + + if sys.version_info >= (3, 11): + from typing import Self + from typing import Unpack + else: + from typing_extensions import Self + from typing_extensions import Unpack + + from upath._chain import FSSpecChainParser + from upath.types.storage_options import ZipStorageOptions + + +__all__ = ["ZipPath"] + + +class ZipPath(UPath): + __slots__ = () + + if TYPE_CHECKING: + + def __init__( + self, + *args: JoinablePathLike, + protocol: Literal["zip"] | None = ..., + chain_parser: FSSpecChainParser = ..., + **storage_options: Unpack[ZipStorageOptions], + ) -> None: ... + + def iterdir(self) -> Iterator[Self]: + if self.is_file(): + raise NotADirectoryError(str(self)) + yield from super().iterdir() + + if sys.version_info >= (3, 11): + + def mkdir( + self, + mode: int = 0o777, + parents: bool = False, + exist_ok: bool = False, + ) -> None: + is_dir = self.is_dir() + if is_dir and not exist_ok: + raise FileExistsError(f"File exists: {self.path!r}") + elif not is_dir: + zipfile = self.fs.zip + zipfile.mkdir(self.path, mode) + + else: + + def mkdir( + self, + mode: int = 0o777, + parents: bool = False, + exist_ok: bool = False, + ) -> None: + is_dir = self.is_dir() + if is_dir and not exist_ok: + raise FileExistsError(f"File exists: {self.path!r}") + elif not is_dir: + dirname = self.path + if dirname and not dirname.endswith("/"): + dirname += "/" + zipfile = self.fs.zip + zinfo = ZipInfo(dirname) + zinfo.compress_size = 0 + zinfo.CRC = 0 + zinfo.external_attr = ((0o40000 | mode) & 0xFFFF) << 16 + zinfo.file_size = 0 + zinfo.external_attr |= 0x10 + zipfile.writestr(zinfo, b"") diff --git a/upath/registry.py b/upath/registry.py index 706914fc..36d2b2d1 100644 --- a/upath/registry.py +++ b/upath/registry.py @@ -67,6 +67,7 @@ from upath.implementations.sftp import SFTPPath as _SFTPPath from upath.implementations.smb import SMBPath as _SMBPath from upath.implementations.webdav import WebdavPath as _WebdavPath + from upath.implementations.zip import ZipPath as _ZipPath __all__ = [ @@ -106,6 +107,7 @@ class _Registry(MutableMapping[str, "type[upath.UPath]"]): "webdav+https": "upath.implementations.webdav.WebdavPath", "github": "upath.implementations.github.GitHubPath", "smb": "upath.implementations.smb.SMBPath", + "zip": "upath.implementations.zip.ZipPath", } if TYPE_CHECKING: @@ -235,6 +237,8 @@ def get_upath_class(protocol: Literal["sftp", "ssh"]) -> type[_SFTPPath]: ... def get_upath_class(protocol: Literal["smb"]) -> type[_SMBPath]: ... @overload def get_upath_class(protocol: Literal["webdav"]) -> type[_WebdavPath]: ... + @overload + def get_upath_class(protocol: Literal["zip"]) -> type[_ZipPath]: ... if sys.platform == "win32": diff --git a/upath/tests/cases.py b/upath/tests/cases.py index 0d397df3..059e5cb4 100644 --- a/upath/tests/cases.py +++ b/upath/tests/cases.py @@ -316,6 +316,12 @@ def test_touch_exists_ok_true(self): f.touch(exist_ok=True) assert f.read_text() == data + def test_touch(self): + path = self.path.joinpath("test_touch.txt") + assert not path.exists() + path.touch() + assert path.exists() + def test_touch_unlink(self): path = self.path.joinpath("test_touch.txt") path.touch() diff --git a/upath/tests/implementations/test_data.py b/upath/tests/implementations/test_data.py index 9f2f0077..5b1fe92e 100644 --- a/upath/tests/implementations/test_data.py +++ b/upath/tests/implementations/test_data.py @@ -133,6 +133,9 @@ def test_rglob(self, pathlib_base): with pytest.raises(NotImplementedError): list(self.path.rglob("*")) + def test_touch(self): + self.path.touch() + def test_touch_exists_ok_false(self): with pytest.raises(FileExistsError): self.path.touch(exist_ok=False) diff --git a/upath/tests/implementations/test_github.py b/upath/tests/implementations/test_github.py index dbcf8431..5bf3553c 100644 --- a/upath/tests/implementations/test_github.py +++ b/upath/tests/implementations/test_github.py @@ -66,6 +66,10 @@ def test_rename(self): def test_rename2(self): pass + @pytest.mark.skip(reason="GitHub filesystem is read-only") + def test_touch(self): + pass + @pytest.mark.skip(reason="GitHub filesystem is read-only") def test_touch_unlink(self): pass diff --git a/upath/tests/implementations/test_http.py b/upath/tests/implementations/test_http.py index 473bb95d..25fe15aa 100644 --- a/upath/tests/implementations/test_http.py +++ b/upath/tests/implementations/test_http.py @@ -93,6 +93,10 @@ def test_makedirs_exist_ok_true(self): def test_makedirs_exist_ok_false(self): pass + @pytest.mark.skip + def test_touch(self): + pass + @pytest.mark.skip def test_touch_unlink(self): pass diff --git a/upath/tests/implementations/test_zip.py b/upath/tests/implementations/test_zip.py new file mode 100644 index 00000000..d4ceae76 --- /dev/null +++ b/upath/tests/implementations/test_zip.py @@ -0,0 +1,147 @@ +import os +import zipfile + +import pytest + +from upath import UPath +from upath.implementations.zip import ZipPath + +from ..cases import BaseTests + + +@pytest.fixture(scope="function") +def zipped_testdir_file(local_testdir, tmp_path_factory): + base = tmp_path_factory.mktemp("zippath") + zip_path = base / "test.zip" + with zipfile.ZipFile(zip_path, "w") as zf: + for root, _, files in os.walk(local_testdir): + for file in files: + full_path = os.path.join(root, file) + arcname = os.path.relpath(full_path, start=local_testdir) + zf.write(full_path, arcname=arcname) + return str(zip_path) + + +@pytest.fixture(scope="function") +def empty_zipped_testdir_file(tmp_path): + tmp_path = tmp_path.joinpath("zippath") + tmp_path.mkdir() + zip_path = tmp_path / "test.zip" + + with zipfile.ZipFile(zip_path, "w"): + pass + return str(zip_path) + + +class TestZipPath(BaseTests): + + @pytest.fixture(autouse=True) + def path(self, zipped_testdir_file, request): + try: + (mode,) = request.param + except (ValueError, TypeError, AttributeError): + mode = "r" + self.path = UPath("zip://", fo=zipped_testdir_file, mode=mode) + # self.prepare_file_system() done outside of UPath + + def test_is_ZipPath(self): + assert isinstance(self.path, ZipPath) + + @pytest.mark.parametrize( + "path", [("w",)], ids=["zipfile_mode_write"], indirect=True + ) + def test_mkdir(self): + super().test_mkdir() + + @pytest.mark.parametrize( + "path", [("w",)], ids=["zipfile_mode_write"], indirect=True + ) + def test_mkdir_exists_ok_true(self): + super().test_mkdir_exists_ok_true() + + @pytest.mark.parametrize( + "path", [("w",)], ids=["zipfile_mode_write"], indirect=True + ) + def test_mkdir_exists_ok_false(self): + super().test_mkdir_exists_ok_false() + + @pytest.mark.parametrize( + "path", [("w",)], ids=["zipfile_mode_write"], indirect=True + ) + def test_mkdir_parents_true_exists_ok_true(self): + super().test_mkdir_parents_true_exists_ok_true() + + @pytest.mark.parametrize( + "path", [("w",)], ids=["zipfile_mode_write"], indirect=True + ) + def test_mkdir_parents_true_exists_ok_false(self): + super().test_mkdir_parents_true_exists_ok_false() + + def test_rename(self): + with pytest.raises(NotImplementedError): + super().test_rename() # delete is not implemented in fsspec + + def test_rename2(self): + with pytest.raises(NotImplementedError): + super().test_rename2() # delete is not implemented in fsspec + + def test_move_local(self, tmp_path): + with pytest.raises(NotImplementedError): + super().test_move_local(tmp_path) # delete is not implemented in fsspec + + def test_move_into_local(self, tmp_path): + with pytest.raises(NotImplementedError): + super().test_move_into_local( + tmp_path + ) # delete is not implemented in fsspec + + def test_move_memory(self, clear_fsspec_memory_cache): + with pytest.raises(NotImplementedError): + super().test_move_memory(clear_fsspec_memory_cache) + + def test_move_into_memory(self, clear_fsspec_memory_cache): + with pytest.raises(NotImplementedError): + super().test_move_into_memory(clear_fsspec_memory_cache) + + @pytest.mark.parametrize( + "path", [("w",)], ids=["zipfile_mode_write"], indirect=True + ) + def test_touch(self): + super().test_touch() + + @pytest.mark.parametrize( + "path", [("w",)], ids=["zipfile_mode_write"], indirect=True + ) + def test_touch_unlink(self): + with pytest.raises(NotImplementedError): + super().test_touch_unlink() # delete is not implemented in fsspec + + @pytest.mark.parametrize( + "path", [("w",)], ids=["zipfile_mode_write"], indirect=True + ) + def test_write_bytes(self): + fn = "test_write_bytes.txt" + s = b"hello_world" + path = self.path.joinpath(fn) + path.write_bytes(s) + so = {**path.storage_options, "mode": "r"} + urlpath = str(path) + path.fs.close() + assert UPath(urlpath, **so).read_bytes() == s + + @pytest.mark.parametrize( + "path", [("w",)], ids=["zipfile_mode_write"], indirect=True + ) + def test_write_text(self): + fn = "test_write_text.txt" + s = "hello_world" + path = self.path.joinpath(fn) + path.write_text(s) + so = {**path.storage_options, "mode": "r"} + urlpath = str(path) + path.fs.close() + assert UPath(urlpath, **so).read_text() == s + + @pytest.mark.skip(reason="fsspec zipfile filesystem is either read xor write mode") + def test_fsspec_compat(self): + pass diff --git a/upath/tests/test_registry.py b/upath/tests/test_registry.py index 3d6e1a38..ce300da8 100644 --- a/upath/tests/test_registry.py +++ b/upath/tests/test_registry.py @@ -37,6 +37,7 @@ "webdav+http", "webdav+https", "github", + "zip", } diff --git a/upath/types/storage_options.py b/upath/types/storage_options.py index f156f221..c5254a1a 100644 --- a/upath/types/storage_options.py +++ b/upath/types/storage_options.py @@ -28,6 +28,7 @@ "SFTPStorageOptions", "SMBStorageOptions", "WebdavStorageOptions", + "ZipStorageOptions", ] @@ -358,3 +359,20 @@ class WebdavStorageOptions(_AbstractStorageOptions, total=False): # Client configuration client: Any | None # webdav4.client.Client instance + + +class ZipStorageOptions( + _AbstractStorageOptions, + _ChainableStorageOptions, + total=False, +): + """Storage options for ZIP archive filesystem""" + + # Archive file settings + fo: str | Any # Path to ZIP file or file-like object + mode: Literal["r", "w", "a"] # Open mode: read, write, or append + + # ZIP compression settings + compression: int # Compression method (e.g., zipfile.ZIP_STORED, ZIP_DEFLATED) + allowZip64: bool # Enable ZIP64 extensions for large files + compresslevel: int | None # Compression level (None uses default for method)