def _copy_dist_from_dir(link_path, location):
"""Copy distribution files in `link_path` to `location`.
Invoked when user requests to install a local directory. E.g.:
pip install .
pip install ~/dev/git-repos/python-prompt-toolkit
"""
# Note: This is currently VERY SLOW if you have a lot of data in the
# directory, because it copies everything with `shutil.copytree`.
# What it should really do is build an sdist and install that.
# See https://github.com/pypa/pip/issues/2195
if os.path.isdir(location):
rmtree(location)
# build an sdist
setup_py = 'setup.py'
sdist_args = [sys.executable]
sdist_args.append('-c')
sdist_args.append(SETUPTOOLS_SHIM % setup_py)
sdist_args.append('sdist')
sdist_args += ['--dist-dir', location]
logger.info('Running setup.py sdist for %s', link_path)
with indent_log():
call_subprocess(sdist_args, cwd=link_path, show_stdout=False)
# unpack sdist into `location`
sdist = os.path.join(location, os.listdir(location)[0])
logger.info('Unpacking sdist %s into %s', sdist, location)
unpack_file(sdist, location, content_type=None, link=None)
python类copytree()的实例源码
def copy_files(src, dst, symlinks=False, ignore=None):
"""Copy files from src to dst."""
for item in os.listdir(src):
s = os.path.join(src, item)
d = os.path.join(dst, item)
if os.path.isdir(s):
shutil.copytree(s, d, symlinks, ignore)
else:
shutil.copy2(s, d)
def copy_files(src, dst, symlinks=False, ignore=None):
"""Copy files from src to dst."""
for item in os.listdir(src):
s = os.path.join(src, item)
d = os.path.join(dst, item)
if os.path.isdir(s):
shutil.copytree(s, d, symlinks, ignore)
else:
shutil.copy2(s, d)
def security_list_copy():
old_dir = security_list.SECURITY_LISTS_DIR
new_dir = tempfile.mkdtemp()
try:
for subdir in os.listdir(old_dir):
shutil.copytree(os.path.join(old_dir, subdir),
os.path.join(new_dir, subdir))
with patch.object(security_list, 'SECURITY_LISTS_DIR', new_dir), \
patch.object(security_list, 'using_copy', True,
create=True):
yield
finally:
shutil.rmtree(new_dir, True)
def set_cache(self, url):
up = urlparse(formaturl(url, 'https'))
if self.cache and up and up.netloc and os.path.isdir(self.path):
try:
cpath = os.path.join(self.cache, up.netloc, re.sub(r'^/', '', up.path))
if not os.path.isdir(cpath):
os.makedirs(cpath)
scm_dir = '.'+self.scm.name
if os.path.isdir(os.path.join(cpath, scm_dir)):
rmtree_readonly(os.path.join(cpath, scm_dir))
shutil.copytree(os.path.join(self.path, scm_dir), os.path.join(cpath, scm_dir))
except Exception:
warning("Unable to cache \"%s\" to \"%s\"" % (self.path, cpath))
return False
def copy(src, dst):
shutil.copytree(src, dst)
# Test specific utils
def run(self):
dest_dir = os.path.join(self.dest, self.app_name)
try:
shutil.copytree(self.source, dest_dir)
except FileExistsError:
print(
'Target directory exists: {dest}, skipping!'.format(
dest=self.dest),
file=sys.stderr)
def copySource(self, localPath):
shutil.copytree(localPath, self.getSourceDir())
##
# @brief Get the source directory path inside of the container.
#
# @return
def _unbundle_iso(sr_path, filename, path):
logging.debug("Unbundling ISO '%s'" % filename)
read_only_path = utils.make_staging_area(sr_path)
try:
utils.run_command(['mount', '-o', 'loop', filename, read_only_path])
try:
shutil.copytree(read_only_path, path)
finally:
utils.run_command(['umount', read_only_path])
finally:
utils.cleanup_staging_area(read_only_path)
def inject(session, sr_path, vdi_uuid, boot_menu_url, ip_address, netmask,
gateway, dns, mkisofs_cmd):
iso_filename = '%s.img' % os.path.join(sr_path, 'iso', vdi_uuid)
# Create staging area so we have a unique path but remove it since
# shutil.copytree will recreate it
staging_path = utils.make_staging_area(sr_path)
utils.cleanup_staging_area(staging_path)
try:
_unbundle_iso(sr_path, iso_filename, staging_path)
# Write Configs
_write_file(os.path.join(staging_path, 'netcfg.ipxe'),
NETCFG_IPXE % {"ip_address": ip_address,
"netmask": netmask,
"gateway": gateway,
"dns": dns,
"boot_menu_url": boot_menu_url})
_write_file(os.path.join(staging_path, 'isolinux.cfg'),
ISOLINUX_CFG)
_create_iso(mkisofs_cmd, iso_filename, staging_path)
finally:
utils.cleanup_staging_area(staging_path)
def run(self):
version = self.distribution.get_version()
pkgName = self.distribution.get_name()
debName = "python-" + pkgName
debDir = self.debDir
assert os.getcwd() == self.cwd, 'Must be in package root: %s' % self.cwd
if os.path.isdir(debDir):
raise Exception('DEB build dir already exists: "%s"' % debDir)
sdist = "dist/%s-%s.tar.gz" % (pkgName, version)
if not os.path.isfile(sdist):
raise Exception("No source distribution; run `setup.py sdist` first.")
# copy sdist to build directory and extract
os.mkdir(debDir)
renamedSdist = '%s_%s.orig.tar.gz' % (debName, version)
print("copy %s => %s" % (sdist, os.path.join(debDir, renamedSdist)))
shutil.copy(sdist, os.path.join(debDir, renamedSdist))
print("cd %s; tar -xzf %s" % (debDir, renamedSdist))
if os.system("cd %s; tar -xzf %s" % (debDir, renamedSdist)) != 0:
raise Exception("Error extracting source distribution.")
buildDir = '%s/%s-%s' % (debDir, pkgName, version)
# copy debian control structure
print("copytree %s => %s" % (self.debTemplate, buildDir+'/debian'))
shutil.copytree(self.debTemplate, buildDir+'/debian')
# Write new changelog
chlog = generateDebianChangelog(pkgName, 'CHANGELOG', version, self.maintainer)
print("write changelog %s" % buildDir+'/debian/changelog')
open(buildDir+'/debian/changelog', 'w').write(chlog)
# build package
print('cd %s; debuild -us -uc' % buildDir)
if os.system('cd %s; debuild -us -uc' % buildDir) != 0:
raise Exception("Error during debuild.")
def run(self):
version = self.distribution.get_version()
pkgName = self.distribution.get_name()
debName = "python-" + pkgName
debDir = self.debDir
assert os.getcwd() == self.cwd, 'Must be in package root: %s' % self.cwd
if os.path.isdir(debDir):
raise Exception('DEB build dir already exists: "%s"' % debDir)
sdist = "dist/%s-%s.tar.gz" % (pkgName, version)
if not os.path.isfile(sdist):
raise Exception("No source distribution; run `setup.py sdist` first.")
# copy sdist to build directory and extract
os.mkdir(debDir)
renamedSdist = '%s_%s.orig.tar.gz' % (debName, version)
print("copy %s => %s" % (sdist, os.path.join(debDir, renamedSdist)))
shutil.copy(sdist, os.path.join(debDir, renamedSdist))
print("cd %s; tar -xzf %s" % (debDir, renamedSdist))
if os.system("cd %s; tar -xzf %s" % (debDir, renamedSdist)) != 0:
raise Exception("Error extracting source distribution.")
buildDir = '%s/%s-%s' % (debDir, pkgName, version)
# copy debian control structure
print("copytree %s => %s" % (self.debTemplate, buildDir+'/debian'))
shutil.copytree(self.debTemplate, buildDir+'/debian')
# Write new changelog
chlog = generateDebianChangelog(pkgName, 'CHANGELOG', version, self.maintainer)
print("write changelog %s" % buildDir+'/debian/changelog')
open(buildDir+'/debian/changelog', 'w').write(chlog)
# build package
print('cd %s; debuild -us -uc' % buildDir)
if os.system('cd %s; debuild -us -uc' % buildDir) != 0:
raise Exception("Error during debuild.")
def copy_to_build_dir(self, req_to_install):
target_dir = req_to_install.editable and self.src_dir or self.build_dir
logger.info("Copying %s to %s" % (req_to_install.name, target_dir))
dest = os.path.join(target_dir, req_to_install.name)
shutil.copytree(req_to_install.source_dir, dest)
call_subprocess(["python", "%s/setup.py" % dest, "clean"], cwd=dest,
command_desc='python setup.py clean')
def install_egg(self, egg_path, tmpdir):
destination = os.path.join(self.install_dir,os.path.basename(egg_path))
destination = os.path.abspath(destination)
if not self.dry_run:
ensure_directory(destination)
dist = self.egg_distribution(egg_path)
if not samefile(egg_path, destination):
if os.path.isdir(destination) and not os.path.islink(destination):
dir_util.remove_tree(destination, dry_run=self.dry_run)
elif os.path.exists(destination):
self.execute(os.unlink,(destination,),"Removing "+destination)
uncache_zipdir(destination)
if os.path.isdir(egg_path):
if egg_path.startswith(tmpdir):
f,m = shutil.move, "Moving"
else:
f,m = shutil.copytree, "Copying"
elif self.should_unzip(dist):
self.mkpath(destination)
f,m = self.unpack_and_compile, "Extracting"
elif egg_path.startswith(tmpdir):
f,m = shutil.move, "Moving"
else:
f,m = shutil.copy2, "Copying"
self.execute(f, (egg_path, destination),
(m+" %s to %s") %
(os.path.basename(egg_path),os.path.dirname(destination)))
self.add_output(destination)
return self.egg_distribution(destination)
def _copy_dist_from_dir(link_path, location):
"""Copy distribution files in `link_path` to `location`.
Invoked when user requests to install a local directory. E.g.:
pip install .
pip install ~/dev/git-repos/python-prompt-toolkit
"""
# Note: This is currently VERY SLOW if you have a lot of data in the
# directory, because it copies everything with `shutil.copytree`.
# What it should really do is build an sdist and install that.
# See https://github.com/pypa/pip/issues/2195
if os.path.isdir(location):
rmtree(location)
# build an sdist
setup_py = 'setup.py'
sdist_args = [sys.executable]
sdist_args.append('-c')
sdist_args.append(SETUPTOOLS_SHIM % setup_py)
sdist_args.append('sdist')
sdist_args += ['--dist-dir', location]
logger.info('Running setup.py sdist for %s', link_path)
with indent_log():
call_subprocess(sdist_args, cwd=link_path, show_stdout=False)
# unpack sdist into `location`
sdist = os.path.join(location, os.listdir(location)[0])
logger.info('Unpacking sdist %s into %s', sdist, location)
unpack_file(sdist, location, content_type=None, link=None)
def copy_to_tmp(source):
"""
Copies ``source`` to a temporary directory, and returns the copied location.
"""
tmp_dir = tempfile.mkdtemp()
# Use pathlib because os.path.basename is different depending on whether
# the path ends in a /
p = pathlib.Path(source)
new_dir = os.path.join(tmp_dir, p.name)
if os.path.isdir(source):
shutil.copytree(source, new_dir)
else:
shutil.copy2(source, new_dir)
return new_dir
def copy_files(src, dst, symlinks=False, ignore=None):
"""Copy files from src to dst."""
for item in os.listdir(src):
s = os.path.join(src, item)
d = os.path.join(dst, item)
if os.path.isdir(s):
shutil.copytree(s, d, symlinks, ignore)
else:
shutil.copy2(s, d)
def copy_dirs(self, src_dirs_path, target_dirs):
if not self.has_internal_dirs(target_dirs):
raise Exception("Directory has not been constructed internally! {0}".format(target_dirs))
target_dirs_path = self.dirs_to_path(target_dirs)
if os.path.exists(target_dirs_path):
if len(os.listdir(target_dirs_path)) > 0:
raise Exception("Target path for copying directories is not empty! Got: {0}".format(target_dirs_path))
else:
os.rmdir(target_dirs_path)
shutil.copytree(src_dirs_path, target_dirs_path)
def copy_files(src, dst, symlinks=False, ignore=None):
"""Copy files from src to dst."""
for item in os.listdir(src):
s = os.path.join(src, item)
d = os.path.join(dst, item)
if os.path.isdir(s):
shutil.copytree(s, d, symlinks, ignore)
else:
shutil.copy2(s, d)
def copy_files(src, dst, symlinks=False, ignore=None):
"""Copy files from src to dst."""
for item in os.listdir(src):
s = os.path.join(src, item)
d = os.path.join(dst, item)
if os.path.isdir(s):
shutil.copytree(s, d, symlinks, ignore)
else:
shutil.copy2(s, d)