21457998286

Committed 28 Jan 2026 10:32PM UTC coverage: 80.281% (+0.01%) from 80.269%

Build # 21457998286

Build Type

Pull #23037

github

Committed by

web-flow

Commit Message

Merge 43b38d939 into 0fdb40370

Pull Request Pull Request #23037: Enable publish without package 2

Run Details

275 of 328 new or added lines in 13 files covered. (83.84%)

46 existing lines in 9 files now uncovered.

78960 of 98355 relevant lines covered (80.28%)

3.36 hits per line

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

54.84

/src/python/pants/backend/python/goals/publish.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
from dataclasses import dataclass
from typing import cast

from pants.backend.python.subsystems.setuptools import PythonDistributionFieldSet
from pants.backend.python.subsystems.twine import TwineSubsystem
from pants.backend.python.target_types import PythonDistribution
from pants.backend.python.util_rules.pex import (
    VenvPexProcess,
    create_venv_pex,
    setup_venv_pex_process,
)
from pants.core.goals.package import PackageFieldSet
from pants.core.goals.publish import (
    PreemptiveSkipRequest,
    PublishFieldSet,
    PublishOutputData,
    PublishPackages,
    PublishProcesses,
    PublishRequest,
    SkippedPublishPackages,
)
from pants.core.util_rules.config_files import ConfigFiles, find_config_file
from pants.core.util_rules.env_vars import environment_vars_subset
from pants.engine.env_vars import EnvironmentVars, EnvironmentVarsRequest
from pants.engine.fs import CreateDigest, MergeDigests, Snapshot
from pants.engine.intrinsics import digest_to_snapshot, merge_digests
from pants.engine.process import InteractiveProcess
from pants.engine.rules import collect_rules, concurrently, implicitly, rule
from pants.engine.target import BoolField, StringSequenceField
from pants.engine.unions import UnionRule
from pants.option.global_options import GlobalOptions
from pants.util.strutil import help_text

logger = logging.getLogger(__name__)


class PythonRepositoriesField(StringSequenceField):
    alias = "repositories"
    help = help_text(
        """
        List of URL addresses or Twine repository aliases where to publish the Python package.

        Twine is used for publishing Python packages, so the address to any kind of repository
        that Twine supports may be used here.

        Aliases are prefixed with `@` to refer to a config section in your Twine configuration,
        such as a `.pypirc` file. Use `@pypi` to upload to the public PyPi repository, which is
        the default when using Twine directly.
        """
    )

    # Twine uploads to 'pypi' by default, but we don't set default to ["@pypi"] here to make it
    # explicit in the BUILD file when a package is meant for public distribution.


class SkipTwineUploadField(BoolField):
    alias = "skip_twine"
    default = False
    help = "If true, don't publish this target's packages using Twine."


class PublishPythonPackageRequest(PublishRequest):
    pass


@dataclass(frozen=True)
class PublishPythonPackageFieldSet(PublishFieldSet):
    publish_request_type = PublishPythonPackageRequest
    required_fields = (PythonRepositoriesField,)

    repositories: PythonRepositoriesField
    skip_twine: SkipTwineUploadField

    def make_skip_request(
        self, package_fs: PackageFieldSet
    ) -> PublishPythonPackageSkipRequest | None:
        return (
            PublishPythonPackageSkipRequest(publish_fs=self, package_fs=package_fs)
            if isinstance(package_fs, PythonDistributionFieldSet)
            else None
        )

    def get_output_data(self) -> PublishOutputData:
        return PublishOutputData(
            {
                "publisher": "twine",
                **super().get_output_data(),
            }
        )


class PublishPythonPackageSkipRequest(PreemptiveSkipRequest[PublishPythonPackageFieldSet]):
    pass


@rule
async def check_if_skip_upload(
    request: PublishPythonPackageSkipRequest, twine_subsystem: TwineSubsystem
) -> SkippedPublishPackages:
    if twine_subsystem.skip:
        reason = f"(by `[{TwineSubsystem.name}].skip_push = True`)"
    elif request.publish_fs.skip_twine.value:
        reason = f"(by `{request.publish_fs.skip_twine.alias}` on {request.address})"
    elif not request.publish_fs.repositories.value:
        reason = f"(no `{request.publish_fs.repositories.alias}` specified for {request.address})"
    else:
        return SkippedPublishPackages.no_skip()
    return SkippedPublishPackages.skip(
        names=[
            cast(PythonDistributionFieldSet, request.package_fs).provides.value.kwargs.get(
                "name", "<unknown python artifact>"
            )
        ],
        description=reason,
        data=request.publish_fs.get_output_data(),
    )


