20857108027

Committed 09 Jan 2026 03:40PM UTC coverage: 80.255% (-0.01%) from 80.269%

Build # 20857108027

Build Type

Pull #22994

github

Committed by

web-flow

Commit Message

Merge 8a3a64977 into 3782956e6

Pull Request Pull Request #22994: Fix incorrect use of `PEX_PYTHON`.

Run Details

2 of 13 new or added lines in 4 files covered. (15.38%)

25 existing lines in 3 files now uncovered.

78782 of 98164 relevant lines covered (80.26%)

3.36 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

51.43

/src/python/pants/backend/python/goals/repl.py

# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import annotations

import os
from collections.abc import Iterable

from pants.backend.python.subsystems import ipython
from pants.backend.python.subsystems.ipython import IPython
from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.target_types import PythonResolveField
from pants.backend.python.util_rules.local_dists import LocalDistsPexRequest, build_local_dists
from pants.backend.python.util_rules.pex import create_pex
from pants.backend.python.util_rules.pex_environment import PexEnvironment
from pants.backend.python.util_rules.pex_from_targets import (
    InterpreterConstraintsRequest,
    RequirementsPexRequest,
    interpreter_constraints_for_targets,
)
from pants.backend.python.util_rules.python_sources import (
    PythonSourceFilesRequest,
    prepare_python_sources,
)
from pants.core.goals.generate_lockfiles import NoCompatibleResolveException
from pants.core.goals.repl import ReplImplementation, ReplRequest
from pants.engine.fs import MergeDigests
from pants.engine.internals.graph import transitive_targets as transitive_targets_get
from pants.engine.intrinsics import merge_digests
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import Target, TransitiveTargetsRequest
from pants.engine.unions import UnionRule
from pants.util.docutil import bin_name
from pants.util.logging import LogLevel
from pants.util.strutil import softwrap


def validate_compatible_resolve(root_targets: Iterable[Target], python_setup: PythonSetup) -> None:
    """Eagerly validate that all roots are compatible.

    We already end up checking this in pex_from_targets.py, but this is a more eager check so that
    we have a better error message.
    """
    root_resolves = {
        root[PythonResolveField].normalized_value(python_setup)
        for root in root_targets
        if root.has_field(PythonResolveField)
    }

    def maybe_get_resolve(t: Target) -> str | None:
        if not t.has_field(PythonResolveField):
            return None
        return t[PythonResolveField].normalized_value(python_setup)

    if len(root_resolves) > 1:
        raise NoCompatibleResolveException.bad_input_roots(
            root_targets,
            maybe_get_resolve=maybe_get_resolve,
            doc_url_slug="docs/python/overview/lockfiles#multiple-lockfiles",
            workaround=softwrap(
                f"""
                To work around this, choose which resolve you want to use from above. Then, run
                `{bin_name()} peek :: | jq -r \'.[] | select(.resolve == "example") |
                .["address"]\' | xargs {bin_name()} repl`, where you replace "example" with the
                resolve name, and possibly replace the specs `::` with what you were using
                before. If the resolve is the `[python].default_resolve`, use
                `select(.resolve == "example" or .resolve == null)`. These queries will result in
                opening a REPL with only targets using the desired resolve.
                """
            ),
        )


class PythonRepl(ReplImplementation):
    name = "python"
    supports_args = False


