"copytree", "move", "rmtree", "Error", "SpecialFileError",

"make_archive", "get_archive_formats",

"register_archive_format", "unregister_archive_format",

"get_unpack_formats", "register_unpack_format",

"unregister_unpack_format", "unpack_archive",

"ignore_patterns", "chown", "which", "get_terminal_size",

"""Copy a regular file content or metadata by using high-performance

try:

infd = fsrc.fileno()

outfd = fdst.fileno()

except Exception as err:

raise _GiveupOnFastCopy(err) # not a regular file

try:

posix._fcopyfile(infd, outfd, flags)

except OSError as err:

err.filename = fsrc.name

err.filename2 = fdst.name

if err.errno in {errno.EINVAL, errno.ENOTSUP}:

raise _GiveupOnFastCopy(err)

else:

"""Determine blocksize for fastcopying on Linux.

try:

blocksize = max(os.fstat(infd).st_size, 2 ** 23) # min 8 MiB

except OSError:

blocksize = 2 ** 27 # 128 MiB

# On 32-bit architectures truncate to 1 GiB to avoid OverflowError,

# see gh-82500.

if sys.maxsize < 2 ** 32:

blocksize = min(blocksize, 2 ** 30)

"""Copy data from one regular mmap-like fd to another by using

try:

infd = fsrc.fileno()

outfd = fdst.fileno()

except Exception as err:

raise _GiveupOnFastCopy(err) # not a regular file

blocksize = _determine_linux_fastcopy_blocksize(infd)

offset = 0

while True:

try:

n_copied = os.copy_file_range(infd, outfd, blocksize, offset_dst=offset)

except OSError as err:

# ...in oder to have a more informative exception.

err.filename = fsrc.name

err.filename2 = fdst.name

if err.errno == errno.ENOSPC: # filesystem is full

raise err from None

# Give up on first call and if no data was copied.

if offset == 0 and os.lseek(outfd, 0, os.SEEK_CUR) == 0:

raise _GiveupOnFastCopy(err)

raise err

else:

if n_copied == 0:

# If no bytes have been copied yet, copy_file_range

# might silently fail.

# https://lore.kernel.org/linux-fsdevel/20210126233840.GG4626@dread.disaster.area/T/#m05753578c7f7882f6e9ffe01f981bc223edef2b0

if offset == 0:

raise _GiveupOnFastCopy()

break

"""Copy data from one regular mmap-like fd to another by using

# Note: copyfileobj() is left alone in order to not introduce any

# unexpected breakage. Possible risks by using zero-copy calls

# in copyfileobj() are:

# - fdst cannot be open in "a"(ppend) mode

# - fsrc and fdst may be open in "t"(ext) mode

# - fsrc may be a BufferedReader (which hides unread data in a buffer),

# GzipFile (which decompresses data), HTTPResponse (which decodes

# chunks).

# - possibly others (e.g. encrypted fs/partition?)

global _USE_CP_SENDFILE

try:

infd = fsrc.fileno()

outfd = fdst.fileno()

except Exception as err:

raise _GiveupOnFastCopy(err) # not a regular file

blocksize = _determine_linux_fastcopy_blocksize(infd)

offset = 0

while True:

try:

sent = os.sendfile(outfd, infd, offset, blocksize)

except OSError as err:

# ...in order to have a more informative exception.

err.filename = fsrc.name

err.filename2 = fdst.name

if err.errno == errno.ENOTSOCK:

# sendfile() on this platform (probably Linux < 2.6.33)

# does not support copies between regular files (only

# sockets).

_USE_CP_SENDFILE = False

raise _GiveupOnFastCopy(err)

if err.errno == errno.ENOSPC: # filesystem is full

raise err from None

# Give up on first call and if no data was copied.

if offset == 0 and os.lseek(outfd, 0, os.SEEK_CUR) == 0:

raise _GiveupOnFastCopy(err)

raise err

else:

if sent == 0:

break # EOF

