23177125175

Committed 17 Mar 2026 03:32AM UTC coverage: 52.677% (-40.3%) from 92.932%

Build # 23177125175

Build Type

Pull #23177

github

Committed by

web-flow

Commit Message

Merge 1824dfbf4 into 0b9fdfb0e

Pull Request Pull Request #23177: Bump the gha-deps group across 1 directory with 4 updates

Coverage Stats

31687 of 60153 relevant lines covered (52.68%)

1.05 hits per line

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

0.0

/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	×
5
6	import logging	×
7	from collections.abc import Iterable	×
8	from dataclasses import dataclass	×
9
10	from pants.backend.helm.goals.package import BuiltHelmArtifact	×
11	from pants.backend.helm.resolve.remotes import HelmRegistry	×
12	from pants.backend.helm.subsystems.helm import HelmSubsystem	×
13	from pants.backend.helm.target_types import (	×
14	HelmChartFieldSet,
15	HelmChartRepositoryField,
16	HelmRegistriesField,
17	HelmSkipPushField,
18	)
19	from pants.backend.helm.util_rules.chart import HelmChartRequest, get_helm_chart	×
20	from pants.backend.helm.util_rules.tool import HelmProcess, helm_process	×
21	from pants.core.goals.package import PackageFieldSet	×
22	from pants.core.goals.publish import (	×
23	CheckSkipRequest,
24	CheckSkipResult,
25	PublishFieldSet,
26	PublishOutputData,
27	PublishPackages,
28	PublishProcesses,
29	PublishRequest,
30	)
31	from pants.engine.process import InteractiveProcess	×
32	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	×
33	from pants.engine.unions import UnionRule	×
34	from pants.util.logging import LogLevel	×
35
36	logger = logging.getLogger(__name__)	×
37
38
39	class PublishHelmChartRequest(PublishRequest):	×
40	pass	×
41
42
43	@dataclass(frozen=True)	×
44	class HelmPublishFieldSet(HelmChartFieldSet, PublishFieldSet):	×
45	publish_request_type = PublishHelmChartRequest	×
46
47	registries: HelmRegistriesField	×
48	repository: HelmChartRepositoryField	×
49	skip_push: HelmSkipPushField	×
50
51	def make_skip_request(self, package_fs: PackageFieldSet) -> PublishHelmChartSkipRequest \| None:	×
52	return PublishHelmChartSkipRequest(publish_fs=self, package_fs=package_fs)	×
53
54	def get_output_data(self) -> PublishOutputData:	×
55	return PublishOutputData(	×
56	{
57	"publisher": "helm",
58	"registries": self.registries.value or (),
59	**super().get_output_data(),
60	}
61	)
62
63
64	class PublishHelmChartSkipRequest(CheckSkipRequest[HelmPublishFieldSet]):	×
65	pass	×
66
67
68	def get_helm_registries(	×
69	helm_subsystem: HelmSubsystem, registries: Iterable[str] \| None
70	) -> list[HelmRegistry]:
71	return list(helm_subsystem.remotes().get(*(registries or [])))	×
72
73
74	@rule	×
75	async def check_if_skip_push(	×
76	request: PublishHelmChartSkipRequest, helm_subsystem: HelmSubsystem
77	) -> CheckSkipResult:
78	registries_to_push = get_helm_registries(helm_subsystem, request.publish_fs.registries.value)	×
79	if registries_to_push and not request.publish_fs.skip_push.value:	×
80	return CheckSkipResult.no_skip()	×
81	reason = (	×
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())	×
87	return CheckSkipResult.skip(	×
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)	×
95	async def publish_helm_chart(	×
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
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():	×
137	return [	×
138	*collect_rules(),
139	*HelmPublishFieldSet.rules(),
140	UnionRule(CheckSkipRequest, PublishHelmChartSkipRequest),
141	]

pantsbuild / pants / 23177125175

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