def twine_upload_args(
    twine_subsystem: TwineSubsystem,
    config_files: ConfigFiles,
    repo: str,
    dists: tuple[str, ...],
    ca_cert: Snapshot | None,
) -> tuple[str, ...]:
    args = ["upload", "--non-interactive"]

    if ca_cert and ca_cert.files:
        args.append(f"--cert={ca_cert.files[0]}")

    if config_files.snapshot.files:
        args.append(f"--config-file={config_files.snapshot.files[0]}")

    args.extend(twine_subsystem.args)

    if repo.startswith("@"):
        # Named repository from the config file.
        args.append(f"--repository={repo[1:]}")
    else:
        args.append(f"--repository-url={repo}")

    args.extend(dists)
    return tuple(args)


def twine_env_suffix(repo: str) -> str:
    return f"_{repo[1:]}".replace("-", "_").upper() if repo.startswith("@") else ""


def twine_env_request(repo: str) -> EnvironmentVarsRequest:
    suffix = twine_env_suffix(repo)
    env_vars = [
        "TWINE_USERNAME",
        "TWINE_PASSWORD",
        "TWINE_REPOSITORY_URL",
    ]
    req = EnvironmentVarsRequest(env_vars + [f"{var}{suffix}" for var in env_vars])
    return req


def twine_env(env: EnvironmentVars, repo: str) -> EnvironmentVars:
    suffix = twine_env_suffix(repo)
    return EnvironmentVars(
        {key.rsplit(suffix, maxsplit=1)[0] if suffix else key: value for key, value in env.items()}
    )


@rule
async def twine_upload(
    request: PublishPythonPackageRequest,
    twine_subsystem: TwineSubsystem,
    global_options: GlobalOptions,
) -> PublishProcesses:
    dists = tuple(
        artifact.relpath
        for pkg in request.packages
        for artifact in pkg.artifacts
        if artifact.relpath
    )

    if not dists:
        return PublishProcesses()

    twine_pex, packages_digest, config_files = await concurrently(
        create_venv_pex(**implicitly(twine_subsystem.to_pex_request())),
        merge_digests(MergeDigests(pkg.digest for pkg in request.packages)),
        find_config_file(twine_subsystem.config_request()),
    )

    ca_cert_request = twine_subsystem.ca_certs_digest_request(global_options.ca_certs_path)
    ca_cert = (
        await digest_to_snapshot(**implicitly({ca_cert_request: CreateDigest}))
        if ca_cert_request
        else None
    )
    ca_cert_digest = (ca_cert.digest,) if ca_cert else ()

    input_digest = await merge_digests(
        MergeDigests((packages_digest, config_files.snapshot.digest, *ca_cert_digest))
    )
    pex_proc_requests: list[VenvPexProcess] = []
    twine_envs = await concurrently(
        environment_vars_subset(**implicitly({twine_env_request(repo): EnvironmentVarsRequest}))
        for repo in request.field_set.repositories.value
    )

    for repo, env in zip(request.field_set.repositories.value, twine_envs):
        pex_proc_requests.append(
            VenvPexProcess(
                twine_pex,
                argv=twine_upload_args(twine_subsystem, config_files, repo, dists, ca_cert),
                input_digest=input_digest,
                extra_env=twine_env(env, repo),
                description=repo,
            )
        )

    processes = await concurrently(
        setup_venv_pex_process(request, **implicitly()) for request in pex_proc_requests
    )

    return PublishProcesses(
        PublishPackages(
            names=dists,
            process=InteractiveProcess.from_process(process),
            description=process.description,
            data=PublishOutputData({"repository": process.description}),
        )
        for process in processes
    )


