shutil

1 """Utility functions for copying and archiving files and directory trees. 2 3 XXX The functions here don't copy the resource fork or other metadata on Mac. 4 5 """ 6 7 import os 8 import sys 9 import stat 10 from os.path import abspath 11 import fnmatch 12 import collections 13 import errno 14 15 try: 16 from pwd import getpwnam 17 except ImportError: 18 getpwnam = None 19 20 try: 21 from grp import getgrnam 22 except ImportError: 23 getgrnam = None 24 25 __all__ = ["copyfileobj", "copyfile", "copymode", "copystat", "copy", "copy2", 26 "copytree", "move", "rmtree", "Error", "SpecialFileError", 27 "ExecError", "make_archive", "get_archive_formats", 28 "register_archive_format", "unregister_archive_format", 29 "ignore_patterns"] 30

31 -class Error(EnvironmentError):

32 pass

33

34 -class SpecialFileError(EnvironmentError):

35 """Raised when trying to do a kind of operation (e.g. copying) which is 36 not supported on a special file (e.g. a named pipe)"""

37

38 -class ExecError(EnvironmentError):

39 """Raised when a command could not be executed"""

40 41 try: 42 WindowsError 43 except NameError: 44 WindowsError = None 45

46 -def copyfileobj(fsrc, fdst, length=16*1024):

47 """copy data from file-like object fsrc to file-like object fdst""" 48 while 1: 49 buf = fsrc.read(length) 50 if not buf: 51 break 52 fdst.write(buf)

53

54 -def _samefile(src, dst):

55 # Macintosh, Unix. 56 if hasattr(os.path, 'samefile'): 57 try: 58 return os.path.samefile(src, dst) 59 except OSError: 60 return False 61 62 # All other platforms: check for same pathname. 63 return (os.path.normcase(os.path.abspath(src)) == 64 os.path.normcase(os.path.abspath(dst)))

65

66 -def copyfile(src, dst):

67 """Copy data from src to dst""" 68 if _samefile(src, dst): 69 raise Error("`%s` and `%s` are the same file" % (src, dst)) 70 71 for fn in [src, dst]: 72 try: 73 st = os.stat(fn) 74 except OSError: 75 # File most likely does not exist 76 pass 77 else: 78 # XXX What about other special files? (sockets, devices...) 79 if stat.S_ISFIFO(st.st_mode): 80 raise SpecialFileError("`%s` is a named pipe" % fn) 81 82 with open(src, 'rb') as fsrc: 83 with open(dst, 'wb') as fdst: 84 copyfileobj(fsrc, fdst)

85

86 -def copymode(src, dst):

87 """Copy mode bits from src to dst""" 88 if hasattr(os, 'chmod'): 89 st = os.stat(src) 90 mode = stat.S_IMODE(st.st_mode) 91 os.chmod(dst, mode)

92

93 -def copystat(src, dst):

94 """Copy all stat info (mode bits, atime, mtime, flags) from src to dst""" 95 st = os.stat(src) 96 mode = stat.S_IMODE(st.st_mode) 97 if hasattr(os, 'utime'): 98 os.utime(dst, (st.st_atime, st.st_mtime)) 99 if hasattr(os, 'chmod'): 100 os.chmod(dst, mode) 101 if hasattr(os, 'chflags') and hasattr(st, 'st_flags'): 102 try: 103 os.chflags(dst, st.st_flags) 104 except OSError, why: 105 if (not hasattr(errno, 'EOPNOTSUPP') or 106 why.errno != errno.EOPNOTSUPP): 107 raise

108

109 -def copy(src, dst):

110 """Copy data and mode bits ("cp src dst"). 111 112 The destination may be a directory. 113 114 """ 115 if os.path.isdir(dst): 116 dst = os.path.join(dst, os.path.basename(src)) 117 copyfile(src, dst) 118 copymode(src, dst)

119

120 -def copy2(src, dst):

121 """Copy data and all stat info ("cp -p src dst"). 122 123 The destination may be a directory. 124 125 """ 126 if os.path.isdir(dst): 127 dst = os.path.join(dst, os.path.basename(src)) 128 copyfile(src, dst) 129 copystat(src, dst)

130

131 -def ignore_patterns(*patterns):

132 """Function that can be used as copytree() ignore parameter. 133 134 Patterns is a sequence of glob-style patterns 135 that are used to exclude files""" 136 def _ignore_patterns(path, names): 137 ignored_names = [] 138 for pattern in patterns: 139 ignored_names.extend(fnmatch.filter(names, pattern)) 140 return set(ignored_names)

141 return _ignore_patterns 142

143 -def copytree(src, dst, symlinks=False, ignore=None):

