Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
manifest = parse_manifest(manifest_text)
if poptions.pex_root:
ENV.set('PEX_ROOT', poptions.pex_root)
else:
poptions.pex_root = ENV.PEX_ROOT
if poptions.cache_dir:
poptions.cache_dir = pexbin.make_relative_to_root(poptions.cache_dir)
poptions.interpreter_cache_dir = pexbin.make_relative_to_root(
poptions.interpreter_cache_dir)
reqs = manifest.get('requirements', [])
with ENV.patch(PEX_VERBOSE=str(poptions.verbosity)):
with TRACER.timed('Building pex'):
pex_builder = pexbin.build_pex(reqs, poptions,
resolver_options_builder)
# Add source files from the manifest
for modmap in manifest.get('modules', []):
src = modmap.get('src')
dst = modmap.get('dest')
# NOTE(agallagher): calls the `add_source` and `add_resource` below
# hard-link the given source into the PEX temp dir. Since OS X and
# Linux behave different when hard-linking a source that is a
# symbolic link (Linux does *not* follow symlinks), resolve any
# layers of symlinks here to get consistent behavior.
try:
pex_builder.add_source(dereference_symlinks(src), dst)
except OSError as err:
def force_local(cls, pex, pex_info):
if pex_info.code_hash is None:
# Do not support force_local if code_hash is not set. (It should always be set.)
return pex
explode_dir = os.path.join(pex_info.zip_unsafe_cache, pex_info.code_hash)
TRACER.log('PEX is not zip safe, exploding to %s' % explode_dir)
if not os.path.exists(explode_dir):
explode_tmp = explode_dir + '.' + uuid.uuid4().hex
with TRACER.timed('Unzipping %s' % pex):
try:
safe_mkdir(explode_tmp)
with open_zip(pex) as pex_zip:
pex_files = (x for x in pex_zip.namelist()
if not x.startswith(PEXBuilder.BOOTSTRAP_DIR) and
not x.startswith(PexInfo.INTERNAL_CACHE))
pex_zip.extractall(explode_tmp, pex_files)
except: # noqa: T803
safe_rmtree(explode_tmp)
raise
TRACER.log('Renaming %s to %s' % (explode_tmp, explode_dir))
rename_if_empty(explode_tmp, explode_dir)
return explode_dir
pex_info.zip_safe = options.zip_safe
pex_info.pex_path = options.pex_path
pex_info.always_write_cache = options.always_write_cache
pex_info.ignore_errors = options.ignore_errors
pex_info.emit_warnings = options.emit_warnings
pex_info.inherit_path = options.inherit_path
if options.interpreter_constraint:
for ic in options.interpreter_constraint:
pex_builder.add_interpreter_constraint(ic)
# NB: `None` means use the default (pypi) index, `[]` means use no indexes.
indexes = None
if options.indexes != [_PYPI] and options.indexes is not None:
indexes = [str(index) for index in options.indexes]
with TRACER.timed('Resolving distributions ({})'.format(reqs + options.requirement_files)):
try:
resolveds = resolve_multi(requirements=reqs,
requirement_files=options.requirement_files,
constraint_files=options.constraint_files,
allow_prereleases=options.allow_prereleases,
transitive=options.transitive,
interpreters=interpreters,
platforms=options.platforms,
indexes=indexes,
find_links=options.find_links,
cache=options.cache_dir,
build=options.build,
use_wheel=options.use_wheel,
compile=options.compile,
manylinux=options.manylinux,
max_parallel_jobs=options.max_parallel_jobs,
def build_pex(args, options, resolver_option_builder, interpreter=None):
if interpreter is None:
with TRACER.timed('Resolving interpreter', V=2):
interpreter = interpreter_from_options(options)
if interpreter is None:
die('Could not find compatible interpreter', CANNOT_SETUP_INTERPRETER)
pex_builder = PEXBuilder(path=safe_mkdtemp(), interpreter=interpreter)
pex_info = pex_builder.info
pex_info.zip_safe = options.zip_safe
pex_info.always_write_cache = options.always_write_cache
pex_info.ignore_errors = options.ignore_errors
pex_info.inherit_path = options.inherit_path
resolvables = [Resolvable.get(arg, resolver_option_builder) for arg in args]
for requirements_txt in options.requirement_files:
resolvables = [Resolvable.get(arg, resolver_option_builder) for arg in args]
for requirements_txt in options.requirement_files:
resolvables.extend(requirements_from_file(requirements_txt, resolver_option_builder))
# pip states the constraints format is identical tor requirements
# https://pip.pypa.io/en/stable/user_guide/#constraints-files
for constraints_txt in options.constraint_files:
constraints = []
for r in requirements_from_file(constraints_txt, resolver_option_builder):
r.is_constraint = True
constraints.append(r)
resolvables.extend(constraints)
with TRACER.timed('Resolving distributions'):
try:
resolveds = resolve_multi(resolvables,
interpreters=setup_interpreters,
platforms=options.platforms,
cache=options.cache_dir,
cache_ttl=options.cache_ttl,
allow_prereleases=resolver_option_builder.prereleases_allowed,
use_manylinux=options.use_manylinux)
for dist in resolveds:
log(' %s' % dist, v=options.verbosity)
pex_builder.add_distribution(dist)
pex_builder.add_requirement(dist.as_requirement())
except Unsatisfiable as e:
die(e)
def _load_internal_cache(cls, pex, pex_info):
"""Possibly cache out the internal cache."""
internal_cache = os.path.join(pex, pex_info.internal_cache)
with TRACER.timed('Searching dependency cache: %s' % internal_cache, V=2):
if len(pex_info.distributions) == 0:
# We have no .deps to load.
return
if os.path.isdir(pex):
search_path = [os.path.join(internal_cache, dist_chroot)
for dist_chroot in os.listdir(internal_cache)]
internal_env = Environment(search_path=search_path)
for dist_name in internal_env:
for dist in internal_env[dist_name]:
yield dist
else:
with open_zip(pex) as zf:
for dist in cls._write_zipped_internal_cache(zf, pex_info):
yield dist