26260209689

Committed 21 May 2026 11:59PM UTC coverage: 75.453% (-15.7%) from 91.156%

Build # 26260209689

Build Type

Pull #23365

github

Committed by

web-flow

Commit Message

Merge 5fe873b58 into 7ea655ba0

Pull Request Pull Request #23365: uv.lock -> pex optimization

Coverage Stats

5 of 16 new or added lines in 1 file covered. (31.25%)

10118 existing lines in 378 files now uncovered.

54669 of 72454 relevant lines covered (75.45%)

2.31 hits per line

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

67.86

/src/python/pants/backend/docker/util_rules/docker_binary.py

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

from __future__ import annotations

import logging
import os
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import cast

from pants.backend.docker.subsystems.docker_options import DockerOptions
from pants.backend.docker.util_rules.docker_build_args import DockerBuildArgs
from pants.core.util_rules.system_binaries import (
    BinaryPath,
    BinaryPathRequest,
    BinaryPathTest,
    BinaryShims,
    BinaryShimsRequest,
    create_binary_shims,
    find_binary,
)
from pants.engine.fs import Digest
from pants.engine.internals.selectors import concurrently
from pants.engine.process import Process, ProcessCacheScope
from pants.engine.rules import collect_rules, implicitly, rule
from pants.util.logging import LogLevel
from pants.util.strutil import pluralize

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class DockerBinary(BinaryPath):
    """The `docker` binary."""

    extra_env: Mapping[str, str]
    extra_input_digests: Mapping[str, Digest] | None

    is_podman: bool

    def __init__(
        self,
        path: str,
        fingerprint: str | None = None,
        extra_env: Mapping[str, str] | None = None,
        extra_input_digests: Mapping[str, Digest] | None = None,
        is_podman: bool = False,
    ) -> None:
        object.__setattr__(self, "extra_env", {} if extra_env is None else extra_env)
        object.__setattr__(self, "extra_input_digests", extra_input_digests)
        object.__setattr__(self, "is_podman", is_podman)
        super().__init__(path, fingerprint)

    def _get_process_environment(self, env: Mapping[str, str]) -> Mapping[str, str]:
        if not self.extra_env:
            return env

        res = {**self.extra_env, **env}

        # Merge the PATH entries, in case they are present in both `env` and `self.extra_env`.
        res["PATH"] = os.pathsep.join(
            p for p in (m.get("PATH") for m in (self.extra_env, env)) if p
        )
        return res

    def build_image(
        self,
        tags: tuple[str, ...],
        digest: Digest,
        dockerfile: str,
        build_args: DockerBuildArgs,
        context_root: str,
        env: Mapping[str, str],
        use_buildx: bool,
        extra_args: tuple[str, ...] = (),
        output_files: tuple[str, ...] = (),
        output_directories: tuple[str, ...] = (),
    ) -> Process:
        if use_buildx:
            build_commands = ["buildx", "build"]
        else:
            build_commands = ["build"]

        args = [self.path, *build_commands, *extra_args]

        for tag in tags:
            args.extend(["--tag", tag])

        for build_arg in build_args:
            args.extend(["--build-arg", build_arg])

        args.extend(["--file", dockerfile])

        # Docker context root.
        args.append(context_root)

        return Process(
            argv=tuple(args),
            description=(
                f"Building docker image {tags[0]}"
                + (f" +{pluralize(len(tags) - 1, 'additional tag')}." if len(tags) > 1 else "")
            ),
            env=self._get_process_environment(env),
            input_digest=digest,
            immutable_input_digests=self.extra_input_digests,
            output_files=output_files,
            output_directories=output_directories,
            # We must run the docker build commands every time, even if nothing has changed,
            # in case the user ran `docker image rm` outside of Pants.
            cache_scope=ProcessCacheScope.PER_SESSION,
        )

    def push_image(self, tag: str, env: Mapping[str, str] | None = None) -> Process:
        return Process(
            argv=(self.path, "push", tag),
            cache_scope=ProcessCacheScope.PER_SESSION,
            description=f"Pushing docker image {tag}",
            env=self._get_process_environment(env or {}),
            immutable_input_digests=self.extra_input_digests,
        )

    def run_image(
        self,
        tag: str,
        *,
        docker_run_args: tuple[str, ...] | None = None,
        image_args: tuple[str, ...] | None = None,
        env: Mapping[str, str] | None = None,
    ) -> Process:
        return Process(
            argv=(self.path, "run", *(docker_run_args or []), tag, *(image_args or [])),
            cache_scope=ProcessCacheScope.PER_SESSION,
            description=f"Running docker image {tag}",
            env=self._get_process_environment(env or {}),
            immutable_input_digests=self.extra_input_digests,
        )