144 """Recursively copy a directory tree using copy2(). 145 146 The destination directory must not already exist. 147 If exception(s) occur, an Error is raised with a list of reasons. 148 149 If the optional symlinks flag is true, symbolic links in the 150 source tree result in symbolic links in the destination tree; if 151 it is false, the contents of the files pointed to by symbolic 152 links are copied. 153 154 The optional ignore argument is a callable. If given, it 155 is called with the `src` parameter, which is the directory 156 being visited by copytree(), and `names` which is the list of 157 `src` contents, as returned by os.listdir(): 158 159 callable(src, names) -> ignored_names 160 161 Since copytree() is called recursively, the callable will be 162 called once for each directory that is copied. It returns a 163 list of names relative to the `src` directory that should 164 not be copied. 165 166 XXX Consider this example code rather than the ultimate tool. 167 168 """ 169 names = os.listdir(src) 170 if ignore is not None: 171 ignored_names = ignore(src, names) 172 else: 173 ignored_names = set() 174 175 os.makedirs(dst) 176 errors = [] 177 for name in names: 178 if name in ignored_names: 179 continue 180 srcname = os.path.join(src, name) 181 dstname = os.path.join(dst, name) 182 try: 183 if symlinks and os.path.islink(srcname): 184 linkto = os.readlink(srcname) 185 os.symlink(linkto, dstname) 186 elif os.path.isdir(srcname): 187 copytree(srcname, dstname, symlinks, ignore) 188 else: 189 # Will raise a SpecialFileError for unsupported file types 190 copy2(srcname, dstname) 191 # catch the Error from the recursive copytree so that we can 192 # continue with other files 193 except Error, err: 194 errors.extend(err.args[0]) 195 except EnvironmentError, why: 196 errors.append((srcname, dstname, str(why))) 197 try: 198 copystat(src, dst) 199 except OSError, why: 200 if WindowsError is not None and isinstance(why, WindowsError): 201 # Copying file access times may fail on Windows 202 pass 203 else: 204 errors.extend((src, dst, str(why))) 205 if errors: 206 raise Error, errors

207

208 -def rmtree(path, ignore_errors=False, onerror=None):

209 """Recursively delete a directory tree. 210 211 If ignore_errors is set, errors are ignored; otherwise, if onerror 212 is set, it is called to handle the error with arguments (func, 213 path, exc_info) where func is os.listdir, os.remove, or os.rmdir; 214 path is the argument to that function that caused it to fail; and 215 exc_info is a tuple returned by sys.exc_info(). If ignore_errors 216 is false and onerror is None, an exception is raised. 217 218 """ 219 if ignore_errors: 220 def onerror(*args): 221 pass

222 elif onerror is None: 223 def onerror(*args): 224 raise 225 try: 226 if os.path.islink(path): 227 # symlinks to directories are forbidden, see bug #1669 228 raise OSError("Cannot call rmtree on a symbolic link") 229 except OSError: 230 onerror(os.path.islink, path, sys.exc_info()) 231 # can't continue even if onerror hook returns 232 return 233 names = [] 234 try: 235 names = os.listdir(path) 236 except os.error, err: 237 onerror(os.listdir, path, sys.exc_info()) 238 for name in names: 239 fullname = os.path.join(path, name) 240 try: 241 mode = os.lstat(fullname).st_mode 242 except os.error: 243 mode = 0 244 if stat.S_ISDIR(mode): 245 rmtree(fullname, ignore_errors, onerror) 246 else: 247 try: 248 os.remove(fullname) 249 except os.error, err: 250 onerror(os.remove, fullname, sys.exc_info()) 251 try: 252 os.rmdir(path) 253 except os.error: 254 onerror(os.rmdir, path, sys.exc_info()) 255 256

257 -def _basename(path):

258 # A basename() variant which first strips the trailing slash, if present. 259 # Thus we always get the last component of the path, even for directories. 260 return os.path.basename(path.rstrip(os.path.sep))

261

262 -def move(src, dst):

263 """Recursively move a file or directory to another location. This is 264 similar to the Unix "mv" command. 265 266 If the destination is a directory or a symlink to a directory, the source 267 is moved inside the directory. The destination path must not already 268 exist. 269 270 If the destination already exists but is not a directory, it may be 271 overwritten depending on os.rename() semantics. 272 273 If the destination is on our current filesystem, then rename() is used. 274 Otherwise, src is copied to the destination and then removed. 275 A lot more could be done here... A look at a mv.c shows a lot of 276 the issues this implementation glosses over. 277 278 """ 279 real_dst = dst 280 if os.path.isdir(dst): 281 if _samefile(src, dst): 282 # We might be on a case insensitive filesystem, 283 # perform the rename anyway. 284 os.rename(src, dst) 285 return 286 287 real_dst = os.path.join(dst, _basename(src)) 288 if os.path.exists(real_dst): 289 raise Error, "Destination path '%s' already exists" % real_dst 290 try: 291 os.rename(src, real_dst) 292 except OSError: 293 if os.path.isdir(src): 294 if _destinsrc(src, dst): 295 raise Error, "Cannot move a directory '%s' into itself '%s'." % (src, dst) 296 copytree(src, real_dst, symlinks=True) 297 rmtree(src) 298 else: 299 copy2(src, real_dst) 300 os.unlink(src)