@rule(level=LogLevel.DEBUG)
async def create_python_repl_request(
    request: PythonRepl, pex_env: PexEnvironment, python_setup: PythonSetup
) -> ReplRequest:
    validate_compatible_resolve(request.targets, python_setup)

    interpreter_constraints, transitive_targets = await concurrently(
        interpreter_constraints_for_targets(
            InterpreterConstraintsRequest(request.addresses), **implicitly()
        ),
        transitive_targets_get(TransitiveTargetsRequest(request.addresses), **implicitly()),
    )

    requirements_request = create_pex(**implicitly(RequirementsPexRequest(request.addresses)))
    local_dists_request = build_local_dists(
        LocalDistsPexRequest(
            request.addresses,
            interpreter_constraints=interpreter_constraints,
        )
    )

    sources_request = prepare_python_sources(
        PythonSourceFilesRequest(transitive_targets.closure, include_files=True), **implicitly()
    )

    requirements_pex, local_dists, sources = await concurrently(
        requirements_request, local_dists_request, sources_request
    )
    merged_digest = await merge_digests(
        MergeDigests(
            (requirements_pex.digest, local_dists.pex.digest, sources.source_files.snapshot.digest)
        )
    )

    complete_pex_env = pex_env.in_workspace()
    args = complete_pex_env.create_argv(
        request.in_chroot(requirements_pex.name), python=requirements_pex.python
    )

    chrooted_source_roots = [request.in_chroot(sr) for sr in sources.source_roots]
    extra_env = {
        **complete_pex_env.environment_dict(python_configured=requirements_pex.python is not None),
        "PEX_EXTRA_SYS_PATH": ":".join(chrooted_source_roots),
        "PEX_PATH": request.in_chroot(local_dists.pex.name),
        "PEX_INTERPRETER_HISTORY": "1" if python_setup.repl_history else "0",
    }

    return ReplRequest(digest=merged_digest, args=args, extra_env=extra_env)


class IPythonRepl(ReplImplementation):
    name = "ipython"
    supports_args = True


@rule(level=LogLevel.DEBUG)
async def create_ipython_repl_request(
    request: IPythonRepl, ipython: IPython, pex_env: PexEnvironment, python_setup: PythonSetup
) -> ReplRequest:
    validate_compatible_resolve(request.targets, python_setup)

    interpreter_constraints, transitive_targets = await concurrently(
        interpreter_constraints_for_targets(
            InterpreterConstraintsRequest(request.addresses), **implicitly()
        ),
        transitive_targets_get(TransitiveTargetsRequest(request.addresses), **implicitly()),
    )

    requirements_request = create_pex(**implicitly(RequirementsPexRequest(request.addresses)))
    sources_request = prepare_python_sources(
        PythonSourceFilesRequest(transitive_targets.closure, include_files=True), **implicitly()
    )

    ipython_request = create_pex(
        ipython.to_pex_request(interpreter_constraints=interpreter_constraints)
    )

    requirements_pex, sources, ipython_pex = await concurrently(
        requirements_request, sources_request, ipython_request
    )

    local_dists = await build_local_dists(
        LocalDistsPexRequest(
            request.addresses,
            interpreter_constraints=interpreter_constraints,
            sources=sources,
        )
    )

    merged_digest = await merge_digests(
        MergeDigests(
            (
                requirements_pex.digest,
                local_dists.pex.digest,
                local_dists.remaining_sources.source_files.snapshot.digest,
                ipython_pex.digest,
            )
        )
    )

    complete_pex_env = pex_env.in_workspace()
    args = list(
        complete_pex_env.create_argv(request.in_chroot(ipython_pex.name), python=ipython_pex.python)
    )
    if ipython.ignore_cwd:
        args.append("--ignore-cwd")

    chrooted_source_roots = [request.in_chroot(sr) for sr in sources.source_roots]
    extra_env = {
        **complete_pex_env.environment_dict(python_configured=ipython_pex.python is not None),
        "PEX_PATH": os.pathsep.join(
            [
                request.in_chroot(requirements_pex.name),
                request.in_chroot(local_dists.pex.name),
            ]
        ),
        "PEX_EXTRA_SYS_PATH": os.pathsep.join(chrooted_source_roots),
    }

    return ReplRequest(digest=merged_digest, args=args, extra_env=extra_env)


