rel 2025

Author: FredHappyface

CHANGELOG.md

@@ -3,6 +3,11 @@
 All major and minor version changes will be documented in this file. Details of
 patch-level version changes can be found in [commit messages](../../commits/master).
 
+## 2025 - 2025/02/23
+
+- migrate to uv
+- fix: incompatibility with in memory (BytesIO) zip (https://github.com/FHPythonUtils/MutableZip/issues/1)
+
 ## 2021 - 2021/11/13
 
 - add pre-commit

mutablezip/__init__.py

@@ -2,14 +2,17 @@
 
 from __future__ import annotations
 
+import os
 from os import PathLike
-from os.path import join
+from pathlib import Path
 from shutil import copyfileobj, move, rmtree
 from tempfile import TemporaryFile, mkdtemp
 from types import TracebackType
 from typing import IO, Literal
 from zipfile import ZIP_STORED, ZipFile, ZipInfo
 
+from typing_extensions import Self
+
 
 class MutableZipFile(ZipFile):
 	"""
@@ -24,12 +27,43 @@ class DeleteMarker:
 
 	def __init__(
 		self,
-		file: str | IO[bytes],
+		file: str | IO[bytes] | os.PathLike,
 		mode: Literal["r", "w", "x", "a"] = "r",
 		compression: int = ZIP_STORED,
-		allowZip64: bool = False,
+		allowZip64: bool = True,  # noqa: FBT001, FBT002 # Normally, I'd address the boolean
+		# typed issue but here we need to maintain compat with ZipFile
+		compresslevel: int | None = None,
+		*,
+		strict_timestamps: bool = True,
 	) -> None:
-		super().__init__(file, mode=mode, compression=compression, allowZip64=allowZip64)
+		"""Open a ZIP file, where file can be a path to a file (a string), a
+		file-like object or a path-like object.
+
+		:param str | IO[bytes] | os.PathLike file: can be a path to a file (a string), a
+		file-like object or a path-like object.
+		:param Literal["r", "w", "x", "a"] mode: parameter should be 'r' to read an
+		existing file, 'w' to truncate and write a new file, 'a' to append to an existing
+		file, or 'x' to exclusively create and write a new file
+		:param int compression: the ZIP compression method to use when writing the
+		archive, and should be ZIP_STORED, ZIP_DEFLATED, ZIP_BZIP2 or ZIP_LZMA
+		:param bool allowZip64: s True (the default) zipfile will create ZIP files
+		that use the ZIP64 extensions when the zipfile is larger than 4 GiB.
+		:param int | None compresslevel: controls the compression level to use when
+		writing files to the archive. When using ZIP_STORED or ZIP_LZMA it has no effect.
+		When using ZIP_DEFLATED integers 0 through 9 are accepted
+		:param bool strict_timestamps: when set to False, allows to zip files older than
+		1980-01-01 and newer than 2107-12-31, defaults to True
+
+		https://docs.python.org/3/library/zipfile.html
+		"""
+		super().__init__(
+			file,
+			mode=mode,
+			compression=compression,
+			allowZip64=allowZip64,
+			compresslevel=compresslevel,
+			strict_timestamps=strict_timestamps,
+		)
 		# track file to override in zip
 		self._replace = {}
 		# Whether the with statement was called
@@ -48,6 +82,14 @@ def writestr(
 		compress_type: int | None = None,
 		compresslevel: int | None = None,
 	) -> None:
+		"""Write a file into the archive. The contents is data, which may be either a
+		str or a bytes instance; if it is a str, it is encoded as UTF-8 first.
+
+		zinfo_or_arcname is either the file name it will be given in the archive, or a
+		ZipInfo instance. If it's an instance, at least the filename, date, and time
+		must be given. If it's a name, the date and time is set to the current date and
+		time. The archive must be opened with mode 'w', 'x' or 'a'.
+		"""
 		if isinstance(zinfo_or_arcname, ZipInfo):
 			name = zinfo_or_arcname.filename
 		else:
@@ -56,7 +98,7 @@ def writestr(
 		# mark the entry, and create a temp-file for it
 		# we allow this only if the with statement is used
 		if self._allowUpdates and name in self.namelist():
-			tempFile = self._replace[name] = self._replace.get(name, TemporaryFile())
+			tempFile = self._replace.setdefault(name, TemporaryFile())
 			if isinstance(data, str):
 				tempFile.write(data.encode("utf-8"))  # strings are unicode
 			else:
@@ -77,14 +119,22 @@ def write(
 		compress_type: int | None = None,
 		compresslevel: int | None = None,
 	) -> None:
+		"""Write the file named filename to the archive, giving it the archive name
+		arcname (by default, this will be the same as filename, but without a drive
+		letter and with leading path separators removed). If given, compress_type
+		overrides the value given for the compression parameter to the constructor
+		for the new entry. Similarly, compresslevel will override the constructor if
+		given. The archive must be open with mode 'w', 'x' or 'a'.
+
+		"""
 		arcname = arcname or filename
 		# If the file exits, and needs to be overridden,
 		# mark the entry, and create a temp-file for it
 		# we allow this only if the with statement is used
 		if self._allowUpdates and arcname in self.namelist():
-			tempFile = self._replace[arcname] = self._replace.get(arcname, TemporaryFile())
-			with open(filename, "rb") as source:
+			with TemporaryFile() as tempFile, Path(filename).open("rb") as source:
 				copyfileobj(source, tempFile)
+
 		# Behave normally
 		else:
 			super().write(
@@ -94,7 +144,7 @@ def write(
 				compresslevel=compresslevel,
 			)
 
-	def __enter__(self):
+	def __enter__(self) -> Self:
 		# Allow updates
 		self._allowUpdates = True
 		return self
@@ -104,7 +154,7 @@ def __exit__(
 		exc_type: type[BaseException] | None,
 		exc_val: BaseException | None,
 		exc_tb: TracebackType | None,
-	):
+	) -> None:
 		# Call base to close zip
 		try:
 			super().__exit__(exc_type, exc_val, exc_tb)
@@ -128,37 +178,40 @@ def removeFile(self, path: str | PathLike[str]) -> None:
 	def _rebuildZip(self) -> None:
 		tempdir = mkdtemp()
 		try:
-			tempZipPath = join(tempdir, "new.zip")
-			with ZipFile(self.file, "r") as zipRead:
-				# Create new zip with assigned properties
-				with ZipFile(
-					tempZipPath,
-					"w",
-					compression=self.compression,
-					allowZip64=self.allowZip64,
-				) as zipWrite:
-					for item in zipRead.infolist():
-						# Check if the file should be replaced / or deleted
-						replacement = self._replace.get(item.filename, None)
-						# If marked for deletion, do not copy file to new zipfile
-						if isinstance(replacement, self.DeleteMarker):
-							del self._replace[item.filename]
-							continue
-						# If marked for replacement, copy temp_file, instead of old file
-						if replacement is not None:
-							del self._replace[item.filename]
-							# Write replacement to archive,
-							# and then close it (deleting the temp file)
-							replacement.seek(0)
-							data = replacement.read()
-							replacement.close()
-						else:
-							data = zipRead.read(item.filename)
-						zipWrite.writestr(item, data)
+			tempZipPath = Path(tempdir) / "new.zip"
+			with ZipFile(self.file, "r") as zipRead, ZipFile(
+				tempZipPath,
+				"w",
+				compression=self.compression,
+				allowZip64=self.allowZip64,
+			) as zipWrite:
+				for item in zipRead.infolist():
+					# Check if the file should be replaced / or deleted
+					replacement = self._replace.get(item.filename, None)
+					# If marked for deletion, do not copy file to new zipfile
+					if isinstance(replacement, self.DeleteMarker):
+						del self._replace[item.filename]
+						continue
+					# If marked for replacement, copy temp_file, instead of old file
+					if replacement is not None:
+						del self._replace[item.filename]
+						# Write replacement to archive,
+						# and then close it ,deleting the temp file
+						replacement.seek(0)
+						data = replacement.read()
+						replacement.close()
+					else:
+						data = zipRead.read(item.filename)
+					zipWrite.writestr(item, data)
 			# Override the archive with the updated one
 			if isinstance(self.file, str):
-				move(tempZipPath, self.file)
+				move(tempZipPath.as_posix(), self.file)
+			elif hasattr(self.file, "name"):
+				move(tempZipPath.as_posix(), self.file.name)
+			elif hasattr(self.file, "write"):
+				self.file.write(tempZipPath.read_bytes())
 			else:
-				move(tempZipPath, self.file.name)
+				msg = f"Sorry but {type(self.file).__name__} is not supported at this time!"
+				raise RuntimeError(msg)
 		finally:
 			rmtree(tempdir)

pyproject.toml

@@ -16,6 +16,9 @@ classifiers = [
     "Topic :: Software Development :: Libraries :: Python Modules",
     "Topic :: Utilities",
 ]
+dependencies = [
+    "typing-extensions>=4.12.2",
+]
 
 [project.urls]
 Homepage = "https://github.com/FHPythonUtils/MutableZip"
@@ -24,11 +27,11 @@ Documentation = "https://github.com/FHPythonUtils/MutableZip/blob/master/README.
 
 [dependency-groups]
 dev = [
-    "pytest>=8.1.1,<9",
-    "handsdown>=2.1.0,<3",
-    "coverage>=7.4.4,<8",
-    "ruff>=0.3.3,<0.4",
-    "pyright>=1.1.354,<2",
+    "coverage>=7.6.1",
+    "handsdown>=2.1.0",
+    "pyright>=1.1.394",
+    "pytest>=8.3.4",
+    "ruff>=0.9.7",
     "safety>=3.3.0",
 ]
 
@@ -40,7 +43,6 @@ target-version = "py38"
 [tool.ruff.lint]
 select = ["ALL"]
 ignore = [
-	"ANN101",  # type annotation for self in method
 	"COM812",  # enforce trailing comma
 	"D2",  # pydocstyle formatting
 	"ISC001",

tests/data/immutable.zip

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from io import BytesIO
 from pathlib import Path
 from zipfile import ZIP_DEFLATED, ZipFile
 
@@ -28,3 +29,18 @@ def test_ZipFile() -> None:
 		zipFile.writestr("foo.txt", b"\n".join(lines))
 	with ZipFile(f"{THISDIR}/data/immutable.zip", "r") as zipFile:
 		assert len(zipFile.namelist()) > 1
+
+
+def test_inmemory() -> None:
+
+	in_memory_zip = BytesIO()
+	with MutableZipFile(in_memory_zip, "w", compression=ZIP_DEFLATED) as file:
+		lines = [b"first line"]
+		file.writestr("foo.txt", b"\n".join(lines))
+
+	with MutableZipFile(in_memory_zip, "a", compression=ZIP_DEFLATED) as file:
+		lines = [b"new line"]
+		file.writestr("foo.txt", b"\n".join(lines))
+
+	with open(f"{THISDIR}/data/inmemory.zip", "wb") as file:
+		file.write(in_memory_zip.getvalue())