20147226056

Committed 11 Dec 2025 08:58PM UTC coverage: 78.827% (-1.5%) from 80.293%

Build # 20147226056

Build Type

push

github

Committed by

web-flow

Commit Message

Forwarded the `style` and `complete-platform` args from pants.toml to PEX (#22910)

## Context

After Apple switched to the `arm64` architecture, some package
publishers stopped releasing `x86_64` variants of their packages for
`darwin`. As a result, generating a universal lockfile now fails because
no single package version is compatible with both `x86_64` and `arm64`
on `darwin`.

The solution is to use the `--style` and `--complete-platform` flags
with PEX. For example:
```
pex3 lock create \
    --style strict \
    --complete-platform 3rdparty/platforms/manylinux_2_28_aarch64.json \
    --complete-platform 3rdparty/platforms/macosx_26_0_arm64.json \
    -r 3rdparty/python/requirements_pyarrow.txt \
    -o python-pyarrow.lock
```

See the Slack discussion here:
https://pantsbuild.slack.com/archives/C046T6T9U/p1760098582461759

## Reproduction

* `BUILD`
```
python_requirement(
    name="awswrangler",
    requirements=["awswrangler==3.12.1"],
    resolve="awswrangler",
)
```
* Run `pants generate-lockfiles --resolve=awswrangler` on macOS with an
`arm64` CPU
```
pip: ERROR: Cannot install awswrangler==3.12.1 because these package versions have conflicting dependencies.
pip: ERROR: ResolutionImpossible: for help visit https://pip.pypa.io/en/latest/topics/dependency-resolution/#dealing-with-dependency-conflicts
pip:  
pip:  The conflict is caused by:
pip:      awswrangler 3.12.1 depends on pyarrow<18.0.0 and >=8.0.0; sys_platform == "darwin" and platform_machine == "x86_64"
pip:      awswrangler 3.12.1 depends on pyarrow<21.0.0 and >=18.0.0; sys_platform != "darwin" or platform_machine != "x86_64"
pip:  
pip:  Additionally, some packages in these conflicts have no matching distributions available for your environment:
pip:      pyarrow
pip:  
pip:  To fix this you could try to:
pip:  1. loosen the range of package versions you've specified
pip:  2. remove package versions to allow pip to attempt to solve the dependency conflict
```

## Implementation
... (continued)

Run Details

77 of 100 new or added lines in 6 files covered. (77.0%)

868 existing lines in 42 files now uncovered.

74471 of 94474 relevant lines covered (78.83%)

3.18 hits per line

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

61.24

/src/python/pants/core/util_rules/external_tool.py

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

from __future__ import annotations

import dataclasses
import logging
import os
import textwrap
from abc import ABCMeta, abstractmethod
from dataclasses import dataclass
from enum import Enum

from packaging.requirements import Requirement

from pants.core.goals.export import (
    ExportedBinary,
    ExportRequest,
    ExportResult,
    ExportResults,
    ExportSubsystem,
)
from pants.core.goals.resolves import ExportableTool, ExportMode
from pants.core.util_rules import archive
from pants.core.util_rules.archive import maybe_extract_archive
from pants.engine.download_file import download_file
from pants.engine.engine_aware import EngineAwareParameter
from pants.engine.fs import CreateDigest, Digest, DownloadFile, FileDigest, FileEntry
from pants.engine.internals.selectors import concurrently
from pants.engine.intrinsics import create_digest, get_digest_entries
from pants.engine.platform import Platform
from pants.engine.rules import collect_rules, implicitly, rule
from pants.engine.unions import UnionMembership, UnionRule
from pants.option.option_types import DictOption, EnumOption, StrListOption, StrOption
from pants.option.subsystem import Subsystem, _construct_subsystem
from pants.util.docutil import doc_url
from pants.util.logging import LogLevel
from pants.util.meta import classproperty
from pants.util.strutil import softwrap

logger = logging.getLogger(__name__)


class UnknownVersion(Exception):
    pass


class ExternalToolError(Exception):
    pass


class UnsupportedVersion(ExternalToolError):
    """The specified version of the tool is not supported, according to the given version
    constraints."""


class UnsupportedVersionUsage(Enum):
    """What action to take in case the requested version of the tool is not supported."""

    RaiseError = "error"
    LogWarning = "warning"


@dataclass(frozen=True)
class ExternalToolRequest:
    download_file_request: DownloadFile
    exe: str


@dataclass(frozen=True)
class DownloadedExternalTool:
    digest: Digest
    exe: str


@dataclass(frozen=True)
class ExternalToolVersion:
    version: str
    platform: str
    sha256: str
    filesize: int
    url_override: str | None = None

    def encode(self) -> str:
        parts = [self.version, self.platform, self.sha256, str(self.filesize)]
        if self.url_override:
            parts.append(self.url_override)
        return "|".join(parts)

    @classmethod
    def decode(cls, version_str: str) -> ExternalToolVersion:
        parts = [x.strip() for x in version_str.split("|")]
        version, platform, sha256, filesize = parts[:4]
        url_override = parts[4] if len(parts) > 4 else None
        return cls(version, platform, sha256, int(filesize), url_override=url_override)


class ExternalToolOptionsMixin:
    """Common options for implementing subsystem providing an `ExternalToolRequest`."""

    @classproperty
    def name(cls):
        """The name of the tool, for use in user-facing messages.

        Derived from the classname, but subclasses can override, e.g., with a classproperty.
        """
        return cls.__name__.lower()

    @classproperty
    def binary_name(cls):
        """The name of the binary, as it normally known.

        This allows renaming a built binary to what users expect, even when the name is different.
        For example, the binary might be "taplo-linux-x86_64" and the name "Taplo", but users expect
        just "taplo".
        """
        return cls.name.lower()

    # The default values for --version and --known-versions, and the supported versions.
    # Subclasses must set appropriately.
    default_version: str
    default_known_versions: list[str]
    version_constraints: str | None = None

    version = StrOption(
        default=lambda cls: cls.default_version,
        advanced=True,
        help=lambda cls: f"Use this version of {cls.name}."
        + (
            f"\n\nSupported {cls.name} versions: {cls.version_constraints}"
            if cls.version_constraints
            else ""
        ),
    )

    # Note that you can compute the length and sha256 conveniently with:
    #   `curl -L $URL | tee >(wc -c) >(shasum -a 256) >/dev/null`
    known_versions = StrListOption(
        default=lambda cls: cls.default_known_versions,
        advanced=True,
        help=textwrap.dedent(
            f"""
        Known versions to verify downloads against.

        Each element is a pipe-separated string of `version|platform|sha256|length` or
        `version|platform|sha256|length|url_override`, where:

          - `version` is the version string
          - `platform` is one of `[{",".join(Platform.__members__.keys())}]`
          - `sha256` is the 64-character hex representation of the expected sha256
            digest of the download file, as emitted by `shasum -a 256`
          - `length` is the expected length of the download file in bytes, as emitted by
            `wc -c`
          - (Optional) `url_override` is a specific url to use instead of the normally
            generated url for this version

        E.g., `3.1.2|macos_x86_64|6d0f18cd84b918c7b3edd0203e75569e0c7caecb1367bbbe409b44e28514f5be|42813`.
        and `3.1.2|macos_arm64 |aca5c1da0192e2fd46b7b55ab290a92c5f07309e7b0ebf4e45ba95731ae98291|50926|https://example.mac.org/bin/v3.1.2/mac-aarch64-v3.1.2.tgz`.

        Values are space-stripped, so pipes can be indented for readability if necessary.
        """
        ),
    )


class ExternalTool(Subsystem, ExportableTool, ExternalToolOptionsMixin, metaclass=ABCMeta):
    """Configuration for an invocable tool that we download from an external source.

    Subclass this to configure a specific tool.


    Idiomatic use:

    class MyExternalTool(ExternalTool):
        options_scope = "my-external-tool"
        default_version = "1.2.3"
        default_known_versions = [
          "1.2.3|linux_arm64 |feed6789feed6789feed6789feed6789feed6789feed6789feed6789feed6789|112233",
          "1.2.3|linux_x86_64|cafebabacafebabacafebabacafebabacafebabacafebabacafebabacafebaba|878986",
          "1.2.3|macos_arm64 |deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef|222222",
          "1.2.3|macos_x86_64|1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd|333333",
        ]

        version_constraints = ">=1.2.3, <2.0"

        def generate_url(self, plat: Platform) -> str:
            ...

        def generate_exe(self, plat: Platform) -> str:
            return "./path-to/binary

    @rule
    async def my_rule(my_external_tool: MyExternalTool, platform: Platform) -> Foo:
        downloaded_tool = await download_external_tool(my_external_tool.get_request(platform))
        ...
    """

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.check_version_constraints()

    export_mode = ExportMode.binary

    use_unsupported_version = EnumOption(
        advanced=True,
        help=lambda cls: textwrap.dedent(
            f"""
                What action to take in case the requested version of {cls.name} is not supported.

                Supported {cls.name} versions: {cls.version_constraints if cls.version_constraints else "unspecified"}
                """
        ),
        default=UnsupportedVersionUsage.RaiseError,
    )

    @abstractmethod
    def generate_url(self, plat: Platform) -> str:
        """Returns the URL for the given version of the tool, runnable on the given os+arch.

        Implementations should raise ExternalToolError if they cannot resolve the arguments
        to a URL. The raised exception need not have a message - a sensible one will be generated.
        """

    def generate_exe(self, plat: Platform) -> str:
        """Returns the path to the tool executable.

        If the downloaded artifact is the executable itself, you can leave this unimplemented.

        If the downloaded artifact is an archive, this should be overridden to provide a
        relative path in the downloaded archive, e.g. `./bin/protoc`.
        """
        return f"./{self.generate_url(plat).rsplit('/', 1)[-1]}"

    def known_version(self, plat: Platform) -> ExternalToolVersion | None:
        for known_version in self.known_versions:
            tool_version = self.decode_known_version(known_version)
            if plat.value == tool_version.platform and tool_version.version == self.version:
                return tool_version
        return None

    def get_request(self, plat: Platform) -> ExternalToolRequest:
        """Generate a request for this tool."""

        tool_version = self.known_version(plat)
        if tool_version:
            return self.get_request_for(
                tool_version.platform,
                tool_version.sha256,
                tool_version.filesize,
                url_override=tool_version.url_override,
            )
        raise UnknownVersion(
            softwrap(
                f"""
                No known version of {self.name} {self.version} for {plat.value} found in
                {self.known_versions}
                """
            )
        )

    @classmethod
    def decode_known_version(cls, known_version: str) -> ExternalToolVersion:
        try:
            return ExternalToolVersion.decode(known_version)
        except ValueError:
            raise ExternalToolError(
                f"Bad value for [{cls.options_scope}].known_versions: {known_version}"
            )

    @classmethod
    def split_known_version_str(cls, known_version: str) -> tuple[str, str, str, int]:
        version = cls.decode_known_version(known_version)
        return version.version, version.platform, version.sha256, version.filesize

    def get_request_for(
        self, plat_val: str, sha256: str, length: int, url_override: str | None = None
    ) -> ExternalToolRequest:
        """Generate a request for this tool from the given info."""
        plat = Platform(plat_val)
        digest = FileDigest(fingerprint=sha256, serialized_bytes_length=length)
        try:
            url = url_override or self.generate_url(plat)
            exe = self.generate_exe(plat)
        except ExternalToolError as e:
            raise ExternalToolError(
                f"Couldn't find {self.name} version {self.version} on {plat.value}"
            ) from e
        return ExternalToolRequest(DownloadFile(url=url, expected_digest=digest), exe)

    def check_version_constraints(self) -> None:
        if not self.version_constraints:
            return None
        # Note that this is not a Python requirement. We're just hackily piggybacking off
        # packaging.requirements.Requirement's ability to check version constraints.
        constraints = Requirement(f"{self.name}{self.version_constraints}")
        if constraints.specifier.contains(self.version):
            # all ok
            return None

        msg = [
            f"The option [{self.options_scope}].version is set to {self.version}, which is not "
            f"compatible with what this release of Pants expects: {constraints}.",
            "Please update the version to a supported value, or consider using a different Pants",
            "release if you cannot change the version.",
        ]

        if self.use_unsupported_version is UnsupportedVersionUsage.LogWarning:
            msg.extend(
                [
                    "Alternatively, you can ignore this warning (at your own peril) by adding this",
                    "to the GLOBAL section of pants.toml:",
                    f'ignore_warnings = ["The option [{self.options_scope}].version is set to"].',
                ]
            )
            logger.warning(" ".join(msg))
        elif self.use_unsupported_version is UnsupportedVersionUsage.RaiseError:
            msg.append(
                softwrap(
                    f"""
                Alternatively, update [{self.options_scope}].use_unsupported_version to be
                'warning'.
                """
                )
            )
            raise UnsupportedVersion(" ".join(msg))


class TemplatedExternalToolOptionsMixin(ExternalToolOptionsMixin):
    """Common options for implementing a subsystem providing an `ExternalToolRequest` via a URL
    template."""

    default_url_template: str
    default_url_platform_mapping: dict[str, str] | None = None

    url_template = StrOption(
        default=lambda cls: cls.default_url_template,
        advanced=True,
        help=softwrap(
            f"""
            URL to download the tool, either as a single binary file or a compressed file
            (e.g. zip file). You can change this to point to your own hosted file, e.g. to
            work with proxies or for access via the filesystem through a `file:$abspath` URL (e.g.
            `file:/this/is/absolute`, possibly by
            [templating the buildroot in a config file]({doc_url("docs/using-pants/key-concepts/options#config-file-entries")})).

            Use `{{version}}` to have the value from `--version` substituted, and `{{platform}}` to
            have a value from `--url-platform-mapping` substituted in, depending on the
            current platform. For example,
            https://github.com/.../protoc-{{version}}-{{platform}}.zip.
            """
        ),
    )

    url_platform_mapping = DictOption[str](
        "--url-platform-mapping",
        default=lambda cls: cls.default_url_platform_mapping,
        advanced=True,
        help=softwrap(
            """
            A dictionary mapping platforms to strings to be used when generating the URL
            to download the tool.

            In `--url-template`, anytime the `{platform}` string is used, Pants will determine the
            current platform, and substitute `{platform}` with the respective value from your dictionary.

            For example, if you define `{"macos_x86_64": "apple-darwin", "linux_x86_64": "unknown-linux"}`,
            and run Pants on Linux with an intel architecture, then `{platform}` will be substituted
            in the `--url-template` option with `unknown-linux`.
            """
        ),
    )


class TemplatedExternalTool(ExternalTool, TemplatedExternalToolOptionsMixin):
    """Extends the ExternalTool to allow url templating for custom/self-hosted source.

    In addition to ExternalTool functionalities, it is needed to set, e.g.:

    default_url_template = "https://tool.url/{version}/{platform}-mytool.zip"
    default_url_platform_mapping = {
        "macos_x86_64": "osx_intel",
        "macos_arm64": "osx_arm",
        "linux_x86_64": "linux",
    }

    The platform mapping dict is optional.
    """

    def generate_url(self, plat: Platform) -> str:
        platform = self.url_platform_mapping.get(plat.value, "")
        return self.url_template.format(version=self.version, platform=platform)


@rule(level=LogLevel.DEBUG)
async def download_external_tool(request: ExternalToolRequest) -> DownloadedExternalTool:
    # Download and extract.
    maybe_archive_digest = await download_file(request.download_file_request, **implicitly())
    extracted_archive = await maybe_extract_archive(**implicitly(maybe_archive_digest))

    # Confirm executable.
    exe_path = request.exe.lstrip("./")
    digest = extracted_archive.digest
    is_not_executable = False
    digest_entries = []
    for entry in await get_digest_entries(digest):
        if isinstance(entry, FileEntry) and entry.path == exe_path and not entry.is_executable:
            # We should recreate the digest with the executable bit set.
            is_not_executable = True
            entry = dataclasses.replace(entry, is_executable=True)
        digest_entries.append(entry)
    if is_not_executable:
        digest = await create_digest(CreateDigest(digest_entries))

    return DownloadedExternalTool(digest, request.exe)


@dataclass(frozen=True)
class ExportExternalToolRequest(ExportRequest):
    pass
    # tool: type[ExternalTool]


@dataclass(frozen=True)
class _ExportExternalToolForResolveRequest(EngineAwareParameter):
    resolve: str


@dataclass(frozen=True)
class MaybeExportResult:
    result: ExportResult | None


@rule(level=LogLevel.DEBUG)
async def export_external_tool(
    req: _ExportExternalToolForResolveRequest, platform: Platform, union_membership: UnionMembership
) -> MaybeExportResult:
    """Export a downloadable tool. Downloads all the tools to `bins`, and symlinks the primary exe
    to the `bin` directory.

    We use the last segment of the exe instead of the resolve because:
    - it's probably the exe name people expect
    - avoids clutter from the resolve name (ex "tfsec" instead of "terraform-tfsec")
    """
    exportables = ExportableTool.filter_for_subclasses(
        union_membership,
        ExternalTool,  # type:ignore[type-abstract]  # ExternalTool is abstract, and mypy doesn't like that we might return it
    )
    maybe_exportable = exportables.get(req.resolve)
    if not maybe_exportable:
        return MaybeExportResult(None)

    tool = await _construct_subsystem(maybe_exportable)
    downloaded_tool = await download_external_tool(tool.get_request(platform))

    dest = os.path.join("bins", tool.name)

    exe = tool.generate_exe(platform)
    return MaybeExportResult(
        ExportResult(
            description=f"Export tool {req.resolve}",
            reldir=dest,
            digest=downloaded_tool.digest,
            resolve=req.resolve,
            exported_binaries=(ExportedBinary(name=tool.binary_name, path_in_export=exe),),
        )
    )


@rule
async def export_external_tools(
    request: ExportExternalToolRequest, export: ExportSubsystem
) -> ExportResults:
    maybe_tools = await concurrently(
        export_external_tool(_ExportExternalToolForResolveRequest(resolve), **implicitly())
        for resolve in export.binaries
    )
    return ExportResults(tool.result for tool in maybe_tools if tool.result is not None)


def rules():
    return (
        *collect_rules(),
        *archive.rules(),
        UnionRule(ExportRequest, ExportExternalToolRequest),
    )

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	11✔
5
6	import dataclasses	11✔
7	import logging	11✔
8	import os	11✔
9	import textwrap	11✔
10	from abc import ABCMeta, abstractmethod	11✔
11	from dataclasses import dataclass	11✔
12	from enum import Enum	11✔
13
14	from packaging.requirements import Requirement	11✔
15
16	from pants.core.goals.export import (	11✔
17	ExportedBinary,
18	ExportRequest,
19	ExportResult,
20	ExportResults,
21	ExportSubsystem,
22	)
23	from pants.core.goals.resolves import ExportableTool, ExportMode	11✔
24	from pants.core.util_rules import archive	11✔
25	from pants.core.util_rules.archive import maybe_extract_archive	11✔
26	from pants.engine.download_file import download_file	11✔
27	from pants.engine.engine_aware import EngineAwareParameter	11✔
28	from pants.engine.fs import CreateDigest, Digest, DownloadFile, FileDigest, FileEntry	11✔
29	from pants.engine.internals.selectors import concurrently	11✔
30	from pants.engine.intrinsics import create_digest, get_digest_entries	11✔
31	from pants.engine.platform import Platform	11✔
32	from pants.engine.rules import collect_rules, implicitly, rule	11✔
33	from pants.engine.unions import UnionMembership, UnionRule	11✔
34	from pants.option.option_types import DictOption, EnumOption, StrListOption, StrOption	11✔
35	from pants.option.subsystem import Subsystem, _construct_subsystem	11✔
36	from pants.util.docutil import doc_url	11✔
37	from pants.util.logging import LogLevel	11✔
38	from pants.util.meta import classproperty	11✔
39	from pants.util.strutil import softwrap	11✔
40
41	logger = logging.getLogger(__name__)	11✔
42
43
44	class UnknownVersion(Exception):	11✔
45	pass	11✔
46
47
48	class ExternalToolError(Exception):	11✔
49	pass	11✔
50
51
52	class UnsupportedVersion(ExternalToolError):	11✔
53	"""The specified version of the tool is not supported, according to the given version
54	constraints."""
55
56
57	class UnsupportedVersionUsage(Enum):	11✔
58	"""What action to take in case the requested version of the tool is not supported."""
59
60	RaiseError = "error"	11✔
61	LogWarning = "warning"	11✔
62
63
64	@dataclass(frozen=True)	11✔
65	class ExternalToolRequest:	11✔
66	download_file_request: DownloadFile	11✔
67	exe: str	11✔
68
69
70	@dataclass(frozen=True)	11✔
71	class DownloadedExternalTool:	11✔
72	digest: Digest	11✔
73	exe: str	11✔
74
75
76	@dataclass(frozen=True)	11✔
77	class ExternalToolVersion:	11✔
78	version: str	11✔
79	platform: str	11✔
80	sha256: str	11✔
81	filesize: int	11✔
82	url_override: str \| None = None	11✔
83
84	def encode(self) -> str:	11✔
85	parts = [self.version, self.platform, self.sha256, str(self.filesize)]	×
86	if self.url_override:	×
87	parts.append(self.url_override)	×
88	return "\|".join(parts)	×
89
90	@classmethod	11✔
91	def decode(cls, version_str: str) -> ExternalToolVersion:	11✔
92	parts = [x.strip() for x in version_str.split("\|")]	2✔
93	version, platform, sha256, filesize = parts[:4]	2✔
94	url_override = parts[4] if len(parts) > 4 else None	2✔
95	return cls(version, platform, sha256, int(filesize), url_override=url_override)	2✔
96
97
98	class ExternalToolOptionsMixin:	11✔
99	"""Common options for implementing subsystem providing an `ExternalToolRequest`."""
100
101	@classproperty	11✔
102	def name(cls):	11✔
103	"""The name of the tool, for use in user-facing messages.
104
105	Derived from the classname, but subclasses can override, e.g., with a classproperty.
106	"""
107	return cls.__name__.lower()	11✔
108
109	@classproperty	11✔
110	def binary_name(cls):	11✔
111	"""The name of the binary, as it normally known.
112
113	This allows renaming a built binary to what users expect, even when the name is different.
114	For example, the binary might be "taplo-linux-x86_64" and the name "Taplo", but users expect
115	just "taplo".
116	"""
117	return cls.name.lower()	11✔
118
119	# The default values for --version and --known-versions, and the supported versions.
120	# Subclasses must set appropriately.
121	default_version: str	11✔
122	default_known_versions: list[str]	11✔
123	version_constraints: str \| None = None	11✔
124
125	version = StrOption(	11✔
126	default=lambda cls: cls.default_version,
127	advanced=True,
128	help=lambda cls: f"Use this version of {cls.name}."
129	+ (
130	f"\n\nSupported {cls.name} versions: {cls.version_constraints}"
131	if cls.version_constraints
132	else ""
133	),
134	)
135
136	# Note that you can compute the length and sha256 conveniently with:
137	# `curl -L $URL \| tee >(wc -c) >(shasum -a 256) >/dev/null`
138	known_versions = StrListOption(	11✔
139	default=lambda cls: cls.default_known_versions,
140	advanced=True,
141	help=textwrap.dedent(
142	f"""
143	Known versions to verify downloads against.
144
145	Each element is a pipe-separated string of `version\|platform\|sha256\|length` or
146	`version\|platform\|sha256\|length\|url_override`, where:
147
148	- `version` is the version string
149	- `platform` is one of `[{",".join(Platform.__members__.keys())}]`
150	- `sha256` is the 64-character hex representation of the expected sha256
151	digest of the download file, as emitted by `shasum -a 256`
152	- `length` is the expected length of the download file in bytes, as emitted by
153	`wc -c`
154	- (Optional) `url_override` is a specific url to use instead of the normally
155	generated url for this version
156
157	E.g., `3.1.2\|macos_x86_64\|6d0f18cd84b918c7b3edd0203e75569e0c7caecb1367bbbe409b44e28514f5be\|42813`.
158	and `3.1.2\|macos_arm64 \|aca5c1da0192e2fd46b7b55ab290a92c5f07309e7b0ebf4e45ba95731ae98291\|50926\|https://example.mac.org/bin/v3.1.2/mac-aarch64-v3.1.2.tgz`.
159
160	Values are space-stripped, so pipes can be indented for readability if necessary.
161	"""
162	),
163	)
164
165
166	class ExternalTool(Subsystem, ExportableTool, ExternalToolOptionsMixin, metaclass=ABCMeta):	11✔
167	"""Configuration for an invocable tool that we download from an external source.
168
169	Subclass this to configure a specific tool.
170
171
172	Idiomatic use:
173
174	class MyExternalTool(ExternalTool):
175	options_scope = "my-external-tool"
176	default_version = "1.2.3"
177	default_known_versions = [
178	"1.2.3\|linux_arm64 \|feed6789feed6789feed6789feed6789feed6789feed6789feed6789feed6789\|112233",
179	"1.2.3\|linux_x86_64\|cafebabacafebabacafebabacafebabacafebabacafebabacafebabacafebaba\|878986",
180	"1.2.3\|macos_arm64 \|deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef\|222222",
181	"1.2.3\|macos_x86_64\|1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd\|333333",
182	]
183
184	version_constraints = ">=1.2.3, <2.0"
185
186	def generate_url(self, plat: Platform) -> str:
187	...
188
189	def generate_exe(self, plat: Platform) -> str:
190	return "./path-to/binary
191
192	@rule
193	async def my_rule(my_external_tool: MyExternalTool, platform: Platform) -> Foo:
194	downloaded_tool = await download_external_tool(my_external_tool.get_request(platform))
195	...
196	"""
197
198	def __init__(self, args, *kwargs):	11✔
UNCOV 199	super().__init__(args, *kwargs)	×
UNCOV 200	self.check_version_constraints()	×
201
202	export_mode = ExportMode.binary	11✔
203
204	use_unsupported_version = EnumOption(	11✔
205	advanced=True,
206	help=lambda cls: textwrap.dedent(
207	f"""
208	What action to take in case the requested version of {cls.name} is not supported.
209
210	Supported {cls.name} versions: {cls.version_constraints if cls.version_constraints else "unspecified"}
211	"""
212	),
213	default=UnsupportedVersionUsage.RaiseError,
214	)
215
216	@abstractmethod	11✔
217	def generate_url(self, plat: Platform) -> str:	11✔
218	"""Returns the URL for the given version of the tool, runnable on the given os+arch.
219
220	Implementations should raise ExternalToolError if they cannot resolve the arguments
221	to a URL. The raised exception need not have a message - a sensible one will be generated.
222	"""
223
224	def generate_exe(self, plat: Platform) -> str:	11✔
225	"""Returns the path to the tool executable.
226
227	If the downloaded artifact is the executable itself, you can leave this unimplemented.
228
229	If the downloaded artifact is an archive, this should be overridden to provide a
230	relative path in the downloaded archive, e.g. `./bin/protoc`.
231	"""
232	return f"./{self.generate_url(plat).rsplit('/', 1)[-1]}"	×
233
234	def known_version(self, plat: Platform) -> ExternalToolVersion \| None:	11✔
UNCOV 235	for known_version in self.known_versions:	×
UNCOV 236	tool_version = self.decode_known_version(known_version)	×
UNCOV 237	if plat.value == tool_version.platform and tool_version.version == self.version:	×
UNCOV 238	return tool_version	×
UNCOV 239	return None	×
240
241	def get_request(self, plat: Platform) -> ExternalToolRequest:	11✔
242	"""Generate a request for this tool."""
243
UNCOV 244	tool_version = self.known_version(plat)	×
UNCOV 245	if tool_version:	×
UNCOV 246	return self.get_request_for(	×
247	tool_version.platform,
248	tool_version.sha256,
249	tool_version.filesize,
250	url_override=tool_version.url_override,
251	)
UNCOV 252	raise UnknownVersion(	×
253	softwrap(
254	f"""
255	No known version of {self.name} {self.version} for {plat.value} found in
256	{self.known_versions}
257	"""
258	)
259	)
260
261	@classmethod	11✔
262	def decode_known_version(cls, known_version: str) -> ExternalToolVersion:	11✔
UNCOV 263	try:	×
UNCOV 264	return ExternalToolVersion.decode(known_version)	×
265	except ValueError:	×
266	raise ExternalToolError(	×
267	f"Bad value for [{cls.options_scope}].known_versions: {known_version}"
268	)
269
270	@classmethod	11✔
271	def split_known_version_str(cls, known_version: str) -> tuple[str, str, str, int]:	11✔
272	version = cls.decode_known_version(known_version)	×
273	return version.version, version.platform, version.sha256, version.filesize	×
274
275	def get_request_for(	11✔
276	self, plat_val: str, sha256: str, length: int, url_override: str \| None = None
277	) -> ExternalToolRequest:
278	"""Generate a request for this tool from the given info."""
UNCOV 279	plat = Platform(plat_val)	×
UNCOV 280	digest = FileDigest(fingerprint=sha256, serialized_bytes_length=length)	×
UNCOV 281	try:	×
UNCOV 282	url = url_override or self.generate_url(plat)	×
UNCOV 283	exe = self.generate_exe(plat)	×
284	except ExternalToolError as e:	×
285	raise ExternalToolError(	×
286	f"Couldn't find {self.name} version {self.version} on {plat.value}"
287	) from e
UNCOV 288	return ExternalToolRequest(DownloadFile(url=url, expected_digest=digest), exe)	×
289
290	def check_version_constraints(self) -> None:	11✔
UNCOV 291	if not self.version_constraints:	×
UNCOV 292	return None	×
293	# Note that this is not a Python requirement. We're just hackily piggybacking off
294	# packaging.requirements.Requirement's ability to check version constraints.
UNCOV 295	constraints = Requirement(f"{self.name}{self.version_constraints}")	×
UNCOV 296	if constraints.specifier.contains(self.version):	×
297	# all ok
UNCOV 298	return None	×
299
UNCOV 300	msg = [	×
301	f"The option [{self.options_scope}].version is set to {self.version}, which is not "
302	f"compatible with what this release of Pants expects: {constraints}.",
303	"Please update the version to a supported value, or consider using a different Pants",
304	"release if you cannot change the version.",
305	]
306
UNCOV 307	if self.use_unsupported_version is UnsupportedVersionUsage.LogWarning:	×
UNCOV 308	msg.extend(	×
309	[
310	"Alternatively, you can ignore this warning (at your own peril) by adding this",
311	"to the GLOBAL section of pants.toml:",
312	f'ignore_warnings = ["The option [{self.options_scope}].version is set to"].',
313	]
314	)
UNCOV 315	logger.warning(" ".join(msg))	×
UNCOV 316	elif self.use_unsupported_version is UnsupportedVersionUsage.RaiseError:	×
UNCOV 317	msg.append(	×
318	softwrap(
319	f"""
320	Alternatively, update [{self.options_scope}].use_unsupported_version to be
321	'warning'.
322	"""
323	)
324	)
UNCOV 325	raise UnsupportedVersion(" ".join(msg))	×
326
327
328	class TemplatedExternalToolOptionsMixin(ExternalToolOptionsMixin):	11✔
329	"""Common options for implementing a subsystem providing an `ExternalToolRequest` via a URL
330	template."""
331
332	default_url_template: str	11✔
333	default_url_platform_mapping: dict[str, str] \| None = None	11✔
334
335	url_template = StrOption(	11✔
336	default=lambda cls: cls.default_url_template,
337	advanced=True,
338	help=softwrap(
339	f"""
340	URL to download the tool, either as a single binary file or a compressed file
341	(e.g. zip file). You can change this to point to your own hosted file, e.g. to
342	work with proxies or for access via the filesystem through a `file:$abspath` URL (e.g.
343	`file:/this/is/absolute`, possibly by
344	[templating the buildroot in a config file]({doc_url("docs/using-pants/key-concepts/options#config-file-entries")})).
345
346	Use `{{version}}` to have the value from `--version` substituted, and `{{platform}}` to
347	have a value from `--url-platform-mapping` substituted in, depending on the
348	current platform. For example,
349	https://github.com/.../protoc-{{version}}-{{platform}}.zip.
350	"""
351	),
352	)
353
354	url_platform_mapping = DictOption[str](	11✔
355	"--url-platform-mapping",
356	default=lambda cls: cls.default_url_platform_mapping,
357	advanced=True,
358	help=softwrap(
359	"""
360	A dictionary mapping platforms to strings to be used when generating the URL
361	to download the tool.
362
363	In `--url-template`, anytime the `{platform}` string is used, Pants will determine the
364	current platform, and substitute `{platform}` with the respective value from your dictionary.
365
366	For example, if you define `{"macos_x86_64": "apple-darwin", "linux_x86_64": "unknown-linux"}`,
367	and run Pants on Linux with an intel architecture, then `{platform}` will be substituted
368	in the `--url-template` option with `unknown-linux`.
369	"""
370	),
371	)
372
373
374	class TemplatedExternalTool(ExternalTool, TemplatedExternalToolOptionsMixin):	11✔
375	"""Extends the ExternalTool to allow url templating for custom/self-hosted source.
376
377	In addition to ExternalTool functionalities, it is needed to set, e.g.:
378
379	default_url_template = "https://tool.url/{version}/{platform}-mytool.zip"
380	default_url_platform_mapping = {
381	"macos_x86_64": "osx_intel",
382	"macos_arm64": "osx_arm",
383	"linux_x86_64": "linux",
384	}
385
386	The platform mapping dict is optional.
387	"""
388
389	def generate_url(self, plat: Platform) -> str:	11✔
UNCOV 390	platform = self.url_platform_mapping.get(plat.value, "")	×
UNCOV 391	return self.url_template.format(version=self.version, platform=platform)	×
392
393
394	@rule(level=LogLevel.DEBUG)	11✔
395	async def download_external_tool(request: ExternalToolRequest) -> DownloadedExternalTool:	11✔
396	# Download and extract.
397	maybe_archive_digest = await download_file(request.download_file_request, **implicitly())	×
398	extracted_archive = await maybe_extract_archive(**implicitly(maybe_archive_digest))	×
399
400	# Confirm executable.
401	exe_path = request.exe.lstrip("./")	×
402	digest = extracted_archive.digest	×
403	is_not_executable = False	×
404	digest_entries = []	×
405	for entry in await get_digest_entries(digest):	×
406	if isinstance(entry, FileEntry) and entry.path == exe_path and not entry.is_executable:	×
407	# We should recreate the digest with the executable bit set.
408	is_not_executable = True	×
409	entry = dataclasses.replace(entry, is_executable=True)	×
410	digest_entries.append(entry)	×
411	if is_not_executable:	×
412	digest = await create_digest(CreateDigest(digest_entries))	×
413
414	return DownloadedExternalTool(digest, request.exe)	×
415
416
417	@dataclass(frozen=True)	11✔
418	class ExportExternalToolRequest(ExportRequest):	11✔
419	pass	11✔
420	# tool: type[ExternalTool]
421
422
423	@dataclass(frozen=True)	11✔
424	class _ExportExternalToolForResolveRequest(EngineAwareParameter):	11✔
425	resolve: str	11✔
426
427
428	@dataclass(frozen=True)	11✔
429	class MaybeExportResult:	11✔
430	result: ExportResult \| None	11✔
431
432
433	@rule(level=LogLevel.DEBUG)	11✔
434	async def export_external_tool(	11✔
435	req: _ExportExternalToolForResolveRequest, platform: Platform, union_membership: UnionMembership
436	) -> MaybeExportResult:
437	"""Export a downloadable tool. Downloads all the tools to `bins`, and symlinks the primary exe
438	to the `bin` directory.
439
440	We use the last segment of the exe instead of the resolve because:
441	- it's probably the exe name people expect
442	- avoids clutter from the resolve name (ex "tfsec" instead of "terraform-tfsec")
443	"""
UNCOV 444	exportables = ExportableTool.filter_for_subclasses(	×
445	union_membership,
446	ExternalTool, # type:ignore[type-abstract] # ExternalTool is abstract, and mypy doesn't like that we might return it
447	)
UNCOV 448	maybe_exportable = exportables.get(req.resolve)	×
UNCOV 449	if not maybe_exportable:	×
450	return MaybeExportResult(None)	×
451
UNCOV 452	tool = await _construct_subsystem(maybe_exportable)	×
UNCOV 453	downloaded_tool = await download_external_tool(tool.get_request(platform))	×
454
UNCOV 455	dest = os.path.join("bins", tool.name)	×
456
UNCOV 457	exe = tool.generate_exe(platform)	×
UNCOV 458	return MaybeExportResult(	×
459	ExportResult(
460	description=f"Export tool {req.resolve}",
461	reldir=dest,
462	digest=downloaded_tool.digest,
463	resolve=req.resolve,
464	exported_binaries=(ExportedBinary(name=tool.binary_name, path_in_export=exe),),
465	)
466	)
467
468
469	@rule	11✔
470	async def export_external_tools(	11✔
471	request: ExportExternalToolRequest, export: ExportSubsystem
472	) -> ExportResults:
473	maybe_tools = await concurrently(	×
474	export_external_tool(_ExportExternalToolForResolveRequest(resolve), **implicitly())
475	for resolve in export.binaries
476	)
477	return ExportResults(tool.result for tool in maybe_tools if tool.result is not None)	×
478
479
480	def rules():	11✔
481	return (	11✔
482	*collect_rules(),
483	*archive.rules(),
484	UnionRule(ExportRequest, ExportExternalToolRequest),
485	)

pantsbuild / pants / 20147226056

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