shutil.py

Go to the documentation of this file.

##
# Utility functions for copying files and directory trees on remote servers via sshUtils.
# 
# XXX The functions here don't copy the resource fork or other metadata on Mac.
# 
# 

import sshUtils as os
import sys
import stat
from os.path import abspath

__all__ = ["copyfileobj","copyfile","copymode","copystat","copy","copy2",
           "copytree","move","rmtree","Error"]

class Error(EnvironmentError):
    pass

##
# copy data from file-like object fsrc to file-like object fdst
def copyfileobj(fsrc, fdst, length=16*1024):
    while 1:
        buf = fsrc.read(length)
        if not buf:
            break
        fdst.write(buf)

def _samefile(src, dst):
    # Macintosh, Unix.
    if hasattr(os.path,'samefile'):
        try:
            return os.path.samefile(src, dst)
        except OSError:
            return False

    # All other platforms: check for same pathname.
    return (os.path.normcase(os.path.abspath(src)) ==
            os.path.normcase(os.path.abspath(dst)))

##
# Copy data from src to dst
def copyfile(src, dst):
    if _samefile(src, dst):
        raise Error, "`%s` and `%s` are the same file" % (src, dst)

    fsrc = None
    fdst = None
    try:
        fsrc = open(src, 'rb')
        fdst = open(dst, 'wb')
        copyfileobj(fsrc, fdst)
    finally:
        if fdst:
            fdst.close()
        if fsrc:
            fsrc.close()

##
# Copy mode bits from src to dst
def copymode(src, dst):
    if hasattr(os, 'chmod'):
        st = os.stat(src)
        mode = stat.S_IMODE(st.st_mode)
        os.chmod(dst, mode)

##
# Copy all stat info (mode bits, atime and mtime) from src to dst
def copystat(src, dst):
    st = os.stat(src)
    mode = stat.S_IMODE(st.st_mode)
    if hasattr(os, 'utime'):
        os.utime(dst, (st.st_atime, st.st_mtime))
    if hasattr(os, 'chmod'):
        os.chmod(dst, mode)


##
# Copy data and mode bits ("cp src dst").
# 
#     The destination may be a directory.
# 
#     
def copy(src, dst):
    if os.path.isdir(dst):
        dst = os.path.join(dst, os.path.basename(src))
    copyfile(src, dst)
    copymode(src, dst)

##
# Copy data and all stat info ("cp -p src dst").
# 
#     The destination may be a directory.
# 
#     
def copy2(src, dst):
    if os.path.isdir(dst):
        dst = os.path.join(dst, os.path.basename(src))
    copyfile(src, dst)
    copystat(src, dst)


##
# Recursively copy a directory tree using copy2().
# 
#     The destination directory must not already exist.
#     If exception(s) occur, an Error is raised with a list of reasons.
# 
#     If the optional symlinks flag is true, symbolic links in the
#     source tree result in symbolic links in the destination tree; if
#     it is false, the contents of the files pointed to by symbolic
#     links are copied.
# 
#     XXX Consider this example code rather than the ultimate tool.
# 
#     
def copytree(src, dst, symlinks=False):
    names = os.listdir(src)
    os.makedirs(dst)
    errors = []
    for name in names:
        srcname = os.path.join(src, name)
        dstname = os.path.join(dst, name)
        try:
            if symlinks and os.path.islink(srcname):
                linkto = os.readlink(srcname)
                os.symlink(linkto, dstname)
            elif os.path.isdir(srcname):
                copytree(srcname, dstname, symlinks)
            else:
                copy2(srcname, dstname)
            # XXX What about devices, sockets etc.?
        except (IOError, os.error), why:
            errors.append((srcname, dstname, str(why)))
        # catch the Error from the recursive copytree so that we can
        # continue with other files
        except Error, err:
            errors.extend(err.args[0])
    try:
        copystat(src, dst)
    except WindowsError:
        # can't copy file access times on Windows
        pass
    except OSError, why:
        errors.extend((src, dst, str(why)))
    if errors:
        raise Error, errors

##
# Recursively delete a directory tree.
# 
#     If ignore_errors is set, errors are ignored; otherwise, if onerror
#     is set, it is called to handle the error with arguments (func,
#     path, exc_info) where func is os.listdir, os.remove, or os.rmdir;
#     path is the argument to that function that caused it to fail; and
#     exc_info is a tuple returned by sys.exc_info().  If ignore_errors
#     is false and onerror is None, an exception is raised.
# 
#     
def rmtree(path, ignore_errors=False, onerror=None):
    if ignore_errors:
        def onerror(*args):
            pass
    elif onerror is None:
        def onerror(*args):
            raise
    names = []
    try:
        names = os.listdir(path)
    except os.error, err:
        onerror(os.listdir, path, sys.exc_info())
    for name in names:
        fullname = os.path.join(path, name)
        try:
            mode = os.lstat(fullname).st_mode
        except os.error:
            mode = 0
        if stat.S_ISDIR(mode):
            rmtree(fullname, ignore_errors, onerror)
        else:
            try:
                os.remove(fullname)
            except os.error, err:
                onerror(os.remove, fullname, sys.exc_info())
    try:
        os.rmdir(path)
    except os.error:
        onerror(os.rmdir, path, sys.exc_info())

##
# Recursively move a file or directory to another location.
# 
#     If the destination is on our current filesystem, then simply use
#     rename.  Otherwise, copy src to the dst and then remove src.
#     A lot more could be done here...  A look at a mv.c shows a lot of
#     the issues this implementation glosses over.
# 
#     
def move(src, dst):

    try:
        os.rename(src, dst)
    except OSError:
        if os.path.isdir(src):
            if destinsrc(src, dst):
                raise Error, "Cannot move a directory '%s' into itself '%s'." % (src, dst)
            copytree(src, dst, symlinks=True)
            rmtree(src)
        else:
            copy2(src,dst)
            os.unlink(src)

def destinsrc(src, dst):
    return abspath(dst).startswith(abspath(src))