"""readinto()/memoryview() based variant of copyfileobj().

# Localize variable access to minimize overhead.

fsrc_readinto = fsrc.readinto

fdst_write = fdst.write

with memoryview(bytearray(length)) as mv:

while True:

n = fsrc_readinto(mv)

if not n:

break

elif n < length:

with mv[:n] as smv:

fdst_write(smv)

break

else:

"""copy data from file-like object fsrc to file-like object fdst"""

if not length:

length = COPY_BUFSIZE

# Localize variable access to minimize overhead.

fsrc_read = fsrc.read

fdst_write = fdst.write

while buf := fsrc_read(length):

# Macintosh, Unix.

if isinstance(src, os.DirEntry) and hasattr(os.path, 'samestat'):

try:

return os.path.samestat(src.stat(), os.stat(dst))

except OSError:

return False

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)) ==

"""Copy data from src to dst in the most efficient way possible.

sys.audit("shutil.copyfile", src, dst)

if _samefile(src, dst):

raise SameFileError("{!r} and {!r} are the same file".format(src, dst))

file_size = 0

for i, fn in enumerate([src, dst]):

try:

st = _stat(fn)

except OSError:

# File most likely does not exist

pass

else:

# XXX What about other special files? (sockets, devices...)

if stat.S_ISFIFO(st.st_mode):

fn = fn.path if isinstance(fn, os.DirEntry) else fn

raise SpecialFileError("`%s` is a named pipe" % fn)

if _WINDOWS and i == 0:

file_size = st.st_size

if not follow_symlinks and _islink(src):

os.symlink(os.readlink(src), dst)

else:

with open(src, 'rb') as fsrc:

try:

with open(dst, 'wb') as fdst:

# macOS

if _HAS_FCOPYFILE:

try:

_fastcopy_fcopyfile(fsrc, fdst, posix._COPYFILE_DATA)

return dst

except _GiveupOnFastCopy:

pass

# Linux / Android / Solaris

elif _USE_CP_SENDFILE or _USE_CP_COPY_FILE_RANGE:

# reflink may be implicit in copy_file_range.

if _USE_CP_COPY_FILE_RANGE:

try:

_fastcopy_copy_file_range(fsrc, fdst)

return dst

except _GiveupOnFastCopy:

pass

if _USE_CP_SENDFILE:

try:

_fastcopy_sendfile(fsrc, fdst)

return dst

except _GiveupOnFastCopy:

pass

# Windows, see:

# https://github.com/python/cpython/pull/7160#discussion_r195405230

elif _WINDOWS and file_size > 0:

_copyfileobj_readinto(fsrc, fdst, min(file_size, COPY_BUFSIZE))

return dst

copyfileobj(fsrc, fdst)

# Issue 43219, raise a less confusing exception

except IsADirectoryError as e:

if not os.path.exists(dst):

raise FileNotFoundError(f'Directory does not exist: {dst}') from e

else:

raise

"""Copy mode bits from src to dst.

sys.audit("shutil.copymode", src, dst)

if not follow_symlinks and _islink(src) and os.path.islink(dst):

if hasattr(os, 'lchmod'):

stat_func, chmod_func = os.lstat, os.lchmod

else:

return

else:

stat_func = _stat

if os.name == 'nt' and os.path.islink(dst):

def chmod_func(*args):

os.chmod(*args, follow_symlinks=True)

else:

chmod_func = os.chmod

st = stat_func(src)

"""Copy extended filesystem attributes from `src` to `dst`.

try:

names = os.listxattr(src, follow_symlinks=follow_symlinks)

except OSError as e:

if e.errno not in (errno.ENOTSUP, errno.ENODATA, errno.EINVAL):

raise

return

for name in names:

try:

value = os.getxattr(src, name, follow_symlinks=follow_symlinks)

os.setxattr(dst, name, value, follow_symlinks=follow_symlinks)

except OSError as e:

if e.errno not in (errno.EPERM, errno.ENOTSUP, errno.ENODATA,

errno.EINVAL, errno.EACCES):

