bpo-14678: Update zipimport to support importlib.invalidate_caches() (GH-24159)

Added an invalidate_caches() method to the zipimport.zipimporter class based on the implementation of importlib.FileFinder.invalidate_caches(). This was done by adding a get_files() method and an _archive_mtime attribute to zipimport.zipimporter to check for updates or cache invalidation whenever the cache of files and toc entry information in the zipimporter is accessed.

Author: desmondcheongzx

Doc/library/zipimport.rst

@@ -166,6 +166,15 @@ zipimporter Objects
 
          Use :meth:`exec_module` instead.
 
+
+   .. method:: invalidate_caches()
+
+      Clear out the internal cache of information about files found within
+      the ZIP archive.
+
+      .. versionadded:: 3.10
+
+
    .. attribute:: archive
 
       The file name of the importer's associated ZIP file, without a possible

Lib/test/test_zipimport.py

@@ -506,6 +506,47 @@ def testZipImporterMethods(self):
         self.assertEqual(zi2.archive, TEMP_ZIP)
         self.assertEqual(zi2.prefix, TESTPACK + os.sep)
 
+    def testInvalidateCaches(self):
+        packdir = TESTPACK + os.sep
+        packdir2 = packdir + TESTPACK2 + os.sep
+        files = {packdir + "__init__" + pyc_ext: (NOW, test_pyc),
+                 packdir2 + "__init__" + pyc_ext: (NOW, test_pyc),
+                 packdir2 + TESTMOD + pyc_ext: (NOW, test_pyc),
+                 "spam" + pyc_ext: (NOW, test_pyc)}
+        self.addCleanup(os_helper.unlink, TEMP_ZIP)
+        with ZipFile(TEMP_ZIP, "w") as z:
+            for name, (mtime, data) in files.items():
+                zinfo = ZipInfo(name, time.localtime(mtime))
+                zinfo.compress_type = self.compression
+                zinfo.comment = b"spam"
+                z.writestr(zinfo, data)
+
+        zi = zipimport.zipimporter(TEMP_ZIP)
+        self.assertEqual(zi._files.keys(), files.keys())
+        # Check that the file information remains accurate after reloading
+        zi.invalidate_caches()
+        self.assertEqual(zi._files.keys(), files.keys())
+        # Add a new file to the ZIP archive
+        newfile = {"spam2" + pyc_ext: (NOW, test_pyc)}
+        files.update(newfile)
+        with ZipFile(TEMP_ZIP, "a") as z:
+            for name, (mtime, data) in newfile.items():
+                zinfo = ZipInfo(name, time.localtime(mtime))
+                zinfo.compress_type = self.compression
+                zinfo.comment = b"spam"
+                z.writestr(zinfo, data)
+        # Check that we can detect the new file after invalidating the cache
+        zi.invalidate_caches()
+        self.assertEqual(zi._files.keys(), files.keys())
+        spec = zi.find_spec('spam2')
+        self.assertIsNotNone(spec)
+        self.assertIsInstance(spec.loader, zipimport.zipimporter)
+        # Check that the cached data is removed if the file is deleted
+        os.remove(TEMP_ZIP)
+        zi.invalidate_caches()
+        self.assertIsNone(zi._files)
+        self.assertIsNone(zipimport._zip_directory_cache.get(zi.archive))
+
     def testZipImporterMethodsInSubDirectory(self):
         packdir = TESTPACK + os.sep
         packdir2 = packdir + TESTPACK2 + os.sep

Lib/zipimport.py

@@ -321,6 +321,16 @@ def get_resource_reader(self, fullname):
         return ZipReader(self, fullname)
 
 
+    def invalidate_caches(self):
+        """Reload the file data of the archive path."""
+        try:
+            self._files = _read_directory(self.archive)
+            _zip_directory_cache[self.archive] = self._files
+        except ZipImportError:
+            _zip_directory_cache.pop(self.archive, None)
+            self._files = None
+
+
     def __repr__(self):
         return f'<zipimporter object "{self.archive}{path_sep}{self.prefix}">'
 

Misc/NEWS.d/next/Library/2021-01-07-21-25-49.bpo-14678.1zniCH.rst

@@ -0,0 +1,3 @@
+Add an invalidate_caches() method to the zipimport.zipimporter class to
+support importlib.invalidate_caches().
+Patch by Desmond Cheong.