def rules():
    return (
        *collect_rules(),
        *PublishPythonPackageFieldSet.rules(),
        PythonDistribution.register_plugin_field(PythonRepositoriesField),
        PythonDistribution.register_plugin_field(SkipTwineUploadField),
        UnionRule(PreemptiveSkipRequest, PublishPythonPackageSkipRequest),
    )

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	1✔
5
6	import logging	1✔
7	from dataclasses import dataclass	1✔
8	from typing import cast	1✔
9
10	from pants.backend.python.subsystems.setuptools import PythonDistributionFieldSet	1✔
11	from pants.backend.python.subsystems.twine import TwineSubsystem	1✔
12	from pants.backend.python.target_types import PythonDistribution	1✔
13	from pants.backend.python.util_rules.pex import (	1✔
14	VenvPexProcess,
15	create_venv_pex,
16	setup_venv_pex_process,
17	)
18	from pants.core.goals.package import PackageFieldSet	1✔
19	from pants.core.goals.publish import (	1✔
20	PreemptiveSkipRequest,
21	PublishFieldSet,
22	PublishOutputData,
23	PublishPackages,
24	PublishProcesses,
25	PublishRequest,
26	SkippedPublishPackages,
27	)
28	from pants.core.util_rules.config_files import ConfigFiles, find_config_file	1✔
29	from pants.core.util_rules.env_vars import environment_vars_subset	1✔
30	from pants.engine.env_vars import EnvironmentVars, EnvironmentVarsRequest	1✔
31	from pants.engine.fs import CreateDigest, MergeDigests, Snapshot	1✔
32	from pants.engine.intrinsics import digest_to_snapshot, merge_digests	1✔
33	from pants.engine.process import InteractiveProcess	1✔
34	from pants.engine.rules import collect_rules, concurrently, implicitly, rule	1✔
35	from pants.engine.target import BoolField, StringSequenceField	1✔
36	from pants.engine.unions import UnionRule	1✔
37	from pants.option.global_options import GlobalOptions	1✔
38	from pants.util.strutil import help_text	1✔
39
40	logger = logging.getLogger(__name__)	1✔
41
42
43	class PythonRepositoriesField(StringSequenceField):	1✔
44	alias = "repositories"	1✔
45	help = help_text(	1✔
46	"""
47	List of URL addresses or Twine repository aliases where to publish the Python package.
48
49	Twine is used for publishing Python packages, so the address to any kind of repository
50	that Twine supports may be used here.
51
52	Aliases are prefixed with `@` to refer to a config section in your Twine configuration,
53	such as a `.pypirc` file. Use `@pypi` to upload to the public PyPi repository, which is
54	the default when using Twine directly.
55	"""
56	)
57
58	# Twine uploads to 'pypi' by default, but we don't set default to ["@pypi"] here to make it
59	# explicit in the BUILD file when a package is meant for public distribution.
60
61
62	class SkipTwineUploadField(BoolField):	1✔
63	alias = "skip_twine"	1✔
64	default = False	1✔
65	help = "If true, don't publish this target's packages using Twine."	1✔
66
67
68	class PublishPythonPackageRequest(PublishRequest):	1✔
69	pass	1✔
70
71
72	@dataclass(frozen=True)	1✔
73	class PublishPythonPackageFieldSet(PublishFieldSet):	1✔
74	publish_request_type = PublishPythonPackageRequest	1✔
75	required_fields = (PythonRepositoriesField,)	1✔
76
77	repositories: PythonRepositoriesField	1✔
78	skip_twine: SkipTwineUploadField	1✔
79
80	def make_skip_request(	1✔
81	self, package_fs: PackageFieldSet
82	) -> PublishPythonPackageSkipRequest \| None:
NEW 83	return (	×
84	PublishPythonPackageSkipRequest(publish_fs=self, package_fs=package_fs)
85	if isinstance(package_fs, PythonDistributionFieldSet)
86	else None
87	)
88
89	def get_output_data(self) -> PublishOutputData:	1✔
90	return PublishOutputData(	×
91	{
92	"publisher": "twine",
93	**super().get_output_data(),
94	}
95	)
96
97
98	class PublishPythonPackageSkipRequest(PreemptiveSkipRequest[PublishPythonPackageFieldSet]):	1✔
99	pass	1✔
100
101
102	@rule	1✔
103	async def check_if_skip_upload(	1✔
104	request: PublishPythonPackageSkipRequest, twine_subsystem: TwineSubsystem
105	) -> SkippedPublishPackages:
NEW 106	if twine_subsystem.skip:	×
NEW 107	reason = f"(by `[{TwineSubsystem.name}].skip_push = True`)"	×
NEW 108	elif request.publish_fs.skip_twine.value:	×
NEW 109	reason = f"(by `{request.publish_fs.skip_twine.alias}` on {request.address})"	×
NEW 110	elif not request.publish_fs.repositories.value:	×
NEW 111	reason = f"(no `{request.publish_fs.repositories.alias}` specified for {request.address})"	×
112	else:
NEW 113	return SkippedPublishPackages.no_skip()	×
NEW 114	return SkippedPublishPackages.skip(	×
115	names=[
116	cast(PythonDistributionFieldSet, request.package_fs).provides.value.kwargs.get(
117	"name", "<unknown python artifact>"
118	)
119	],
120	description=reason,
121	data=request.publish_fs.get_output_data(),
122	)
123
124
125	def twine_upload_args(	1✔
126	twine_subsystem: TwineSubsystem,
127	config_files: ConfigFiles,
128	repo: str,
129	dists: tuple[str, ...],
130	ca_cert: Snapshot \| None,
131	) -> tuple[str, ...]:
132	args = ["upload", "--non-interactive"]	×
133
134	if ca_cert and ca_cert.files:	×
135	args.append(f"--cert={ca_cert.files[0]}")	×
136
137	if config_files.snapshot.files:	×
138	args.append(f"--config-file={config_files.snapshot.files[0]}")	×
139
140	args.extend(twine_subsystem.args)	×
141
142	if repo.startswith("@"):	×
143	# Named repository from the config file.
144	args.append(f"--repository={repo[1:]}")	×
145	else:
146	args.append(f"--repository-url={repo}")	×
147
148	args.extend(dists)	×
149	return tuple(args)	×
150
151
152	def twine_env_suffix(repo: str) -> str:	1✔
153	return f"_{repo[1:]}".replace("-", "_").upper() if repo.startswith("@") else ""	×
154
155
156	def twine_env_request(repo: str) -> EnvironmentVarsRequest:	1✔
157	suffix = twine_env_suffix(repo)	×
158	env_vars = [	×
159	"TWINE_USERNAME",
160	"TWINE_PASSWORD",
161	"TWINE_REPOSITORY_URL",
162	]
163	req = EnvironmentVarsRequest(env_vars + [f"{var}{suffix}" for var in env_vars])	×
164	return req	×
165
166
167	def twine_env(env: EnvironmentVars, repo: str) -> EnvironmentVars:	1✔
168	suffix = twine_env_suffix(repo)	×
169	return EnvironmentVars(	×
170	{key.rsplit(suffix, maxsplit=1)[0] if suffix else key: value for key, value in env.items()}
171	)
172
173
174	@rule	1✔
175	async def twine_upload(	1✔
176	request: PublishPythonPackageRequest,
177	twine_subsystem: TwineSubsystem,
178	global_options: GlobalOptions,
179	) -> PublishProcesses:
180	dists = tuple(	×
181	artifact.relpath
182	for pkg in request.packages
183	for artifact in pkg.artifacts
184	if artifact.relpath
185	)
186
NEW 187	if not dists:	×
188	return PublishProcesses()	×
189
UNCOV 190	twine_pex, packages_digest, config_files = await concurrently(	×
191	create_venv_pex(**implicitly(twine_subsystem.to_pex_request())),
192	merge_digests(MergeDigests(pkg.digest for pkg in request.packages)),
193	find_config_file(twine_subsystem.config_request()),
194	)
195
196	ca_cert_request = twine_subsystem.ca_certs_digest_request(global_options.ca_certs_path)	×
197	ca_cert = (	×
198	await digest_to_snapshot(**implicitly({ca_cert_request: CreateDigest}))
199	if ca_cert_request
200	else None
201	)
202	ca_cert_digest = (ca_cert.digest,) if ca_cert else ()	×
203
204	input_digest = await merge_digests(	×
205	MergeDigests((packages_digest, config_files.snapshot.digest, *ca_cert_digest))
206	)
207	pex_proc_requests: list[VenvPexProcess] = []	×
208	twine_envs = await concurrently(	×
209	environment_vars_subset(**implicitly({twine_env_request(repo): EnvironmentVarsRequest}))
210	for repo in request.field_set.repositories.value
211	)
212
213	for repo, env in zip(request.field_set.repositories.value, twine_envs):	×
214	pex_proc_requests.append(	×
215	VenvPexProcess(
216	twine_pex,
217	argv=twine_upload_args(twine_subsystem, config_files, repo, dists, ca_cert),
218	input_digest=input_digest,
219	extra_env=twine_env(env, repo),
220	description=repo,
221	)
222	)
223
224	processes = await concurrently(	×
225	setup_venv_pex_process(request, **implicitly()) for request in pex_proc_requests
226	)
227
228	return PublishProcesses(	×
229	PublishPackages(
230	names=dists,
231	process=InteractiveProcess.from_process(process),
232	description=process.description,
233	data=PublishOutputData({"repository": process.description}),
234	)
235	for process in processes
236	)
237
238
239	def rules():	1✔
240	return (	1✔
241	*collect_rules(),
242	*PublishPythonPackageFieldSet.rules(),
243	PythonDistribution.register_plugin_field(PythonRepositoriesField),
244	PythonDistribution.register_plugin_field(SkipTwineUploadField),
245	UnionRule(PreemptiveSkipRequest, PublishPythonPackageSkipRequest),
246	)

pantsbuild / pants / 21457998286

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