"""Copy file metadata

sys.audit("shutil.copystat", src, dst)

def _nop(*args, ns=None, follow_symlinks=None):

pass

# follow symlinks (aka don't not follow symlinks)

follow = follow_symlinks or not (_islink(src) and os.path.islink(dst))

if follow:

# use the real function if it exists

def lookup(name):

return getattr(os, name, _nop)

else:

# use the real function only if it exists

# *and* it supports follow_symlinks

def lookup(name):

fn = getattr(os, name, _nop)

if fn in os.supports_follow_symlinks:

return fn

return _nop

if isinstance(src, os.DirEntry):

st = src.stat(follow_symlinks=follow)

else:

st = lookup("stat")(src, follow_symlinks=follow)

mode = stat.S_IMODE(st.st_mode)

lookup("utime")(dst, ns=(st.st_atime_ns, st.st_mtime_ns),

follow_symlinks=follow)

# We must copy extended attributes before the file is (potentially)

# chmod()'ed read-only, otherwise setxattr() will error with -EACCES.

_copyxattr(src, dst, follow_symlinks=follow)

try:

lookup("chmod")(dst, mode, follow_symlinks=follow)

except NotImplementedError:

# if we got a NotImplementedError, it's because

# * follow_symlinks=False,

# * lchown() is unavailable, and

# * either

# * fchownat() is unavailable or

# * fchownat() doesn't implement AT_SYMLINK_NOFOLLOW.

# (it returned ENOSUP.)

# therefore we're out of options--we simply cannot chown the

# symlink. give up, suppress the error.

# (which is what shutil always did in this circumstance.)

pass

if hasattr(st, 'st_flags'):

try:

lookup("chflags")(dst, st.st_flags, follow_symlinks=follow)

except OSError as why:

for err in 'EOPNOTSUPP', 'ENOTSUP':

if hasattr(errno, err) and why.errno == getattr(errno, err):

break

else:

"""Copy data and mode bits ("cp src dst"). Return the file's destination.

if os.path.isdir(dst):

dst = os.path.join(dst, os.path.basename(src))

copyfile(src, dst, follow_symlinks=follow_symlinks)

copymode(src, dst, follow_symlinks=follow_symlinks)

"""Copy data and metadata. Return the file's destination.

if os.path.isdir(dst):

dst = os.path.join(dst, os.path.basename(src))

if hasattr(_winapi, "CopyFile2"):

src_ = os.fsdecode(src)

dst_ = os.fsdecode(dst)

flags = _winapi.COPY_FILE_ALLOW_DECRYPTED_DESTINATION # for compat

if not follow_symlinks:

flags |= _winapi.COPY_FILE_COPY_SYMLINK

try:

_winapi.CopyFile2(src_, dst_, flags)

return dst

except OSError as exc:

if (exc.winerror == _winapi.ERROR_PRIVILEGE_NOT_HELD

and not follow_symlinks):

# Likely encountered a symlink we aren't allowed to create.

# Fall back on the old code

pass

elif exc.winerror == _winapi.ERROR_ACCESS_DENIED:

# Possibly encountered a hidden or readonly file we can't

# overwrite. Fall back on old code

pass

else:

raise

copyfile(src, dst, follow_symlinks=follow_symlinks)

copystat(src, dst, follow_symlinks=follow_symlinks)

"""Function that can be used as copytree() ignore parameter.

def _ignore_patterns(path, names):

ignored_names = []

for pattern in patterns:

ignored_names.extend(fnmatch.filter(names, pattern))

return set(ignored_names)

ignore_dangling_symlinks, dirs_exist_ok=False):

if ignore is not None:

ignored_names = ignore(os.fspath(src), [x.name for x in entries])

else:

ignored_names = ()

os.makedirs(dst, exist_ok=dirs_exist_ok)

errors = []

use_srcentry = copy_function is copy2 or copy_function is copy

for srcentry in entries:

if srcentry.name in ignored_names:

continue

srcname = os.path.join(src, srcentry.name)

dstname = os.path.join(dst, srcentry.name)

srcobj = srcentry if use_srcentry else srcname

try:

is_symlink = srcentry.is_symlink()

if is_symlink and os.name == 'nt':

# Special check for directory junctions, which appear as

# symlinks but we want to recurse.

lstat = srcentry.stat(follow_symlinks=False)

if lstat.st_reparse_tag == stat.IO_REPARSE_TAG_MOUNT_POINT:

is_symlink = False

if is_symlink:

linkto = os.readlink(srcname)

if symlinks:

# We can't just leave it to `copy_function` because legacy

# code with a custom `copy_function` may rely on copytree

# doing the right thing.

os.symlink(linkto, dstname)

copystat(srcobj, dstname, follow_symlinks=not symlinks)

else:

# ignore dangling symlink if the flag is on

