twisted/python/zippath.py

   1 # -*- test-case-name: twisted.test.test_paths.ZipFilePathTestCase -*-
   2 # Copyright (c) Twisted Matrix Laboratories.
   3 # See LICENSE for details.
   4
   5 """
   6 This module contains implementations of IFilePath for zip files.
   7
   8 See the constructor for ZipArchive for use.
   9 """
  10
  11 __metaclass__ = type
  12
  13 import os
  14 import time
  15 import errno
  16
  17
  18 # Python 2.6 includes support for incremental unzipping of zipfiles, and
  19 # thus obviates the need for ChunkingZipFile.
  20 import sys
  21 if sys.version_info[:2] >= (2, 6):
  22     _USE_ZIPFILE = True
  23     from zipfile import ZipFile
  24 else:
  25     _USE_ZIPFILE = False
  26     from twisted.python.zipstream import ChunkingZipFile
  27
  28 from twisted.python.filepath import IFilePath, FilePath, AbstractFilePath
  29
  30 from zope.interface import implements
  31
  32 # using FilePath here exclusively rather than os to make sure that we don't do
  33 # anything OS-path-specific here.
  34
  35 ZIP_PATH_SEP = '/'              # In zipfiles, "/" is universally used as the
  36                                 # path separator, regardless of platform.
  37
  38
  39 class ZipPath(AbstractFilePath):
  40     """
  41     I represent a file or directory contained within a zip file.
  42     """
  43
  44     implements(IFilePath)
  45
  46     sep = ZIP_PATH_SEP
  47
  48     def __init__(self, archive, pathInArchive):
  49         """
  50         Don't construct me directly.  Use ZipArchive.child().
  51
  52         @param archive: a ZipArchive instance.
  53
  54         @param pathInArchive: a ZIP_PATH_SEP-separated string.
  55         """
  56         self.archive = archive
  57         self.pathInArchive = pathInArchive
  58         # self.path pretends to be os-specific because that's the way the
  59         # 'zipimport' module does it.
  60         self.path = os.path.join(archive.zipfile.filename,
  61                                  *(self.pathInArchive.split(ZIP_PATH_SEP)))
  62
  63     def __cmp__(self, other):
  64         if not isinstance(other, ZipPath):
  65             return NotImplemented
  66         return cmp((self.archive, self.pathInArchive),
  67                    (other.archive, other.pathInArchive))
  68
  69
  70     def __repr__(self):
  71         parts = [os.path.abspath(self.archive.path)]
  72         parts.extend(self.pathInArchive.split(ZIP_PATH_SEP))
  73         path = os.sep.join(parts)
  74         return "ZipPath('%s')" % (path.encode('string-escape'),)
  75
  76
  77     def parent(self):
  78         splitup = self.pathInArchive.split(ZIP_PATH_SEP)
  79         if len(splitup) == 1:
  80             return self.archive
  81         return ZipPath(self.archive, ZIP_PATH_SEP.join(splitup[:-1]))
  82
  83
  84     def child(self, path):
  85         """
  86         Return a new ZipPath representing a path in C{self.archive} which is
  87         a child of this path.
  88
  89         @note: Requesting the C{".."} (or other special name) child will not
  90             cause L{InsecurePath} to be raised since these names do not have
  91             any special meaning inside a zip archive.  Be particularly
  92             careful with the C{path} attribute (if you absolutely must use
  93             it) as this means it may include special names with special
  94             meaning outside of the context of a zip archive.
  95         """
  96         return ZipPath(self.archive, ZIP_PATH_SEP.join([self.pathInArchive, path]))
  97
  98
  99     def sibling(self, path):
 100         return self.parent().child(path)
 101
 102     # preauthChild = child
 103
 104     def exists(self):
 105         return self.isdir() or self.isfile()
 106
 107     def isdir(self):
 108         return self.pathInArchive in self.archive.childmap
 109
 110     def isfile(self):
 111         return self.pathInArchive in self.archive.zipfile.NameToInfo
 112
 113     def islink(self):
 114         return False
 115
 116     def listdir(self):
 117         if self.exists():
 118             if self.isdir():
 119                 return self.archive.childmap[self.pathInArchive].keys()
 120             else:
 121                 raise OSError(errno.ENOTDIR, "Leaf zip entry listed")
 122         else:
 123             raise OSError(errno.ENOENT, "Non-existent zip entry listed")
 124
 125
 126     def splitext(self):
 127         """
 128         Return a value similar to that returned by os.path.splitext.
 129         """
 130         # This happens to work out because of the fact that we use OS-specific
 131         # path separators in the constructor to construct our fake 'path'
 132         # attribute.
 133         return os.path.splitext(self.path)
 134
 135
 136     def basename(self):
 137         return self.pathInArchive.split(ZIP_PATH_SEP)[-1]
 138
 139     def dirname(self):
 140         # XXX NOTE: This API isn't a very good idea on filepath, but it's even
 141         # less meaningful here.
 142         return self.parent().path
 143
 144     def open(self, mode="r"):
 145         if _USE_ZIPFILE:
 146             return self.archive.zipfile.open(self.pathInArchive, mode=mode)
 147         else:
 148             # XXX oh man, is this too much hax?
 149             self.archive.zipfile.mode = mode
 150             return self.archive.zipfile.readfile(self.pathInArchive)
 151
 152     def changed(self):
 153         pass
 154
 155     def getsize(self):
 156         """
 157         Retrieve this file's size.
 158
 159         @return: file size, in bytes
 160         """
 161
 162         return self.archive.zipfile.NameToInfo[self.pathInArchive].file_size
 163
 164     def getAccessTime(self):
 165         """
 166         Retrieve this file's last access-time.  This is the same as the last access
 167         time for the archive.
 168
 169         @return: a number of seconds since the epoch
 170         """
 171         return self.archive.getAccessTime()
 172
 173
 174     def getModificationTime(self):
 175         """
 176         Retrieve this file's last modification time.  This is the time of
 177         modification recorded in the zipfile.
 178
 179         @return: a number of seconds since the epoch.
 180         """
 181         return time.mktime(
 182             self.archive.zipfile.NameToInfo[self.pathInArchive].date_time
 183             + (0, 0, 0))
 184
 185
 186     def getStatusChangeTime(self):
 187         """
 188         Retrieve this file's last modification time.  This name is provided for
 189         compatibility, and returns the same value as getmtime.
 190
 191         @return: a number of seconds since the epoch.
 192         """
 193         return self.getModificationTime()
 194
 195
 196
 197 class ZipArchive(ZipPath):
 198     """ I am a FilePath-like object which can wrap a zip archive as if it were a
 199     directory.
 200     """
 201     archive = property(lambda self: self)
 202     def __init__(self, archivePathname):
 203         """Create a ZipArchive, treating the archive at archivePathname as a zip file.
 204
 205         @param archivePathname: a str, naming a path in the filesystem.
 206         """
 207         if _USE_ZIPFILE:
 208             self.zipfile = ZipFile(archivePathname)
 209         else:
 210             self.zipfile = ChunkingZipFile(archivePathname)
 211         self.path = archivePathname
 212         self.pathInArchive = ''
 213         # zipfile is already wasting O(N) memory on cached ZipInfo instances,
 214         # so there's no sense in trying to do this lazily or intelligently
 215         self.childmap = {}      # map parent: list of children
 216
 217         for name in self.zipfile.namelist():
 218             name = name.split(ZIP_PATH_SEP)
 219             for x in range(len(name)):
 220                 child = name[-x]
 221                 parent = ZIP_PATH_SEP.join(name[:-x])
 222                 if parent not in self.childmap:
 223                     self.childmap[parent] = {}
 224                 self.childmap[parent][child] = 1
 225             parent = ''
 226
 227     def child(self, path):
 228         """
 229         Create a ZipPath pointing at a path within the archive.
 230
 231         @param path: a str with no path separators in it, either '/' or the
 232         system path separator, if it's different.
 233         """
 234         return ZipPath(self, path)
 235
 236     def exists(self):
 237         """
 238         Returns true if the underlying archive exists.
 239         """
 240         return FilePath(self.zipfile.filename).exists()
 241
 242
 243     def getAccessTime(self):
 244         """
 245         Return the archive file's last access time.
 246         """
 247         return FilePath(self.zipfile.filename).getAccessTime()
 248
 249
 250     def getModificationTime(self):
 251         """
 252         Return the archive file's modification time.
 253         """
 254         return FilePath(self.zipfile.filename).getModificationTime()
 255
 256
 257     def getStatusChangeTime(self):
 258         """
 259         Return the archive file's status change time.
 260         """
 261         return FilePath(self.zipfile.filename).getStatusChangeTime()
 262
 263
 264     def __repr__(self):
 265         return 'ZipArchive(%r)' % (os.path.abspath(self.path),)
 266
 267
 268 __all__ = ['ZipArchive', 'ZipPath']