Add upath.implementations.zip.ZipPath

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: |

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 }}
 

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":
 

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"")

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":
 

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()

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)

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

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

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