def rules():
    return [
        *collect_rules(),
        *ipython.rules(),
        UnionRule(ReplImplementation, PythonRepl),
        UnionRule(ReplImplementation, IPythonRepl),
    ]

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	7✔
5
6	import os	7✔
7	from collections.abc import Iterable	7✔
8
9	from pants.backend.python.subsystems import ipython	7✔
10	from pants.backend.python.subsystems.ipython import IPython	7✔
11	from pants.backend.python.subsystems.setup import PythonSetup	7✔
12	from pants.backend.python.target_types import PythonResolveField	7✔
13	from pants.backend.python.util_rules.local_dists import LocalDistsPexRequest, build_local_dists	7✔
14	from pants.backend.python.util_rules.pex import create_pex	7✔
15	from pants.backend.python.util_rules.pex_environment import PexEnvironment	7✔
16	from pants.backend.python.util_rules.pex_from_targets import (	7✔
17	InterpreterConstraintsRequest,
18	RequirementsPexRequest,
19	interpreter_constraints_for_targets,
20	)
21	from pants.backend.python.util_rules.python_sources import (	7✔
22	PythonSourceFilesRequest,
23	prepare_python_sources,
24	)
25	from pants.core.goals.generate_lockfiles import NoCompatibleResolveException	7✔
26	from pants.core.goals.repl import ReplImplementation, ReplRequest	7✔
27	from pants.engine.fs import MergeDigests	7✔
28	from pants.engine.internals.graph import transitive_targets as transitive_targets_get	7✔
29	from pants.engine.intrinsics import merge_digests	7✔
30	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	7✔
31	from pants.engine.target import Target, TransitiveTargetsRequest	7✔
32	from pants.engine.unions import UnionRule	7✔
33	from pants.util.docutil import bin_name	7✔
34	from pants.util.logging import LogLevel	7✔
35	from pants.util.strutil import softwrap	7✔
36
37
38	def validate_compatible_resolve(root_targets: Iterable[Target], python_setup: PythonSetup) -> None:	7✔
39	"""Eagerly validate that all roots are compatible.
40
41	We already end up checking this in pex_from_targets.py, but this is a more eager check so that
42	we have a better error message.
43	"""
44	root_resolves = {	×
45	root[PythonResolveField].normalized_value(python_setup)
46	for root in root_targets
47	if root.has_field(PythonResolveField)
48	}
49
50	def maybe_get_resolve(t: Target) -> str \| None:	×
51	if not t.has_field(PythonResolveField):	×
52	return None	×
53	return t[PythonResolveField].normalized_value(python_setup)	×
54
55	if len(root_resolves) > 1:	×
56	raise NoCompatibleResolveException.bad_input_roots(	×
57	root_targets,
58	maybe_get_resolve=maybe_get_resolve,
59	doc_url_slug="docs/python/overview/lockfiles#multiple-lockfiles",
60	workaround=softwrap(
61	f"""
62	To work around this, choose which resolve you want to use from above. Then, run
63	`{bin_name()} peek :: \| jq -r \'.[] \| select(.resolve == "example") \|
64	.["address"]\' \| xargs {bin_name()} repl`, where you replace "example" with the
65	resolve name, and possibly replace the specs `::` with what you were using
66	before. If the resolve is the `[python].default_resolve`, use
67	`select(.resolve == "example" or .resolve == null)`. These queries will result in
68	opening a REPL with only targets using the desired resolve.
69	"""
70	),
71	)
72
73
74	class PythonRepl(ReplImplementation):	7✔
75	name = "python"	7✔
76	supports_args = False	7✔
77
78
79	@rule(level=LogLevel.DEBUG)	7✔
80	async def create_python_repl_request(	7✔
81	request: PythonRepl, pex_env: PexEnvironment, python_setup: PythonSetup
82	) -> ReplRequest:
83	validate_compatible_resolve(request.targets, python_setup)	×
84
85	interpreter_constraints, transitive_targets = await concurrently(	×
86	interpreter_constraints_for_targets(
87	InterpreterConstraintsRequest(request.addresses), **implicitly()
88	),
89	transitive_targets_get(TransitiveTargetsRequest(request.addresses), **implicitly()),
90	)
91
92	requirements_request = create_pex(**implicitly(RequirementsPexRequest(request.addresses)))	×
93	local_dists_request = build_local_dists(	×
94	LocalDistsPexRequest(
95	request.addresses,
96	interpreter_constraints=interpreter_constraints,
97	)
98	)
99
100	sources_request = prepare_python_sources(	×
101	PythonSourceFilesRequest(transitive_targets.closure, include_files=True), **implicitly()
102	)
103
104	requirements_pex, local_dists, sources = await concurrently(	×
105	requirements_request, local_dists_request, sources_request
106	)
107	merged_digest = await merge_digests(	×
108	MergeDigests(
109	(requirements_pex.digest, local_dists.pex.digest, sources.source_files.snapshot.digest)
110	)
111	)
112
113	complete_pex_env = pex_env.in_workspace()	×
NEW 114	args = complete_pex_env.create_argv(	×
115	request.in_chroot(requirements_pex.name), python=requirements_pex.python
116	)
117
118	chrooted_source_roots = [request.in_chroot(sr) for sr in sources.source_roots]	×
119	extra_env = {	×
120	**complete_pex_env.environment_dict(python_configured=requirements_pex.python is not None),
121	"PEX_EXTRA_SYS_PATH": ":".join(chrooted_source_roots),
122	"PEX_PATH": request.in_chroot(local_dists.pex.name),
123	"PEX_INTERPRETER_HISTORY": "1" if python_setup.repl_history else "0",
124	}
125
126	return ReplRequest(digest=merged_digest, args=args, extra_env=extra_env)	×
127
128
129	class IPythonRepl(ReplImplementation):	7✔
130	name = "ipython"	7✔
131	supports_args = True	7✔
132
133
134	@rule(level=LogLevel.DEBUG)	7✔
135	async def create_ipython_repl_request(	7✔
136	request: IPythonRepl, ipython: IPython, pex_env: PexEnvironment, python_setup: PythonSetup
137	) -> ReplRequest:
138	validate_compatible_resolve(request.targets, python_setup)	×
139
140	interpreter_constraints, transitive_targets = await concurrently(	×
141	interpreter_constraints_for_targets(
142	InterpreterConstraintsRequest(request.addresses), **implicitly()
143	),
144	transitive_targets_get(TransitiveTargetsRequest(request.addresses), **implicitly()),
145	)
146
147	requirements_request = create_pex(**implicitly(RequirementsPexRequest(request.addresses)))	×
148	sources_request = prepare_python_sources(	×
149	PythonSourceFilesRequest(transitive_targets.closure, include_files=True), **implicitly()
150	)
151
152	ipython_request = create_pex(	×
153	ipython.to_pex_request(interpreter_constraints=interpreter_constraints)
154	)
155
156	requirements_pex, sources, ipython_pex = await concurrently(	×
157	requirements_request, sources_request, ipython_request
158	)
159
160	local_dists = await build_local_dists(	×
161	LocalDistsPexRequest(
162	request.addresses,
163	interpreter_constraints=interpreter_constraints,
164	sources=sources,
165	)
166	)
167
168	merged_digest = await merge_digests(	×
169	MergeDigests(
170	(
171	requirements_pex.digest,
172	local_dists.pex.digest,
173	local_dists.remaining_sources.source_files.snapshot.digest,
174	ipython_pex.digest,
175	)
176	)
177	)
178
179	complete_pex_env = pex_env.in_workspace()	×
NEW 180	args = list(	×
181	complete_pex_env.create_argv(request.in_chroot(ipython_pex.name), python=ipython_pex.python)
182	)
183	if ipython.ignore_cwd:	×
184	args.append("--ignore-cwd")	×
185
186	chrooted_source_roots = [request.in_chroot(sr) for sr in sources.source_roots]	×
187	extra_env = {	×
188	**complete_pex_env.environment_dict(python_configured=ipython_pex.python is not None),
189	"PEX_PATH": os.pathsep.join(
190	[
191	request.in_chroot(requirements_pex.name),
192	request.in_chroot(local_dists.pex.name),
193	]
194	),
195	"PEX_EXTRA_SYS_PATH": os.pathsep.join(chrooted_source_roots),
196	}
197
198	return ReplRequest(digest=merged_digest, args=args, extra_env=extra_env)	×
199
200
201	def rules():	7✔
202	return [	7✔
203	*collect_rules(),
204	*ipython.rules(),
205	UnionRule(ReplImplementation, PythonRepl),
206	UnionRule(ReplImplementation, IPythonRepl),
207	]

pantsbuild / pants / 20857108027

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous