18143316655

Committed 30 Sep 2025 09:00PM UTC coverage: 80.263% (-0.01%) from 80.275%

Build # 18143316655

Build Type

push

github

Committed by

web-flow

Commit Message

Write Python lockfile metadata to separate files (#22713)

Currently we tack it on as a header to the lockfile, which
makes the lockfile unusable when working directly with Pex
without first manually editing it to remove the header.

Instead, we now (optionally) write to a separate metadata
sibling file. 

We always unconditionally try and read the metadata file,
falling back to the header if it doesn't exist. This will
allow us to regenerate the embedded lockfiles without
worrying about whether the user has the new metadata
files enabled.

Run Details

42 of 87 new or added lines in 7 files covered. (48.28%)

1 existing line in 1 file now uncovered.

77226 of 96216 relevant lines covered (80.26%)

3.37 hits per line

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

52.0

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

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

from __future__ import annotations

import itertools
import os.path
from collections import defaultdict
from dataclasses import dataclass
from operator import itemgetter

from pants.backend.python.subsystems.python_tool_base import PythonToolBase
from pants.backend.python.subsystems.setup import PythonSetup
from pants.backend.python.target_types import (
    PythonRequirementFindLinksField,
    PythonRequirementResolveField,
    PythonRequirementsField,
)
from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints
from pants.backend.python.util_rules.lockfile_diff import _generate_python_lockfile_diff
from pants.backend.python.util_rules.lockfile_metadata import PythonLockfileMetadata
from pants.backend.python.util_rules.pex import find_interpreter
from pants.backend.python.util_rules.pex_cli import PexCliProcess, maybe_log_pex_stderr
from pants.backend.python.util_rules.pex_environment import PexSubsystem
from pants.backend.python.util_rules.pex_requirements import (
    PexRequirements,
    ResolvePexConfig,
    ResolvePexConfigRequest,
    determine_resolve_pex_config,
)
from pants.core.goals.generate_lockfiles import (
    DEFAULT_TOOL_LOCKFILE,
    GenerateLockfile,
    GenerateLockfileResult,
    GenerateLockfilesSubsystem,
    KnownUserResolveNames,
    KnownUserResolveNamesRequest,
    RequestedUserResolveNames,
    UserGenerateLockfiles,
    WrappedGenerateLockfile,
)
from pants.core.goals.resolves import ExportableTool
from pants.core.util_rules.lockfile_metadata import calculate_invalidation_digest
from pants.engine.fs import CreateDigest, Digest, FileContent, MergeDigests
from pants.engine.internals.synthetic_targets import SyntheticAddressMaps, SyntheticTargetsRequest
from pants.engine.internals.target_adaptor import TargetAdaptor
from pants.engine.intrinsics import create_digest, get_digest_contents, merge_digests
from pants.engine.process import ProcessCacheScope, fallible_to_exec_result_or_raise
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.target import AllTargets
from pants.engine.unions import UnionMembership, UnionRule
from pants.option.subsystem import _construct_subsystem
from pants.util.docutil import bin_name
from pants.util.logging import LogLevel
from pants.util.ordered_set import FrozenOrderedSet
from pants.util.pip_requirement import PipRequirement


@dataclass(frozen=True)
class GeneratePythonLockfile(GenerateLockfile):
    requirements: FrozenOrderedSet[str]
    find_links: FrozenOrderedSet[str]
    interpreter_constraints: InterpreterConstraints

    @property
    def requirements_hex_digest(self) -> str:
        """Produces a hex digest of the requirements input for this lockfile."""
        return calculate_invalidation_digest(self.requirements)


@rule
async def wrap_python_lockfile_request(request: GeneratePythonLockfile) -> WrappedGenerateLockfile:
    return WrappedGenerateLockfile(request)


@dataclass(frozen=True)
class _PipArgsAndConstraintsSetup:
    resolve_config: ResolvePexConfig
    args: tuple[str, ...]
    digest: Digest


async def _setup_pip_args_and_constraints_file(resolve_name: str) -> _PipArgsAndConstraintsSetup:
    resolve_config = await determine_resolve_pex_config(
        ResolvePexConfigRequest(resolve_name), **implicitly()
    )

    args = list(resolve_config.pex_args())
    digests: list[Digest] = []

    if resolve_config.constraints_file:
        args.append(f"--constraints={resolve_config.constraints_file.path}")
        digests.append(resolve_config.constraints_file.digest)

    input_digest = await merge_digests(MergeDigests(digests))
    return _PipArgsAndConstraintsSetup(resolve_config, tuple(args), input_digest)


@rule(desc="Generate Python lockfile", level=LogLevel.DEBUG)
async def generate_lockfile(
    req: GeneratePythonLockfile,
    generate_lockfiles_subsystem: GenerateLockfilesSubsystem,
    python_setup: PythonSetup,
    pex_subsystem: PexSubsystem,
) -> GenerateLockfileResult:
    if not req.requirements:
        raise ValueError(
            f"Cannot generate lockfile with no requirements. Please add some requirements to {req.resolve_name}."
        )

    pip_args_setup = await _setup_pip_args_and_constraints_file(req.resolve_name)
    header_delimiter = "//"

    python = await find_interpreter(req.interpreter_constraints, **implicitly())

    result = await fallible_to_exec_result_or_raise(
        **implicitly(
            PexCliProcess(
                subcommand=("lock", "create"),
                extra_args=(
                    f"--output={req.lockfile_dest}",
                    # See https://github.com/pantsbuild/pants/issues/12458. For now, we always
                    # generate universal locks because they have the best compatibility. We may
                    # want to let users change this, as `style=strict` is safer.
                    "--style=universal",
                    "--pip-version",
                    python_setup.pip_version,
                    "--resolver-version",
                    "pip-2020-resolver",
                    "--preserve-pip-download-log",
                    "pex-pip-download.log",
                    # PEX files currently only run on Linux and Mac machines; so we hard code this
                    # limit on lock universality to avoid issues locking due to irrelevant
                    # Windows-only dependency issues. See this Pex issue that originated from a
                    # Pants user issue presented in Slack:
                    #   https://github.com/pex-tool/pex/issues/1821
                    #
                    # At some point it will probably make sense to expose `--target-system` for
                    # configuration.
                    "--target-system",
                    "linux",
                    "--target-system",
                    "mac",
                    # This makes diffs more readable when lockfiles change.
                    "--indent=2",
                    f"--python-path={python.path}",
                    *(f"--find-links={link}" for link in req.find_links),
                    *pip_args_setup.args,
                    *req.interpreter_constraints.generate_pex_arg_list(),
                    *req.requirements,
                ),
                additional_input_digest=pip_args_setup.digest,
                output_files=(req.lockfile_dest,),
                description=f"Generate lockfile for {req.resolve_name}",
                # Instead of caching lockfile generation with LMDB, we instead use the invalidation
                # scheme from `lockfile_metadata.py` to check for stale/invalid lockfiles. This is
                # necessary so that our invalidation is resilient to deleting LMDB or running on a
                # new machine.
                #
                # We disable caching with LMDB so that when you generate a lockfile, you always get
                # the most up-to-date snapshot of the world. This is generally desirable and also
                # necessary to avoid an awkward edge case where different developers generate
                # different lockfiles even when generating at the same time. See
                # https://github.com/pantsbuild/pants/issues/12591.
                cache_scope=ProcessCacheScope.PER_SESSION,
            )
        )
    )
    maybe_log_pex_stderr(result.stderr, pex_subsystem.verbosity)

    metadata = PythonLockfileMetadata.new(
        valid_for_interpreter_constraints=req.interpreter_constraints,
        requirements={
            PipRequirement.parse(
                i,
                description_of_origin=f"the lockfile {req.lockfile_dest} for the resolve {req.resolve_name}",
            )
            for i in req.requirements
        },
        manylinux=pip_args_setup.resolve_config.manylinux,
        requirement_constraints=(
            set(pip_args_setup.resolve_config.constraints_file.constraints)
            if pip_args_setup.resolve_config.constraints_file
            else set()
        ),
        only_binary=set(pip_args_setup.resolve_config.only_binary),
        no_binary=set(pip_args_setup.resolve_config.no_binary),
        excludes=set(pip_args_setup.resolve_config.excludes),
        overrides=set(pip_args_setup.resolve_config.overrides),
    )
    regenerate_command = (
        generate_lockfiles_subsystem.custom_command
        or f"{bin_name()} generate-lockfiles --resolve={req.resolve_name}"
    )
    if python_setup.separate_lockfile_metadata_file:
        descr = f"This lockfile was generated by Pants. To regenerate, run: {regenerate_command}"
        metadata_digest = await create_digest(
            CreateDigest(
                [
                    FileContent(
                        PythonLockfileMetadata.metadata_location_for_lockfile(req.lockfile_dest),
                        metadata.to_json(with_description=descr).encode(),
                    ),
                ]
            )
        )
        final_lockfile_digest = await merge_digests(
            MergeDigests([metadata_digest, result.output_digest])
        )
    else:
        initial_lockfile_digest_contents = await get_digest_contents(result.output_digest)
        lockfile_with_header = metadata.add_header_to_lockfile(
            initial_lockfile_digest_contents[0].content,
            regenerate_command=regenerate_command,
            delimeter=header_delimiter,
        )
        final_lockfile_digest = await create_digest(
            CreateDigest(
                [
                    FileContent(req.lockfile_dest, lockfile_with_header),
                ]
            )
        )

    if req.diff:
        diff = await _generate_python_lockfile_diff(
            final_lockfile_digest, req.resolve_name, req.lockfile_dest
        )
    else:
        diff = None

    return GenerateLockfileResult(final_lockfile_digest, req.resolve_name, req.lockfile_dest, diff)


class RequestedPythonUserResolveNames(RequestedUserResolveNames):
    pass


class KnownPythonUserResolveNamesRequest(KnownUserResolveNamesRequest):
    pass


@rule
async def determine_python_user_resolves(
    _: KnownPythonUserResolveNamesRequest,
    python_setup: PythonSetup,
    union_membership: UnionMembership,
) -> KnownUserResolveNames:
    """Find all know Python resolves, from both user-created resolves and internal tools."""
    python_tool_resolves = ExportableTool.filter_for_subclasses(union_membership, PythonToolBase)

    tools_using_default_resolve = [
        resolve_name
        for resolve_name, subsystem_cls in python_tool_resolves.items()
        if (await _construct_subsystem(subsystem_cls)).install_from_resolve is None
    ]

    return KnownUserResolveNames(
        names=(
            *python_setup.resolves.keys(),
            *tools_using_default_resolve,
        ),  # the order of the keys doesn't matter since shadowing is done in `setup_user_lockfile_requests`
        option_name="[python].resolves",
        requested_resolve_names_cls=RequestedPythonUserResolveNames,
    )


@rule
async def setup_user_lockfile_requests(
    requested: RequestedPythonUserResolveNames,
    all_targets: AllTargets,
    python_setup: PythonSetup,
    union_membership: UnionMembership,
) -> UserGenerateLockfiles:
    """Transform the names of resolves requested into the `GeneratePythonLockfile` request object.

    Shadowing is done here by only checking internal resolves if the resolve is not a user-created
    resolve.
    """
    if not (python_setup.enable_resolves and python_setup.resolves_generate_lockfiles):
        return UserGenerateLockfiles()

    resolve_to_requirements_fields = defaultdict(set)
    find_links: set[str] = set()
    for tgt in all_targets:
        if not tgt.has_fields((PythonRequirementResolveField, PythonRequirementsField)):
            continue
        resolve = tgt[PythonRequirementResolveField].normalized_value(python_setup)
        resolve_to_requirements_fields[resolve].add(tgt[PythonRequirementsField])
        find_links.update(tgt[PythonRequirementFindLinksField].value or ())

    tools = ExportableTool.filter_for_subclasses(union_membership, PythonToolBase)

    out = set()
    for resolve in requested:
        if resolve in python_setup.resolves:
            out.add(
                GeneratePythonLockfile(
                    requirements=PexRequirements.req_strings_from_requirement_fields(
                        resolve_to_requirements_fields[resolve]
                    ),
                    find_links=FrozenOrderedSet(find_links),
                    interpreter_constraints=InterpreterConstraints(
                        python_setup.resolves_to_interpreter_constraints.get(
                            resolve, python_setup.interpreter_constraints
                        )
                    ),
                    resolve_name=resolve,
                    lockfile_dest=python_setup.resolves[resolve],
                    diff=False,
                )
            )
        else:
            tool_cls: type[PythonToolBase] = tools[resolve]
            tool = await _construct_subsystem(tool_cls)

            # TODO: we shouldn't be managing default ICs in lockfile identification.
            #   We should find a better place to do this or a better way to default
            if tool.register_interpreter_constraints:
                ic = tool.interpreter_constraints
            else:
                ic = InterpreterConstraints(tool.default_interpreter_constraints)

            out.add(
                GeneratePythonLockfile(
                    requirements=FrozenOrderedSet(sorted(tool.requirements)),
                    find_links=FrozenOrderedSet(find_links),
                    interpreter_constraints=ic,
                    resolve_name=resolve,
                    lockfile_dest=DEFAULT_TOOL_LOCKFILE,
                    diff=False,
                )
            )

    return UserGenerateLockfiles(out)


@dataclass(frozen=True)
class PythonSyntheticLockfileTargetsRequest(SyntheticTargetsRequest):
    """Register the type used to create synthetic targets for Python lockfiles.

    As the paths for all lockfiles are known up-front, we set the `path` field to
    `SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS` so that we get a single request for all
    our synthetic targets rather than one request per directory.
    """

    path: str = SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS


def synthetic_lockfile_target_name(resolve: str) -> str:
    return f"_{resolve}_lockfile"


@rule
async def python_lockfile_synthetic_targets(
    request: PythonSyntheticLockfileTargetsRequest,
    python_setup: PythonSetup,
) -> SyntheticAddressMaps:
    if not python_setup.enable_synthetic_lockfiles:
        return SyntheticAddressMaps()

    resolves = [
        (os.path.dirname(lockfile), os.path.basename(lockfile), name)
        for name, lockfile in python_setup.resolves.items()
    ]

    return SyntheticAddressMaps.for_targets_request(
        request,
        [
            (
                os.path.join(spec_path, "BUILD.python-lockfiles"),
                tuple(
                    TargetAdaptor(
                        "_lockfiles",
                        name=synthetic_lockfile_target_name(name),
                        sources=(lockfile,),
                        __description_of_origin__=f"the [python].resolves option {name!r}",
                    )
                    for _, lockfile, name in lockfiles
                ),
            )
            for spec_path, lockfiles in itertools.groupby(sorted(resolves), key=itemgetter(0))
        ],
    )


def rules():
    return (
        *collect_rules(),
        UnionRule(GenerateLockfile, GeneratePythonLockfile),
        UnionRule(KnownUserResolveNamesRequest, KnownPythonUserResolveNamesRequest),
        UnionRule(RequestedUserResolveNames, RequestedPythonUserResolveNames),
        UnionRule(SyntheticTargetsRequest, PythonSyntheticLockfileTargetsRequest),
    )

1	# Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	12✔
5
6	import itertools	12✔
7	import os.path	12✔
8	from collections import defaultdict	12✔
9	from dataclasses import dataclass	12✔
10	from operator import itemgetter	12✔
11
12	from pants.backend.python.subsystems.python_tool_base import PythonToolBase	12✔
13	from pants.backend.python.subsystems.setup import PythonSetup	12✔
14	from pants.backend.python.target_types import (	12✔
15	PythonRequirementFindLinksField,
16	PythonRequirementResolveField,
17	PythonRequirementsField,
18	)
19	from pants.backend.python.util_rules.interpreter_constraints import InterpreterConstraints	12✔
20	from pants.backend.python.util_rules.lockfile_diff import _generate_python_lockfile_diff	12✔
21	from pants.backend.python.util_rules.lockfile_metadata import PythonLockfileMetadata	12✔
22	from pants.backend.python.util_rules.pex import find_interpreter	12✔
23	from pants.backend.python.util_rules.pex_cli import PexCliProcess, maybe_log_pex_stderr	12✔
24	from pants.backend.python.util_rules.pex_environment import PexSubsystem	12✔
25	from pants.backend.python.util_rules.pex_requirements import (	12✔
26	PexRequirements,
27	ResolvePexConfig,
28	ResolvePexConfigRequest,
29	determine_resolve_pex_config,
30	)
31	from pants.core.goals.generate_lockfiles import (	12✔
32	DEFAULT_TOOL_LOCKFILE,
33	GenerateLockfile,
34	GenerateLockfileResult,
35	GenerateLockfilesSubsystem,
36	KnownUserResolveNames,
37	KnownUserResolveNamesRequest,
38	RequestedUserResolveNames,
39	UserGenerateLockfiles,
40	WrappedGenerateLockfile,
41	)
42	from pants.core.goals.resolves import ExportableTool	12✔
43	from pants.core.util_rules.lockfile_metadata import calculate_invalidation_digest	12✔
44	from pants.engine.fs import CreateDigest, Digest, FileContent, MergeDigests	12✔
45	from pants.engine.internals.synthetic_targets import SyntheticAddressMaps, SyntheticTargetsRequest	12✔
46	from pants.engine.internals.target_adaptor import TargetAdaptor	12✔
47	from pants.engine.intrinsics import create_digest, get_digest_contents, merge_digests	12✔
48	from pants.engine.process import ProcessCacheScope, fallible_to_exec_result_or_raise	12✔
49	from pants.engine.rules import collect_rules, implicitly, rule	12✔
50	from pants.engine.target import AllTargets	12✔
51	from pants.engine.unions import UnionMembership, UnionRule	12✔
52	from pants.option.subsystem import _construct_subsystem	12✔
53	from pants.util.docutil import bin_name	12✔
54	from pants.util.logging import LogLevel	12✔
55	from pants.util.ordered_set import FrozenOrderedSet	12✔
56	from pants.util.pip_requirement import PipRequirement	12✔
57
58
59	@dataclass(frozen=True)	12✔
60	class GeneratePythonLockfile(GenerateLockfile):	12✔
61	requirements: FrozenOrderedSet[str]	12✔
62	find_links: FrozenOrderedSet[str]	12✔
63	interpreter_constraints: InterpreterConstraints	12✔
64
65	@property	12✔
66	def requirements_hex_digest(self) -> str:	12✔
67	"""Produces a hex digest of the requirements input for this lockfile."""
68	return calculate_invalidation_digest(self.requirements)	×
69
70
71	@rule	12✔
72	async def wrap_python_lockfile_request(request: GeneratePythonLockfile) -> WrappedGenerateLockfile:	12✔
73	return WrappedGenerateLockfile(request)	×
74
75
76	@dataclass(frozen=True)	12✔
77	class _PipArgsAndConstraintsSetup:	12✔
78	resolve_config: ResolvePexConfig	12✔
79	args: tuple[str, ...]	12✔
80	digest: Digest	12✔
81
82
83	async def _setup_pip_args_and_constraints_file(resolve_name: str) -> _PipArgsAndConstraintsSetup:	12✔
84	resolve_config = await determine_resolve_pex_config(	×
85	ResolvePexConfigRequest(resolve_name), **implicitly()
86	)
87
88	args = list(resolve_config.pex_args())	×
89	digests: list[Digest] = []	×
90
91	if resolve_config.constraints_file:	×
92	args.append(f"--constraints={resolve_config.constraints_file.path}")	×
93	digests.append(resolve_config.constraints_file.digest)	×
94
95	input_digest = await merge_digests(MergeDigests(digests))	×
96	return _PipArgsAndConstraintsSetup(resolve_config, tuple(args), input_digest)	×
97
98
99	@rule(desc="Generate Python lockfile", level=LogLevel.DEBUG)	12✔
100	async def generate_lockfile(	12✔
101	req: GeneratePythonLockfile,
102	generate_lockfiles_subsystem: GenerateLockfilesSubsystem,
103	python_setup: PythonSetup,
104	pex_subsystem: PexSubsystem,
105	) -> GenerateLockfileResult:
106	if not req.requirements:	×
107	raise ValueError(	×
108	f"Cannot generate lockfile with no requirements. Please add some requirements to {req.resolve_name}."
109	)
110
111	pip_args_setup = await _setup_pip_args_and_constraints_file(req.resolve_name)	×
112	header_delimiter = "//"	×
113
114	python = await find_interpreter(req.interpreter_constraints, **implicitly())	×
115
116	result = await fallible_to_exec_result_or_raise(	×
117	**implicitly(
118	PexCliProcess(
119	subcommand=("lock", "create"),
120	extra_args=(
121	f"--output={req.lockfile_dest}",
122	# See https://github.com/pantsbuild/pants/issues/12458. For now, we always
123	# generate universal locks because they have the best compatibility. We may
124	# want to let users change this, as `style=strict` is safer.
125	"--style=universal",
126	"--pip-version",
127	python_setup.pip_version,
128	"--resolver-version",
129	"pip-2020-resolver",
130	"--preserve-pip-download-log",
131	"pex-pip-download.log",
132	# PEX files currently only run on Linux and Mac machines; so we hard code this
133	# limit on lock universality to avoid issues locking due to irrelevant
134	# Windows-only dependency issues. See this Pex issue that originated from a
135	# Pants user issue presented in Slack:
136	# https://github.com/pex-tool/pex/issues/1821
137	#
138	# At some point it will probably make sense to expose `--target-system` for
139	# configuration.
140	"--target-system",
141	"linux",
142	"--target-system",
143	"mac",
144	# This makes diffs more readable when lockfiles change.
145	"--indent=2",
146	f"--python-path={python.path}",
147	*(f"--find-links={link}" for link in req.find_links),
148	*pip_args_setup.args,
149	*req.interpreter_constraints.generate_pex_arg_list(),
150	*req.requirements,
151	),
152	additional_input_digest=pip_args_setup.digest,
153	output_files=(req.lockfile_dest,),
154	description=f"Generate lockfile for {req.resolve_name}",
155	# Instead of caching lockfile generation with LMDB, we instead use the invalidation
156	# scheme from `lockfile_metadata.py` to check for stale/invalid lockfiles. This is
157	# necessary so that our invalidation is resilient to deleting LMDB or running on a
158	# new machine.
159	#
160	# We disable caching with LMDB so that when you generate a lockfile, you always get
161	# the most up-to-date snapshot of the world. This is generally desirable and also
162	# necessary to avoid an awkward edge case where different developers generate
163	# different lockfiles even when generating at the same time. See
164	# https://github.com/pantsbuild/pants/issues/12591.
165	cache_scope=ProcessCacheScope.PER_SESSION,
166	)
167	)
168	)
UNCOV 169	maybe_log_pex_stderr(result.stderr, pex_subsystem.verbosity)	×
170
171	metadata = PythonLockfileMetadata.new(	×
172	valid_for_interpreter_constraints=req.interpreter_constraints,
173	requirements={
174	PipRequirement.parse(
175	i,
176	description_of_origin=f"the lockfile {req.lockfile_dest} for the resolve {req.resolve_name}",
177	)
178	for i in req.requirements
179	},
180	manylinux=pip_args_setup.resolve_config.manylinux,
181	requirement_constraints=(
182	set(pip_args_setup.resolve_config.constraints_file.constraints)
183	if pip_args_setup.resolve_config.constraints_file
184	else set()
185	),
186	only_binary=set(pip_args_setup.resolve_config.only_binary),
187	no_binary=set(pip_args_setup.resolve_config.no_binary),
188	excludes=set(pip_args_setup.resolve_config.excludes),
189	overrides=set(pip_args_setup.resolve_config.overrides),
190	)
NEW 191	regenerate_command = (	×
192	generate_lockfiles_subsystem.custom_command
193	or f"{bin_name()} generate-lockfiles --resolve={req.resolve_name}"
194	)
NEW 195	if python_setup.separate_lockfile_metadata_file:	×
NEW 196	descr = f"This lockfile was generated by Pants. To regenerate, run: {regenerate_command}"	×
NEW 197	metadata_digest = await create_digest(	×
198	CreateDigest(
199	[
200	FileContent(
201	PythonLockfileMetadata.metadata_location_for_lockfile(req.lockfile_dest),
202	metadata.to_json(with_description=descr).encode(),
203	),
204	]
205	)
206	)
NEW 207	final_lockfile_digest = await merge_digests(	×
208	MergeDigests([metadata_digest, result.output_digest])
209	)
210	else:
NEW 211	initial_lockfile_digest_contents = await get_digest_contents(result.output_digest)	×
NEW 212	lockfile_with_header = metadata.add_header_to_lockfile(	×
213	initial_lockfile_digest_contents[0].content,
214	regenerate_command=regenerate_command,
215	delimeter=header_delimiter,
216	)
NEW 217	final_lockfile_digest = await create_digest(	×
218	CreateDigest(
219	[
220	FileContent(req.lockfile_dest, lockfile_with_header),
221	]
222	)
223	)
224
225	if req.diff:	×
226	diff = await _generate_python_lockfile_diff(	×
227	final_lockfile_digest, req.resolve_name, req.lockfile_dest
228	)
229	else:
230	diff = None	×
231
232	return GenerateLockfileResult(final_lockfile_digest, req.resolve_name, req.lockfile_dest, diff)	×
233
234
235	class RequestedPythonUserResolveNames(RequestedUserResolveNames):	12✔
236	pass	12✔
237
238
239	class KnownPythonUserResolveNamesRequest(KnownUserResolveNamesRequest):	12✔
240	pass	12✔
241
242
243	@rule	12✔
244	async def determine_python_user_resolves(	12✔
245	_: KnownPythonUserResolveNamesRequest,
246	python_setup: PythonSetup,
247	union_membership: UnionMembership,
248	) -> KnownUserResolveNames:
249	"""Find all know Python resolves, from both user-created resolves and internal tools."""
250	python_tool_resolves = ExportableTool.filter_for_subclasses(union_membership, PythonToolBase)	×
251
252	tools_using_default_resolve = [	×
253	resolve_name
254	for resolve_name, subsystem_cls in python_tool_resolves.items()
255	if (await _construct_subsystem(subsystem_cls)).install_from_resolve is None
256	]
257
258	return KnownUserResolveNames(	×
259	names=(
260	*python_setup.resolves.keys(),
261	*tools_using_default_resolve,
262	), # the order of the keys doesn't matter since shadowing is done in `setup_user_lockfile_requests`
263	option_name="[python].resolves",
264	requested_resolve_names_cls=RequestedPythonUserResolveNames,
265	)
266
267
268	@rule	12✔
269	async def setup_user_lockfile_requests(	12✔
270	requested: RequestedPythonUserResolveNames,
271	all_targets: AllTargets,
272	python_setup: PythonSetup,
273	union_membership: UnionMembership,
274	) -> UserGenerateLockfiles:
275	"""Transform the names of resolves requested into the `GeneratePythonLockfile` request object.
276
277	Shadowing is done here by only checking internal resolves if the resolve is not a user-created
278	resolve.
279	"""
280	if not (python_setup.enable_resolves and python_setup.resolves_generate_lockfiles):	×
281	return UserGenerateLockfiles()	×
282
283	resolve_to_requirements_fields = defaultdict(set)	×
284	find_links: set[str] = set()	×
285	for tgt in all_targets:	×
286	if not tgt.has_fields((PythonRequirementResolveField, PythonRequirementsField)):	×
287	continue	×
288	resolve = tgt[PythonRequirementResolveField].normalized_value(python_setup)	×
289	resolve_to_requirements_fields[resolve].add(tgt[PythonRequirementsField])	×
290	find_links.update(tgt[PythonRequirementFindLinksField].value or ())	×
291
292	tools = ExportableTool.filter_for_subclasses(union_membership, PythonToolBase)	×
293
294	out = set()	×
295	for resolve in requested:	×
296	if resolve in python_setup.resolves:	×
297	out.add(	×
298	GeneratePythonLockfile(
299	requirements=PexRequirements.req_strings_from_requirement_fields(
300	resolve_to_requirements_fields[resolve]
301	),
302	find_links=FrozenOrderedSet(find_links),
303	interpreter_constraints=InterpreterConstraints(
304	python_setup.resolves_to_interpreter_constraints.get(
305	resolve, python_setup.interpreter_constraints
306	)
307	),
308	resolve_name=resolve,
309	lockfile_dest=python_setup.resolves[resolve],
310	diff=False,
311	)
312	)
313	else:
314	tool_cls: type[PythonToolBase] = tools[resolve]	×
315	tool = await _construct_subsystem(tool_cls)	×
316
317	# TODO: we shouldn't be managing default ICs in lockfile identification.
318	# We should find a better place to do this or a better way to default
319	if tool.register_interpreter_constraints:	×
320	ic = tool.interpreter_constraints	×
321	else:
322	ic = InterpreterConstraints(tool.default_interpreter_constraints)	×
323
324	out.add(	×
325	GeneratePythonLockfile(
326	requirements=FrozenOrderedSet(sorted(tool.requirements)),
327	find_links=FrozenOrderedSet(find_links),
328	interpreter_constraints=ic,
329	resolve_name=resolve,
330	lockfile_dest=DEFAULT_TOOL_LOCKFILE,
331	diff=False,
332	)
333	)
334
335	return UserGenerateLockfiles(out)	×
336
337
338	@dataclass(frozen=True)	12✔
339	class PythonSyntheticLockfileTargetsRequest(SyntheticTargetsRequest):	12✔
340	"""Register the type used to create synthetic targets for Python lockfiles.
341
342	As the paths for all lockfiles are known up-front, we set the `path` field to
343	`SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS` so that we get a single request for all
344	our synthetic targets rather than one request per directory.
345	"""
346
347	path: str = SyntheticTargetsRequest.SINGLE_REQUEST_FOR_ALL_TARGETS	12✔
348
349
350	def synthetic_lockfile_target_name(resolve: str) -> str:	12✔
351	return f"_{resolve}_lockfile"	×
352
353
354	@rule	12✔
355	async def python_lockfile_synthetic_targets(	12✔
356	request: PythonSyntheticLockfileTargetsRequest,
357	python_setup: PythonSetup,
358	) -> SyntheticAddressMaps:
359	if not python_setup.enable_synthetic_lockfiles:	×
360	return SyntheticAddressMaps()	×
361
362	resolves = [	×
363	(os.path.dirname(lockfile), os.path.basename(lockfile), name)
364	for name, lockfile in python_setup.resolves.items()
365	]
366
367	return SyntheticAddressMaps.for_targets_request(	×
368	request,
369	[
370	(
371	os.path.join(spec_path, "BUILD.python-lockfiles"),
372	tuple(
373	TargetAdaptor(
374	"_lockfiles",
375	name=synthetic_lockfile_target_name(name),
376	sources=(lockfile,),
377	__description_of_origin__=f"the [python].resolves option {name!r}",
378	)
379	for _, lockfile, name in lockfiles
380	),
381	)
382	for spec_path, lockfiles in itertools.groupby(sorted(resolves), key=itemgetter(0))
383	],
384	)
385
386
387	def rules():	12✔
388	return (	11✔
389	*collect_rules(),
390	UnionRule(GenerateLockfile, GeneratePythonLockfile),
391	UnionRule(KnownUserResolveNamesRequest, KnownPythonUserResolveNamesRequest),
392	UnionRule(RequestedUserResolveNames, RequestedPythonUserResolveNames),
393	UnionRule(SyntheticTargetsRequest, PythonSyntheticLockfileTargetsRequest),
394	)

pantsbuild / pants / 18143316655

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