21648319458

Committed 03 Feb 2026 09:21PM UTC coverage: 80.296% (+0.02%) from 80.278%

Build # 21648319458

Build Type

push

github

Committed by

web-flow

Commit Message

Skip Preemptive Helm (#23054)

Implements [preemptive
skipping](https://github.com/pantsbuild/pants/pull/23052) for helm

Run Details

51 of 53 new or added lines in 2 files covered. (96.23%)

1 existing line in 1 file now uncovered.

78517 of 97785 relevant lines covered (80.3%)

3.36 hits per line

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

86.27

/src/python/pants/backend/helm/goals/publish.py

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

from __future__ import annotations

import logging
from collections.abc import Iterable
from dataclasses import dataclass

from pants.backend.helm.goals.package import BuiltHelmArtifact
from pants.backend.helm.resolve.remotes import HelmRegistry
from pants.backend.helm.subsystems.helm import HelmSubsystem
from pants.backend.helm.target_types import (
    HelmChartFieldSet,
    HelmChartRepositoryField,
    HelmRegistriesField,
    HelmSkipPushField,
)
from pants.backend.helm.util_rules.chart import HelmChartRequest, get_helm_chart
from pants.backend.helm.util_rules.tool import HelmProcess, helm_process
from pants.core.goals.package import PackageFieldSet
from pants.core.goals.publish import (
    CheckSkipRequest,
    CheckSkipResult,
    PublishFieldSet,
    PublishOutputData,
    PublishPackages,
    PublishProcesses,
    PublishRequest,
)
from pants.engine.process import InteractiveProcess
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.unions import UnionRule
from pants.util.logging import LogLevel

logger = logging.getLogger(__name__)


class PublishHelmChartRequest(PublishRequest):
    pass


@dataclass(frozen=True)
class HelmPublishFieldSet(HelmChartFieldSet, PublishFieldSet):
    publish_request_type = PublishHelmChartRequest

    registries: HelmRegistriesField
    repository: HelmChartRepositoryField
    skip_push: HelmSkipPushField

    def make_skip_request(self, package_fs: PackageFieldSet) -> PublishHelmChartSkipRequest | None:
        return PublishHelmChartSkipRequest(publish_fs=self, package_fs=package_fs)

    def get_output_data(self) -> PublishOutputData:
        return PublishOutputData(
            {
                "publisher": "helm",
                "registries": self.registries.value or (),
                **super().get_output_data(),
            }
        )


class PublishHelmChartSkipRequest(CheckSkipRequest[HelmPublishFieldSet]):
    pass


def get_helm_registries(
    helm_subsystem: HelmSubsystem, registries: Iterable[str] | None
) -> list[HelmRegistry]:
    return list(helm_subsystem.remotes().get(*(registries or [])))


@rule
async def check_if_skip_push(
    request: PublishHelmChartSkipRequest, helm_subsystem: HelmSubsystem
) -> CheckSkipResult:
    registries_to_push = get_helm_registries(helm_subsystem, request.publish_fs.registries.value)
    if registries_to_push and not request.publish_fs.skip_push.value:
        return CheckSkipResult.no_skip()
    reason = (
        f"missing `{request.publish_fs.registries.alias}`"
        if not registries_to_push
        else f"`{request.publish_fs.skip_push.alias}`"
    )
    chart = await get_helm_chart(HelmChartRequest(request.publish_fs), **implicitly())
    return CheckSkipResult.skip(
        names=[chart.info.artifact_name],
        description=f"(by {reason} on {request.publish_fs.address})",
        data=request.publish_fs.get_output_data(),
    )


@rule(desc="Push Helm chart to OCI registries", level=LogLevel.DEBUG)
async def publish_helm_chart(
    request: PublishHelmChartRequest, helm_subsystem: HelmSubsystem
) -> PublishProcesses:
    built_artifacts = [
        (pkg, artifact, artifact.info)
        for pkg in request.packages
        for artifact in pkg.artifacts
        if isinstance(artifact, BuiltHelmArtifact) and artifact.info
    ]

    registries_to_push = get_helm_registries(helm_subsystem, request.field_set.registries.value)
    push_repository = (
        request.field_set.repository.value or helm_subsystem.default_registry_repository
    )
    publish_refs = [
        registry.package_ref(metadata.artifact_name, repository=push_repository)
        for _, _, metadata in built_artifacts
        for registry in registries_to_push
    ]
    processes = await concurrently(
        helm_process(
            HelmProcess(
                ["push", artifact.relpath, registry.repository_ref(push_repository)],
                input_digest=pkg.digest,
                description=f"Pushing Helm chart '{metadata.name}' with version '{metadata.version}' into OCI registry: {registry.address}",
            ),
            **implicitly(),
        )
        for pkg, artifact, metadata in built_artifacts
        if artifact.relpath
        for registry in registries_to_push
    )

    return PublishProcesses(
        [
            PublishPackages(names=(package_ref,), process=InteractiveProcess.from_process(process))
            for package_ref, process in zip(publish_refs, processes)
        ]
    )


def rules():
    return [
        *collect_rules(),
        *HelmPublishFieldSet.rules(),
        UnionRule(CheckSkipRequest, PublishHelmChartSkipRequest),
    ]

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	from __future__ import annotations	1✔
5
6	import logging	1✔
7	from collections.abc import Iterable	1✔
8	from dataclasses import dataclass	1✔
9
10	from pants.backend.helm.goals.package import BuiltHelmArtifact	1✔
11	from pants.backend.helm.resolve.remotes import HelmRegistry	1✔
12	from pants.backend.helm.subsystems.helm import HelmSubsystem	1✔
13	from pants.backend.helm.target_types import (	1✔
14	HelmChartFieldSet,
15	HelmChartRepositoryField,
16	HelmRegistriesField,
17	HelmSkipPushField,
18	)
19	from pants.backend.helm.util_rules.chart import HelmChartRequest, get_helm_chart	1✔
20	from pants.backend.helm.util_rules.tool import HelmProcess, helm_process	1✔
21	from pants.core.goals.package import PackageFieldSet	1✔
22	from pants.core.goals.publish import (	1✔
23	CheckSkipRequest,
24	CheckSkipResult,
25	PublishFieldSet,
26	PublishOutputData,
27	PublishPackages,
28	PublishProcesses,
29	PublishRequest,
30	)
31	from pants.engine.process import InteractiveProcess	1✔
32	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	1✔
33	from pants.engine.unions import UnionRule	1✔
34	from pants.util.logging import LogLevel	1✔
35
36	logger = logging.getLogger(__name__)	1✔
37
38
39	class PublishHelmChartRequest(PublishRequest):	1✔
40	pass	1✔
41
42
43	@dataclass(frozen=True)	1✔
44	class HelmPublishFieldSet(HelmChartFieldSet, PublishFieldSet):	1✔
45	publish_request_type = PublishHelmChartRequest	1✔
46
47	registries: HelmRegistriesField	1✔
48	repository: HelmChartRepositoryField	1✔
49	skip_push: HelmSkipPushField	1✔
50
51	def make_skip_request(self, package_fs: PackageFieldSet) -> PublishHelmChartSkipRequest \| None:	1✔
NEW 52	return PublishHelmChartSkipRequest(publish_fs=self, package_fs=package_fs)	×
53
54	def get_output_data(self) -> PublishOutputData:	1✔
55	return PublishOutputData(	1✔
56	{
57	"publisher": "helm",
58	"registries": self.registries.value or (),
59	**super().get_output_data(),
60	}
61	)
62
63
64	class PublishHelmChartSkipRequest(CheckSkipRequest[HelmPublishFieldSet]):	1✔
65	pass	1✔
66
67
68	def get_helm_registries(	1✔
69	helm_subsystem: HelmSubsystem, registries: Iterable[str] \| None
70	) -> list[HelmRegistry]:
71	return list(helm_subsystem.remotes().get(*(registries or [])))	1✔
72
73
74	@rule	1✔
75	async def check_if_skip_push(	1✔
76	request: PublishHelmChartSkipRequest, helm_subsystem: HelmSubsystem
77	) -> CheckSkipResult:
78	registries_to_push = get_helm_registries(helm_subsystem, request.publish_fs.registries.value)	1✔
79	if registries_to_push and not request.publish_fs.skip_push.value:	1✔
80	return CheckSkipResult.no_skip()	1✔
81	reason = (	1✔
82	f"missing `{request.publish_fs.registries.alias}`"
83	if not registries_to_push
84	else f"`{request.publish_fs.skip_push.alias}`"
85	)
86	chart = await get_helm_chart(HelmChartRequest(request.publish_fs), **implicitly())	1✔
87	return CheckSkipResult.skip(	1✔
88	names=[chart.info.artifact_name],
89	description=f"(by {reason} on {request.publish_fs.address})",
90	data=request.publish_fs.get_output_data(),
91	)
92
93
94	@rule(desc="Push Helm chart to OCI registries", level=LogLevel.DEBUG)	1✔
95	async def publish_helm_chart(	1✔
96	request: PublishHelmChartRequest, helm_subsystem: HelmSubsystem
97	) -> PublishProcesses:
98	built_artifacts = [	×
99	(pkg, artifact, artifact.info)
100	for pkg in request.packages
101	for artifact in pkg.artifacts
102	if isinstance(artifact, BuiltHelmArtifact) and artifact.info
103	]
104
NEW 105	registries_to_push = get_helm_registries(helm_subsystem, request.field_set.registries.value)	×
106	push_repository = (	×
107	request.field_set.repository.value or helm_subsystem.default_registry_repository
108	)
109	publish_refs = [	×
110	registry.package_ref(metadata.artifact_name, repository=push_repository)
111	for _, _, metadata in built_artifacts
112	for registry in registries_to_push
113	]
114	processes = await concurrently(	×
115	helm_process(
116	HelmProcess(
117	["push", artifact.relpath, registry.repository_ref(push_repository)],
118	input_digest=pkg.digest,
119	description=f"Pushing Helm chart '{metadata.name}' with version '{metadata.version}' into OCI registry: {registry.address}",
120	),
121	**implicitly(),
122	)
123	for pkg, artifact, metadata in built_artifacts
124	if artifact.relpath
125	for registry in registries_to_push
126	)
127
128	return PublishProcesses(	×
129	[
130	PublishPackages(names=(package_ref,), process=InteractiveProcess.from_process(process))
131	for package_ref, process in zip(publish_refs, processes)
132	]
133	)
134
135
136	def rules():	1✔
137	return [	1✔
138	*collect_rules(),
139	*HelmPublishFieldSet.rules(),
140	UnionRule(CheckSkipRequest, PublishHelmChartSkipRequest),
141	]

pantsbuild / pants / 21648319458

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