How to use pex - 10 common examples

pex_verbosity = ENV.PEX_VERBOSE
    pip_verbosity = pex_verbosity // 3
    if pip_verbosity > 0:
      pip_args.append('-{}'.format('v' * pip_verbosity))
    else:
      pip_args.append('-q')

    if cache:
      pip_args.extend(['--cache-dir', cache])
    else:
      pip_args.append('--no-cache-dir')

    command = pip_args + args
    with ENV.strip().patch(PEX_ROOT=ENV.PEX_ROOT, PEX_VERBOSE=str(pex_verbosity)) as env:
      from pex.pex import PEX
      pip = PEX(pex=self._pip_pex_path, interpreter=interpreter)
      return Job(
        command=pip.cmdline(command),
        process=pip.run(
          args=command,
          env=env,
          blocking=False
        )

def bdist_pex(project_dir, bdist_args=None):
  with temporary_dir() as dist_dir:
    cmd = ['setup.py', 'bdist_pex', '--bdist-dir={}'.format(dist_dir)]
    if bdist_args:
      cmd.extend(bdist_args)

    spawn_python_job(args=cmd, cwd=project_dir).wait()
    dists = os.listdir(dist_dir)
    assert len(dists) == 1
    yield os.path.join(dist_dir, dists[0])

def test_issues_539_abi3_resolution():
  # The cryptography team releases the following relevant pre-built wheels for version 2.6.1:
  # cryptography-2.6.1-cp27-cp27m-macosx_10_6_intel.whl
  # cryptography-2.6.1-cp27-cp27m-manylinux1_x86_64.whl
  # cryptography-2.6.1-cp27-cp27mu-manylinux1_x86_64.whl
  # cryptography-2.6.1-cp34-abi3-macosx_10_6_intel.whl
  # cryptography-2.6.1-cp34-abi3-manylinux1_x86_64.whl
  # With pex in --no-build mode, we force a test that pex abi3 resolution works when this test is
  # run under CPython>3.4,<4 on OSX and linux.

  with temporary_dir() as td:
    # The dependency graph for cryptography-2.6.1 includes pycparser which is only released as an
    # sdist. Since we want to test in --no-build, we pre-resolve/build the pycparser wheel here and
    # add the resulting wheelhouse to the --no-build pex command.
    download_dir = os.path.join(td, '.downloads')
    get_pip().spawn_download_distributions(
      download_dir=download_dir,
      requirements=['pycparser']
    ).wait()
    wheel_dir = os.path.join(td, '.wheels')
    get_pip().spawn_build_wheels(
      wheel_dir=wheel_dir,
      distributions=glob.glob(os.path.join(download_dir, '*'))
    ).wait()

    cryptography_pex = os.path.join(td, 'cryptography.pex')
    res = run_pex_command(['-f', wheel_dir,

def test_pex_run():
  with named_temporary_file() as fake_stdout:
    with temporary_dir() as temp_dir:
      pex = write_simple_pex(
        temp_dir,
        'import sys; sys.stdout.write("hello"); sys.stderr.write("hello"); sys.exit(0)'
      )
      rc = PEX(pex.path()).run(stdin=None, stdout=fake_stdout, stderr=fake_stdout)
      assert rc == 0

      fake_stdout.seek(0)
      assert fake_stdout.read() == b'hellohello'

with temporary_dir() as td:
    target = os.path.join(td, 'foo.pex')
    should_create = os.path.join(td, 'foo.1')

    tempfile_preamble = "\n".join([
      "import sys",
      "open('{0}', 'w').close()".format(should_create),
      "sys.exit(3)"
    ])

    pex_builder = PEXBuilder(preamble=tempfile_preamble)
    pex_builder.build(target)

    assert not os.path.exists(should_create)

    pex = PEX(target)
    process = pex.run(blocking=False)
    process.wait()

    assert process.returncode == 3
    assert os.path.exists(should_create)

def do_resolve(req_name, req_version, platform, extra_flags=None):
      extra_flags = extra_flags or ''
      pex_path = os.path.join(output_dir, 'test.pex')
      results = run_pex_command(['--disable-cache',
                                 '--no-build',
                                 '%s==%s' % (req_name, req_version),
                                 '--platform=%s' % (platform),
                                 '-o', pex_path] + extra_flags.split())
      return pex_path, results

def test_pex_repl_built():
  """Tests the REPL in the context of a built pex."""
  stdin_payload = b'import requests; import sys; sys.exit(3)'

  with temporary_dir() as output_dir:
    # Create a temporary pex containing just `requests` with no entrypoint.
    pex_path = os.path.join(output_dir, 'requests.pex')
    results = run_pex_command(['--disable-cache', 'requests', '-o', pex_path])
    results.assert_success()

    # Test that the REPL is functional.
    stdout, rc = run_simple_pex(pex_path, stdin=stdin_payload)
    assert rc == 3
    assert b'>>>' in stdout

def test_interpreter_resolution_pex_python_path_precedence_over_pex_python():
  with temporary_dir() as td:
    pexrc_path = os.path.join(td, '.pexrc')
    with open(pexrc_path, 'w') as pexrc:
      # set both PPP and PP
      pex_python_path = ':'.join([
        ensure_python_interpreter(PY27),
        ensure_python_interpreter(PY36)
      ])
      pexrc.write("PEX_PYTHON_PATH=%s\n" % pex_python_path)
      pex_python = '/path/to/some/python'
      pexrc.write("PEX_PYTHON=%s" % pex_python)

    pex_out_path = os.path.join(td, 'pex.pex')
    res = run_pex_command(['--disable-cache',
      '--rcfile=%s' % pexrc_path,
      '--interpreter-constraint=>3,<3.8',
      '-o', pex_out_path])
    res.assert_success()

    stdin_payload = b'import sys; print(sys.executable); sys.exit(0)'
    stdout, rc = run_simple_pex(pex_out_path, stdin=stdin_payload)
    assert rc == 0
    correct_interpreter_path = pex_python_path.split(':')[1].encode()
    assert correct_interpreter_path in stdout

def test_interpreter_constraints_honored_without_ppp_or_pp():
  # Create a pex with interpreter constraints, but for not the default interpreter in the path.
  with temporary_dir() as td:
    py36_path = ensure_python_interpreter(PY36)
    py35_path = ensure_python_interpreter(PY35)

    pex_out_path = os.path.join(td, 'pex.pex')
    env = make_env(
      PEX_IGNORE_RCFILES="1",
      PATH=os.pathsep.join([
        os.path.dirname(py35_path),
        os.path.dirname(py36_path),
      ])
    )
    res = run_pex_command(['--disable-cache',
      '--interpreter-constraint===%s' % PY36,
      '-o', pex_out_path],
      env=env
    )
    res.assert_success()

    # We want to try to run that pex with no environment variables set
    stdin_payload = b'import sys; print(sys.executable); sys.exit(0)'

    stdout, rc = run_simple_pex(pex_out_path, stdin=stdin_payload, env=env)
    assert rc == 0

    # If the constraints are honored, it will have run python3.6 and not python3.5
    # Without constraints, we would expect it to use python3.5 as it is the minimum interpreter
    # in the PATH.
    assert str(py36_path).encode() in stdout

test_file_path = os.path.join(td, 'build_and_run_child_pex.py')
            with open(test_file_path, 'w') as fh:
              fh.write(dedent("""
                import sys
                print(sys.executable)
                """))
            pex_out_path = os.path.join(td, 'child.pex')
            res = run_pex_command(['--disable-cache',
              '-o', pex_out_path])
            stdin_payload = b'import sys; print(sys.executable); sys.exit(0)'
            stdout, rc = run_simple_pex(pex_out_path, stdin=stdin_payload)
            print(stdout)
        '''.format(ensure_python_interpreter(child_pex_interpreter_version))))

    pex_out_path = os.path.join(td, 'parent.pex')
    res = run_pex_command(['--disable-cache',
      'pex',
      '{}'.format(td),
      '-e', 'testing:tester',
      '-o', pex_out_path])
    res.assert_success()

    stdout, rc = run_simple_pex(pex_out_path)
    assert rc == 0
    # Ensure that child pex used the proper interpreter as specified by its pexrc.
    correct_interpreter_path = ensure_python_interpreter(child_pex_interpreter_version)
    correct_interpreter_path = correct_interpreter_path.encode()  # Py 2/3 compatibility
    assert correct_interpreter_path in stdout