async def _get_docker_tools_shims(
    *,
    tools: Sequence[str],
    optional_tools: Sequence[str],
    search_path: Sequence[str],
    rationale: str,
) -> BinaryShims:
    all_binary_first_paths: list[BinaryPath] = []

    if tools:
        tools_requests = [
            BinaryPathRequest(binary_name=binary_name, search_path=search_path)
            for binary_name in tools
        ]

        tools_paths = await concurrently(
            find_binary(tools_request, **implicitly()) for tools_request in tools_requests
        )

        all_binary_first_paths.extend(
            [
                path.first_path_or_raise(request, rationale=rationale)
                for request, path in zip(tools_requests, tools_paths)
            ]
        )

    if optional_tools:
        optional_tools_requests = [
            BinaryPathRequest(binary_name=binary_name, search_path=search_path)
            for binary_name in optional_tools
        ]

        optional_tools_paths = await concurrently(
            find_binary(optional_tools_request, **implicitly())
            for optional_tools_request in optional_tools_requests
        )

        all_binary_first_paths.extend(
            [
                cast(BinaryPath, path.first_path)  # safe since we check for non-empty paths below
                for path in optional_tools_paths
                if path.paths
            ]
        )

    tools_shims = await create_binary_shims(
        BinaryShimsRequest.for_paths(
            *all_binary_first_paths,
            rationale=rationale,
        ),
        **implicitly(),
    )

    return tools_shims


@rule(desc="Finding the `docker` binary and related tooling", level=LogLevel.DEBUG)
async def get_docker(
    docker_options: DockerOptions, docker_options_env_aware: DockerOptions.EnvironmentAware
) -> DockerBinary:
    search_path = docker_options_env_aware.executable_search_path

    first_path: BinaryPath | None = None
    is_podman = False

    if getattr(docker_options.options, "experimental_enable_podman", False):
        # Enable podman support with `pants.backend.experimental.docker.podman`
        request = BinaryPathRequest(
            binary_name="podman",
            search_path=search_path,
            test=BinaryPathTest(args=["-v"]),
        )
        paths = await find_binary(request, **implicitly())
        first_path = paths.first_path
        if first_path:
            is_podman = True
            logger.warning("podman found. Podman support is experimental.")

    if not first_path:
        request = BinaryPathRequest(
            binary_name="docker",
            search_path=search_path,
            test=BinaryPathTest(args=["-v"]),
        )
        paths = await find_binary(request, **implicitly())
        first_path = paths.first_path_or_raise(request, rationale="interact with the docker daemon")

    if not docker_options.tools and not docker_options.optional_tools:
        return DockerBinary(first_path.path, first_path.fingerprint, is_podman=is_podman)

    tools_shims = await _get_docker_tools_shims(
        tools=docker_options.tools,
        optional_tools=docker_options.optional_tools,
        search_path=search_path,
        rationale="use docker",
    )

    extra_env = {"PATH": tools_shims.path_component}
    extra_input_digests = tools_shims.immutable_input_digests

    return DockerBinary(
        first_path.path,
        first_path.fingerprint,
        extra_env=extra_env,
        extra_input_digests=extra_input_digests,
        is_podman=is_podman,
    )


