22285099215

Committed 22 Feb 2026 08:52PM UTC coverage: 75.854% (-17.1%) from 92.936%

Build # 22285099215

Build Type

Pull #23121

github

Committed by

web-flow

Commit Message

Merge c7299df9c into ba8359840

Pull Request Pull Request #23121: fix issue with optional fields in dependency validator

Run Details

28 of 29 new or added lines in 2 files covered. (96.55%)

11174 existing lines in 400 files now uncovered.

53694 of 70786 relevant lines covered (75.85%)

1.88 hits per line

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

77.14

/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	2✔
5
6	import os	2✔
7	from collections.abc import Iterable	2✔
8
9	from pants.backend.python.subsystems import ipython	2✔
10	from pants.backend.python.subsystems.ipython import IPython	2✔
11	from pants.backend.python.subsystems.setup import PythonSetup	2✔
12	from pants.backend.python.target_types import PythonResolveField	2✔
13	from pants.backend.python.util_rules.local_dists import LocalDistsPexRequest, build_local_dists	2✔
14	from pants.backend.python.util_rules.pex import create_pex	2✔
15	from pants.backend.python.util_rules.pex_environment import PexEnvironment	2✔
16	from pants.backend.python.util_rules.pex_from_targets import (	2✔
17	InterpreterConstraintsRequest,
18	RequirementsPexRequest,
19	interpreter_constraints_for_targets,
20	)
21	from pants.backend.python.util_rules.python_sources import (	2✔
22	PythonSourceFilesRequest,
23	prepare_python_sources,
24	)
25	from pants.core.goals.generate_lockfiles import NoCompatibleResolveException	2✔
26	from pants.core.goals.repl import ReplImplementation, ReplRequest	2✔
27	from pants.engine.fs import MergeDigests	2✔
28	from pants.engine.internals.graph import transitive_targets as transitive_targets_get	2✔
29	from pants.engine.intrinsics import merge_digests	2✔
30	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	2✔
31	from pants.engine.target import Target, TransitiveTargetsRequest	2✔
32	from pants.engine.unions import UnionRule	2✔
33	from pants.util.docutil import bin_name	2✔
34	from pants.util.logging import LogLevel	2✔
35	from pants.util.strutil import softwrap	2✔
36
37
38	def validate_compatible_resolve(root_targets: Iterable[Target], python_setup: PythonSetup) -> None:	2✔
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 = {	1✔
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:	1✔
UNCOV 51	if not t.has_field(PythonResolveField):	×
52	return None	×
UNCOV 53	return t[PythonResolveField].normalized_value(python_setup)	×
54
55	if len(root_resolves) > 1:	1✔
UNCOV 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):	2✔
75	name = "python"	2✔
76	supports_args = False	2✔
77
78
79	@rule(level=LogLevel.DEBUG)	2✔
80	async def create_python_repl_request(	2✔
81	request: PythonRepl, pex_env: PexEnvironment, python_setup: PythonSetup
82	) -> ReplRequest:
UNCOV 83	validate_compatible_resolve(request.targets, python_setup)	×
84
UNCOV 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
UNCOV 92	requirements_request = create_pex(**implicitly(RequirementsPexRequest(request.addresses)))	×
UNCOV 93	local_dists_request = build_local_dists(	×
94	LocalDistsPexRequest(
95	request.addresses,
96	interpreter_constraints=interpreter_constraints,
97	)
98	)
99
UNCOV 100	sources_request = prepare_python_sources(	×
101	PythonSourceFilesRequest(transitive_targets.closure, include_files=True), **implicitly()
102	)
103
UNCOV 104	requirements_pex, local_dists, sources = await concurrently(	×
105	requirements_request, local_dists_request, sources_request
106	)
UNCOV 107	merged_digest = await merge_digests(	×
108	MergeDigests(
109	(requirements_pex.digest, local_dists.pex.digest, sources.source_files.snapshot.digest)
110	)
111	)
112
UNCOV 113	complete_pex_env = pex_env.in_workspace()	×
UNCOV 114	args = complete_pex_env.create_argv(	×
115	request.in_chroot(requirements_pex.name), python=requirements_pex.python
116	)
117
UNCOV 118	chrooted_source_roots = [request.in_chroot(sr) for sr in sources.source_roots]	×
UNCOV 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
UNCOV 126	return ReplRequest(digest=merged_digest, args=args, extra_env=extra_env)	×
127
128
129	class IPythonRepl(ReplImplementation):	2✔
130	name = "ipython"	2✔
131	supports_args = True	2✔
132
133
134	@rule(level=LogLevel.DEBUG)	2✔
135	async def create_ipython_repl_request(	2✔
136	request: IPythonRepl, ipython: IPython, pex_env: PexEnvironment, python_setup: PythonSetup
137	) -> ReplRequest:
138	validate_compatible_resolve(request.targets, python_setup)	1✔
139
140	interpreter_constraints, transitive_targets = await concurrently(	1✔
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)))	1✔
148	sources_request = prepare_python_sources(	1✔
149	PythonSourceFilesRequest(transitive_targets.closure, include_files=True), **implicitly()
150	)
151
152	ipython_request = create_pex(	1✔
153	ipython.to_pex_request(interpreter_constraints=interpreter_constraints)
154	)
155
156	requirements_pex, sources, ipython_pex = await concurrently(	1✔
157	requirements_request, sources_request, ipython_request
158	)
159
160	local_dists = await build_local_dists(	1✔
161	LocalDistsPexRequest(
162	request.addresses,
163	interpreter_constraints=interpreter_constraints,
164	sources=sources,
165	)
166	)
167
168	merged_digest = await merge_digests(	1✔
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()	1✔
180	args = list(	1✔
181	complete_pex_env.create_argv(request.in_chroot(ipython_pex.name), python=ipython_pex.python)
182	)
183	if ipython.ignore_cwd:	1✔
184	args.append("--ignore-cwd")	1✔
185
186	chrooted_source_roots = [request.in_chroot(sr) for sr in sources.source_roots]	1✔
187	extra_env = {	1✔
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)	1✔
199
200
201	def rules():	2✔
202	return [	2✔
203	*collect_rules(),
204	*ipython.rules(),
205	UnionRule(ReplImplementation, PythonRepl),
206	UnionRule(ReplImplementation, IPythonRepl),
207	]

pantsbuild / pants / 22285099215

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