if not os.path.exists(linkto) and ignore_dangling_symlinks:

continue

# otherwise let the copy occur. copy2 will raise an error

if srcentry.is_dir():

copytree(srcobj, dstname, symlinks, ignore,

copy_function, ignore_dangling_symlinks,

dirs_exist_ok)

else:

copy_function(srcobj, dstname)

elif srcentry.is_dir():

copytree(srcobj, dstname, symlinks, ignore, copy_function,

ignore_dangling_symlinks, dirs_exist_ok)

else:

# Will raise a SpecialFileError for unsupported file types

copy_function(srcobj, dstname)

# catch the Error from the recursive copytree so that we can

# continue with other files

except Error as err:

errors.extend(err.args[0])

except OSError as why:

errors.append((srcname, dstname, str(why)))

try:

copystat(src, dst)

except OSError as why:

# Copying file access times may fail on Windows

if getattr(why, 'winerror', None) is None:

errors.append((src, dst, str(why)))

if errors:

raise Error(errors)

ignore_dangling_symlinks=False, dirs_exist_ok=False):

"""Recursively copy a directory tree and return the destination directory.

sys.audit("shutil.copytree", src, dst)

with os.scandir(src) as itr:

entries = list(itr)

return _copytree(entries=entries, src=src, dst=dst, symlinks=symlinks,

ignore=ignore, copy_function=copy_function,

ignore_dangling_symlinks=ignore_dangling_symlinks,

if dir_fd is not None:

raise NotImplementedError("dir_fd unavailable on this platform")

try:

st = os.lstat(path)

except OSError as err:

onexc(os.lstat, path, err)

return

try:

if _rmtree_islink(st):

# symlinks to directories are forbidden, see bug #1669

raise OSError("Cannot call rmtree on a symbolic link")

except OSError as err:

onexc(os.path.islink, path, err)

# can't continue even if onexc hook returns

return

def onerror(err):

if not isinstance(err, FileNotFoundError):

onexc(os.scandir, err.filename, err)

results = os.walk(path, topdown=False, onerror=onerror, followlinks=os._walk_symlinks_as_files)

for dirpath, dirnames, filenames in results:

for name in dirnames:

fullname = os.path.join(dirpath, name)

try:

os.rmdir(fullname)

except FileNotFoundError:

continue

except OSError as err:

onexc(os.rmdir, fullname, err)

for name in filenames:

fullname = os.path.join(dirpath, name)

try:

os.unlink(fullname)

except FileNotFoundError:

continue

except OSError as err:

onexc(os.unlink, fullname, err)

try:

os.rmdir(path)

except FileNotFoundError:

pass

except OSError as err:

# While the unsafe rmtree works fine on bytes, the fd based does not.

if isinstance(path, bytes):

path = os.fsdecode(path)

stack = [(os.lstat, dir_fd, path, None)]

try:

while stack:

_rmtree_safe_fd_step(stack, onexc)

finally:

# Close any file descriptors still on the stack.

while stack:

func, fd, path, entry = stack.pop()

if func is not os.close:

continue

try:

os.close(fd)

except OSError as err:

# Each stack item has four elements:

# * func: The first operation to perform: os.lstat, os.close or os.rmdir.

# Walking a directory starts with an os.lstat() to detect symlinks; in

# this case, func is updated before subsequent operations and passed to

# onexc() if an error occurs.

# * dirfd: Open file descriptor, or None if we're processing the top-level

# directory given to rmtree() and the user didn't supply dir_fd.

# * path: Path of file to operate upon. This is passed to onexc() if an

# error occurs.

# * orig_entry: os.DirEntry, or None if we're processing the top-level

# directory given to rmtree(). We used the cached stat() of the entry to

# save a call to os.lstat() when walking subdirectories.

func, dirfd, path, orig_entry = stack.pop()

name = path if orig_entry is None else orig_entry.name

parent_fd = None if func is os.close else dirfd

try:

if func is os.close:

os.close(dirfd)

return

if func is os.rmdir:

os.rmdir(name, dir_fd=dirfd)

return

# Note: To guard against symlink races, we use the standard

# lstat()/open()/fstat() trick.

assert func is os.lstat

if orig_entry is None:

orig_st = os.lstat(name, dir_fd=dirfd)

else:

orig_st = orig_entry.stat(follow_symlinks=False)

func = os.open # For error reporting.

topfd = os.open(name, os.O_RDONLY | os.O_NONBLOCK, dir_fd=dirfd)

func = os.path.islink # For error reporting.

try:

if not os.path.samestat(orig_st, os.fstat(topfd)):

# Symlinks to directories are forbidden, see GH-46010.

raise OSError("Cannot call rmtree on a symbolic link")

stack.append((os.rmdir, dirfd, path, orig_entry))

finally:

stack.append((os.close, topfd, path, orig_entry))

func = os.scandir # For error reporting.

with os.scandir(topfd) as scandir_it:

entries = list(scandir_it)

for entry in entries:

fullname = os.path.join(path, entry.name)

try:

if entry.is_dir(follow_symlinks=False):

# Traverse into sub-directory.

stack.append((os.lstat, topfd, fullname, entry))

continue

except FileNotFoundError:

continue

except OSError:

pass

try:

os.unlink(entry.name, dir_fd=topfd)

except FileNotFoundError:

continue

except OSError as err:

onexc(os.unlink, fullname, err, direntry=entry, dir_fd=topfd)

except FileNotFoundError as err:

if orig_entry is None or func is os.close:

err.filename = path

onexc(func, path, err, direntry=orig_entry, dir_fd=parent_fd)

except OSError as err:

err.filename = path

_onexc_kwargs=False):

"""Recursively delete a directory tree.

sys.audit("shutil.rmtree", path, dir_fd)

if ignore_errors:

def onexc(*args, **kwargs):

pass

elif onerror is None and onexc is None:

def onexc(*args, **kwargs):

raise

elif onexc is None:

if onerror is None:

def onexc(*args, **kwargs):

raise

else:

# delegate to onerror

def onexc(*args, **kwargs):

func, path, exc = args

if exc is None:

exc_info = None, None, None

else:

exc_info = type(exc), exc, exc.__traceback__

return onerror(func, path, exc_info)

elif not _onexc_kwargs:

# Only the internal caller in tempfile asks for the extra arguments.

_onexc = onexc

def onexc(func, path, err, **kwargs):

return _onexc(func, path, err)

"""A basename() variant which first strips the trailing slash, if present.

path = os.fspath(path)

sep = os.path.sep + (os.path.altsep or '')

"""Recursively move a file or directory to another location. This is

sys.audit("shutil.move", src, dst)

real_dst = dst

if os.path.isdir(dst):

if _samefile(src, dst) and not os.path.islink(src):

# We might be on a case insensitive filesystem,

# perform the rename anyway.

os.rename(src, dst)

return

# Using _basename instead of os.path.basename is important, as we must

# ignore any trailing slash to avoid the basename returning ''

real_dst = os.path.join(dst, _basename(src))

if os.path.exists(real_dst):

raise Error("Destination path '%s' already exists" % real_dst)

try:

os.rename(src, real_dst)

except OSError:

if os.path.islink(src):

linkto = os.readlink(src)

os.symlink(linkto, real_dst)

os.unlink(src)

elif os.path.isdir(src):

if _destinsrc(src, dst):

raise Error("Cannot move a directory '%s' into itself"

" '%s'." % (src, dst))

if (_is_immutable(src)

or (not os.access(src, os.W_OK) and os.listdir(src)

and sys.platform == 'darwin')):

raise PermissionError("Cannot move the non-empty directory "

"'%s': Lacking write permission to '%s'."

% (src, src))

copytree(src, real_dst, copy_function=copy_function,

symlinks=True)

rmtree(src)

else:

copy_function(src, real_dst)

os.unlink(src)

src = os.path.realpath(src)

dst = os.path.realpath(dst)

if not src.endswith(os.path.sep):

src += os.path.sep

if not dst.endswith(os.path.sep):

dst += os.path.sep

"""Returns a gid, given a group name."""

if name is None:

return None

try:

from grp import getgrnam

except ImportError:

return None

try:

result = getgrnam(name)

except KeyError:

result = None

if result is not None:

return result[2]

"""Returns an uid, given a user name."""

if name is None:

return None

try:

from pwd import getpwnam

except ImportError:

return None

try:

result = getpwnam(name)

except KeyError:

result = None

if result is not None:

return result[2]