def rules():
    return collect_rules()

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	4✔
5
6	import logging	4✔
7	import os	4✔
8	from collections.abc import Mapping, Sequence	4✔
9	from dataclasses import dataclass	4✔
10	from typing import cast	4✔
11
12	from pants.backend.docker.subsystems.docker_options import DockerOptions	4✔
13	from pants.backend.docker.util_rules.docker_build_args import DockerBuildArgs	4✔
14	from pants.core.util_rules.system_binaries import (	4✔
15	BinaryPath,
16	BinaryPathRequest,
17	BinaryPathTest,
18	BinaryShims,
19	BinaryShimsRequest,
20	create_binary_shims,
21	find_binary,
22	)
23	from pants.engine.fs import Digest	4✔
24	from pants.engine.internals.selectors import concurrently	4✔
25	from pants.engine.process import Process, ProcessCacheScope	4✔
26	from pants.engine.rules import collect_rules, implicitly, rule	4✔
27	from pants.util.logging import LogLevel	4✔
28	from pants.util.strutil import pluralize	4✔
29
30	logger = logging.getLogger(__name__)	4✔
31
32
33	@dataclass(frozen=True)	4✔
34	class DockerBinary(BinaryPath):	4✔
35	"""The `docker` binary."""
36
37	extra_env: Mapping[str, str]	4✔
38	extra_input_digests: Mapping[str, Digest] \| None	4✔
39
40	is_podman: bool	4✔
41
42	def __init__(	4✔
43	self,
44	path: str,
45	fingerprint: str \| None = None,
46	extra_env: Mapping[str, str] \| None = None,
47	extra_input_digests: Mapping[str, Digest] \| None = None,
48	is_podman: bool = False,
49	) -> None:
50	object.__setattr__(self, "extra_env", {} if extra_env is None else extra_env)	1✔
51	object.__setattr__(self, "extra_input_digests", extra_input_digests)	1✔
52	object.__setattr__(self, "is_podman", is_podman)	1✔
53	super().__init__(path, fingerprint)	1✔
54
55	def _get_process_environment(self, env: Mapping[str, str]) -> Mapping[str, str]:	4✔
56	if not self.extra_env:	1✔
57	return env	1✔
58
59	res = {self.extra_env, env}	×
60
61	# Merge the PATH entries, in case they are present in both `env` and `self.extra_env`.
62	res["PATH"] = os.pathsep.join(	×
63	p for p in (m.get("PATH") for m in (self.extra_env, env)) if p
64	)
65	return res	×
66
67	def build_image(	4✔
68	self,
69	tags: tuple[str, ...],
70	digest: Digest,
71	dockerfile: str,
72	build_args: DockerBuildArgs,
73	context_root: str,
74	env: Mapping[str, str],
75	use_buildx: bool,
76	extra_args: tuple[str, ...] = (),
77	output_files: tuple[str, ...] = (),
78	output_directories: tuple[str, ...] = (),
79	) -> Process:
80	if use_buildx:	1✔
UNCOV 81	build_commands = ["buildx", "build"]	×
82	else:
83	build_commands = ["build"]	1✔
84
85	args = [self.path, build_commands, extra_args]	1✔
86
87	for tag in tags:	1✔
88	args.extend(["--tag", tag])	1✔
89
90	for build_arg in build_args:	1✔
91	args.extend(["--build-arg", build_arg])	1✔
92
93	args.extend(["--file", dockerfile])	1✔
94
95	# Docker context root.
96	args.append(context_root)	1✔
97
98	return Process(	1✔
99	argv=tuple(args),
100	description=(
101	f"Building docker image {tags[0]}"
102	+ (f" +{pluralize(len(tags) - 1, 'additional tag')}." if len(tags) > 1 else "")
103	),
104	env=self._get_process_environment(env),
105	input_digest=digest,
106	immutable_input_digests=self.extra_input_digests,
107	output_files=output_files,
108	output_directories=output_directories,
109	# We must run the docker build commands every time, even if nothing has changed,
110	# in case the user ran `docker image rm` outside of Pants.
111	cache_scope=ProcessCacheScope.PER_SESSION,
112	)
113
114	def push_image(self, tag: str, env: Mapping[str, str] \| None = None) -> Process:	4✔
UNCOV 115	return Process(	×
116	argv=(self.path, "push", tag),
117	cache_scope=ProcessCacheScope.PER_SESSION,
118	description=f"Pushing docker image {tag}",
119	env=self._get_process_environment(env or {}),
120	immutable_input_digests=self.extra_input_digests,
121	)
122
123	def run_image(	4✔
124	self,
125	tag: str,
126	*,
127	docker_run_args: tuple[str, ...] \| None = None,
128	image_args: tuple[str, ...] \| None = None,
129	env: Mapping[str, str] \| None = None,
130	) -> Process:
UNCOV 131	return Process(	×
132	argv=(self.path, "run", (docker_run_args or []), tag, (image_args or [])),
133	cache_scope=ProcessCacheScope.PER_SESSION,
134	description=f"Running docker image {tag}",
135	env=self._get_process_environment(env or {}),
136	immutable_input_digests=self.extra_input_digests,
137	)
138
139
140	async def _get_docker_tools_shims(	4✔
141	*,
142	tools: Sequence[str],
143	optional_tools: Sequence[str],
144	search_path: Sequence[str],
145	rationale: str,
146	) -> BinaryShims:
UNCOV 147	all_binary_first_paths: list[BinaryPath] = []	×
148
UNCOV 149	if tools:	×
UNCOV 150	tools_requests = [	×
151	BinaryPathRequest(binary_name=binary_name, search_path=search_path)
152	for binary_name in tools
153	]
154
UNCOV 155	tools_paths = await concurrently(	×
156	find_binary(tools_request, **implicitly()) for tools_request in tools_requests
157	)
158
UNCOV 159	all_binary_first_paths.extend(	×
160	[
161	path.first_path_or_raise(request, rationale=rationale)
162	for request, path in zip(tools_requests, tools_paths)
163	]
164	)
165
UNCOV 166	if optional_tools:	×
UNCOV 167	optional_tools_requests = [	×
168	BinaryPathRequest(binary_name=binary_name, search_path=search_path)
169	for binary_name in optional_tools
170	]
171
UNCOV 172	optional_tools_paths = await concurrently(	×
173	find_binary(optional_tools_request, **implicitly())
174	for optional_tools_request in optional_tools_requests
175	)
176
UNCOV 177	all_binary_first_paths.extend(	×
178	[
179	cast(BinaryPath, path.first_path) # safe since we check for non-empty paths below
180	for path in optional_tools_paths
181	if path.paths
182	]
183	)
184
UNCOV 185	tools_shims = await create_binary_shims(	×
186	BinaryShimsRequest.for_paths(
187	*all_binary_first_paths,
188	rationale=rationale,
189	),
190	**implicitly(),
191	)
192
UNCOV 193	return tools_shims	×
194
195
196	@rule(desc="Finding the `docker` binary and related tooling", level=LogLevel.DEBUG)	4✔
197	async def get_docker(	4✔
198	docker_options: DockerOptions, docker_options_env_aware: DockerOptions.EnvironmentAware
199	) -> DockerBinary:
200	search_path = docker_options_env_aware.executable_search_path	1✔
201
202	first_path: BinaryPath \| None = None	1✔
203	is_podman = False	1✔
204
205	if getattr(docker_options.options, "experimental_enable_podman", False):	1✔
206	# Enable podman support with `pants.backend.experimental.docker.podman`
UNCOV 207	request = BinaryPathRequest(	×
208	binary_name="podman",
209	search_path=search_path,
210	test=BinaryPathTest(args=["-v"]),
211	)
UNCOV 212	paths = await find_binary(request, **implicitly())	×
UNCOV 213	first_path = paths.first_path	×
UNCOV 214	if first_path:	×
UNCOV 215	is_podman = True	×
UNCOV 216	logger.warning("podman found. Podman support is experimental.")	×
217
218	if not first_path:	1✔
219	request = BinaryPathRequest(	1✔
220	binary_name="docker",
221	search_path=search_path,
222	test=BinaryPathTest(args=["-v"]),
223	)
224	paths = await find_binary(request, **implicitly())	1✔
225	first_path = paths.first_path_or_raise(request, rationale="interact with the docker daemon")	1✔
226
227	if not docker_options.tools and not docker_options.optional_tools:	1✔
228	return DockerBinary(first_path.path, first_path.fingerprint, is_podman=is_podman)	1✔
229
UNCOV 230	tools_shims = await _get_docker_tools_shims(	×
231	tools=docker_options.tools,
232	optional_tools=docker_options.optional_tools,
233	search_path=search_path,
234	rationale="use docker",
235	)
236
UNCOV 237	extra_env = {"PATH": tools_shims.path_component}	×
UNCOV 238	extra_input_digests = tools_shims.immutable_input_digests	×
239
UNCOV 240	return DockerBinary(	×
241	first_path.path,
242	first_path.fingerprint,
243	extra_env=extra_env,
244	extra_input_digests=extra_input_digests,
245	is_podman=is_podman,
246	)
247
248
249	def rules():	4✔
250	return collect_rules()	4✔

pantsbuild / pants / 26260209689

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