301

302 -def _destinsrc(src, dst):

303 src = abspath(src) 304 dst = abspath(dst) 305 if not src.endswith(os.path.sep): 306 src += os.path.sep 307 if not dst.endswith(os.path.sep): 308 dst += os.path.sep 309 return dst.startswith(src)

310

311 -def _get_gid(name):

312 """Returns a gid, given a group name.""" 313 if getgrnam is None or name is None: 314 return None 315 try: 316 result = getgrnam(name) 317 except KeyError: 318 result = None 319 if result is not None: 320 return result[2] 321 return None

322

323 -def _get_uid(name):

324 """Returns an uid, given a user name.""" 325 if getpwnam is None or name is None: 326 return None 327 try: 328 result = getpwnam(name) 329 except KeyError: 330 result = None 331 if result is not None: 332 return result[2] 333 return None

334

335 -def _make_tarball(base_name, base_dir, compress="gzip", verbose=0, dry_run=0, 336 owner=None, group=None, logger=None):

337 """Create a (possibly compressed) tar file from all the files under 338 'base_dir'. 339 340 'compress' must be "gzip" (the default), "bzip2", or None. 341 342 'owner' and 'group' can be used to define an owner and a group for the 343 archive that is being built. If not provided, the current owner and group 344 will be used. 345 346 The output tar file will be named 'base_name' + ".tar", possibly plus 347 the appropriate compression extension (".gz", or ".bz2"). 348 349 Returns the output filename. 350 """ 351 tar_compression = {'gzip': 'gz', 'bzip2': 'bz2', None: ''} 352 compress_ext = {'gzip': '.gz', 'bzip2': '.bz2'} 353 354 # flags for compression program, each element of list will be an argument 355 if compress is not None and compress not in compress_ext.keys(): 356 raise ValueError, \ 357 ("bad value for 'compress': must be None, 'gzip' or 'bzip2'") 358 359 archive_name = base_name + '.tar' + compress_ext.get(compress, '') 360 archive_dir = os.path.dirname(archive_name) 361 362 if not os.path.exists(archive_dir): 363 if logger is not None: 364 logger.info("creating %s", archive_dir) 365 if not dry_run: 366 os.makedirs(archive_dir) 367 368 369 # creating the tarball 370 import tarfile # late import so Python build itself doesn't break 371 372 if logger is not None: 373 logger.info('Creating tar archive') 374 375 uid = _get_uid(owner) 376 gid = _get_gid(group) 377 378 def _set_uid_gid(tarinfo): 379 if gid is not None: 380 tarinfo.gid = gid 381 tarinfo.gname = group 382 if uid is not None: 383 tarinfo.uid = uid 384 tarinfo.uname = owner 385 return tarinfo

386 387 if not dry_run: 388 tar = tarfile.open(archive_name, 'w|%s' % tar_compression[compress]) 389 try: 390 tar.add(base_dir, filter=_set_uid_gid) 391 finally: 392 tar.close() 393 394 return archive_name 395

396 -def _call_external_zip(base_dir, zip_filename, verbose=False, dry_run=False):

397 # XXX see if we want to keep an external call here 398 if verbose: 399 zipoptions = "-r" 400 else: 401 zipoptions = "-rq" 402 from distutils.errors import DistutilsExecError 403 from distutils.spawn import spawn 404 try: 405 spawn(["zip", zipoptions, zip_filename, base_dir], dry_run=dry_run) 406 except DistutilsExecError: 407 # XXX really should distinguish between "couldn't find 408 # external 'zip' command" and "zip failed". 409 raise ExecError, \ 410 ("unable to create zip file '%s': " 411 "could neither import the 'zipfile' module nor " 412 "find a standalone zip utility") % zip_filename

413

414 -def _make_zipfile(base_name, base_dir, verbose=0, dry_run=0, logger=None):

415 """Create a zip file from all the files under 'base_dir'. 416 417 The output zip file will be named 'base_name' + ".zip". Uses either the 418 "zipfile" Python module (if available) or the InfoZIP "zip" utility 419 (if installed and found on the default search path). If neither tool is 420 available, raises ExecError. Returns the name of the output zip 421 file. 422 """ 423 zip_filename = base_name + ".zip" 424 archive_dir = os.path.dirname(base_name) 425 426 if not os.path.exists(archive_dir): 427 if logger is not None: 428 logger.info("creating %s", archive_dir) 429 if not dry_run: 430 os.makedirs(archive_dir) 431 432 # If zipfile module is not available, try spawning an external 'zip' 433 # command. 434 try: 435 import zipfile 436 except ImportError: 437 zipfile = None 438 439 if zipfile is None: 440 _call_external_zip(base_dir, zip_filename, verbose, dry_run) 441 else: 442 if logger is not None: 443 logger.info("creating '%s' and adding '%s' to it", 444 zip_filename, base_dir) 445 446 if not dry_run: 447 zip = zipfile.ZipFile(zip_filename, "w", 448 compression=zipfile.ZIP_DEFLATED) 449 450 for dirpath, dirnames, filenames in os.walk(base_dir): 451 for name in filenames: 452 path = os.path.normpath(os.path.join(dirpath, name)) 453 if os.path.isfile(path): 454 zip.write(path, path) 455 if logger is not None: 456 logger.info("adding '%s'", path) 457 zip.close() 458 459 return zip_filename

460 461 _ARCHIVE_FORMATS = { 462 'gztar': (_make_tarball, [('compress', 'gzip')], "gzip'ed tar-file"), 463 'bztar': (_make_tarball, [('compress', 'bzip2')], "bzip2'ed tar-file"), 464 'tar': (_make_tarball, [('compress', None)], "uncompressed tar file"), 465 'zip': (_make_zipfile, [],"ZIP file") 466 } 467

468 -def get_archive_formats():

469 """Returns a list of supported formats for archiving and unarchiving. 470 471 Each element of the returned sequence is a tuple (name, description) 472 """ 473 formats = [(name, registry[2]) for name, registry in 474 _ARCHIVE_FORMATS.items()] 475 formats.sort() 476 return formats

477

478 -def register_archive_format(name, function, extra_args=None, description=''):

479 """Registers an archive format. 480 481 name is the name of the format. function is the callable that will be 482 used to create archives. If provided, extra_args is a sequence of 483 (name, value) tuples that will be passed as arguments to the callable. 484 description can be provided to describe the format, and will be returned 485 by the get_archive_formats() function. 486 """ 487 if extra_args is None: 488 extra_args = [] 489 if not isinstance(function, collections.Callable): 490 raise TypeError('The %s object is not callable' % function) 491 if not isinstance(extra_args, (tuple, list)): 492 raise TypeError('extra_args needs to be a sequence') 493 for element in extra_args: 494 if not isinstance(element, (tuple, list)) or len(element) !=2 : 495 raise TypeError('extra_args elements are : (arg_name, value)') 496 497 _ARCHIVE_FORMATS[name] = (function, extra_args, description)

498

499 -def unregister_archive_format(name):

500 del _ARCHIVE_FORMATS[name]

501

502 -def make_archive(base_name, format, root_dir=None, base_dir=None, verbose=0, 503 dry_run=0, owner=None, group=None, logger=None):

504 """Create an archive file (eg. zip or tar). 505 506 'base_name' is the name of the file to create, minus any format-specific 507 extension; 'format' is the archive format: one of "zip", "tar", "bztar" 508 or "gztar". 509 510 'root_dir' is a directory that will be the root directory of the 511 archive; ie. we typically chdir into 'root_dir' before creating the 512 archive. 'base_dir' is the directory where we start archiving from; 513 ie. 'base_dir' will be the common prefix of all files and 514 directories in the archive. 'root_dir' and 'base_dir' both default 515 to the current directory. Returns the name of the archive file. 516 517 'owner' and 'group' are used when creating a tar archive. By default, 518 uses the current owner and group. 519 """ 520 save_cwd = os.getcwd() 521 if root_dir is not None: 522 if logger is not None: 523 logger.debug("changing into '%s'", root_dir) 524 base_name = os.path.abspath(base_name) 525 if not dry_run: 526 os.chdir(root_dir) 527 528 if base_dir is None: 529 base_dir = os.curdir 530 531 kwargs = {'dry_run': dry_run, 'logger': logger} 532 533 try: 534 format_info = _ARCHIVE_FORMATS[format] 535 except KeyError: 536 raise ValueError, "unknown archive format '%s'" % format 537 538 func = format_info[0] 539 for arg, val in format_info[1]: 540 kwargs[arg] = val 541 542 if format != 'zip': 543 kwargs['owner'] = owner 544 kwargs['group'] = group 545 546 try: 547 filename = func(base_name, base_dir, **kwargs) 548 finally: 549 if root_dir is not None: 550 if logger is not None: 551 logger.debug("changing back to '%s'", save_cwd) 552 os.chdir(save_cwd) 553 554 return filename

555

Source Code for Module shutil