21374897774

Committed 26 Jan 2026 09:37PM UTC coverage: 80.008% (-0.3%) from 80.269%

Build # 21374897774

Build Type

Pull #23037

github

Committed by

web-flow

Commit Message

Merge 4023b9eee into 09b8ecaa1

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

Run Details

105 of 178 new or added lines in 11 files covered. (58.99%)

238 existing lines in 14 files now uncovered.

78628 of 98275 relevant lines covered (80.01%)

3.35 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,
    )


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	)
122
123
124	def twine_upload_args(	1✔
125	twine_subsystem: TwineSubsystem,
126	config_files: ConfigFiles,
127	repo: str,
128	dists: tuple[str, ...],
129	ca_cert: Snapshot \| None,
130	) -> tuple[str, ...]:
131	args = ["upload", "--non-interactive"]	×
132
133	if ca_cert and ca_cert.files:	×
134	args.append(f"--cert={ca_cert.files[0]}")	×
135
136	if config_files.snapshot.files:	×
137	args.append(f"--config-file={config_files.snapshot.files[0]}")	×
138
139	args.extend(twine_subsystem.args)	×
140
141	if repo.startswith("@"):	×
142	# Named repository from the config file.
143	args.append(f"--repository={repo[1:]}")	×
144	else:
145	args.append(f"--repository-url={repo}")	×
146
147	args.extend(dists)	×
148	return tuple(args)	×
149
150
151	def twine_env_suffix(repo: str) -> str:	1✔
152	return f"_{repo[1:]}".replace("-", "_").upper() if repo.startswith("@") else ""	×
153
154
155	def twine_env_request(repo: str) -> EnvironmentVarsRequest:	1✔
156	suffix = twine_env_suffix(repo)	×
157	env_vars = [	×
158	"TWINE_USERNAME",
159	"TWINE_PASSWORD",
160	"TWINE_REPOSITORY_URL",
161	]
162	req = EnvironmentVarsRequest(env_vars + [f"{var}{suffix}" for var in env_vars])	×
163	return req	×
164
165
166	def twine_env(env: EnvironmentVars, repo: str) -> EnvironmentVars:	1✔
167	suffix = twine_env_suffix(repo)	×
168	return EnvironmentVars(	×
169	{key.rsplit(suffix, maxsplit=1)[0] if suffix else key: value for key, value in env.items()}
170	)
171
172
173	@rule	1✔
174	async def twine_upload(	1✔
175	request: PublishPythonPackageRequest,
176	twine_subsystem: TwineSubsystem,
177	global_options: GlobalOptions,
178	) -> PublishProcesses:
179	dists = tuple(	×
180	artifact.relpath
181	for pkg in request.packages
182	for artifact in pkg.artifacts
183	if artifact.relpath
184	)
185
NEW 186	if not dists:	×
187	return PublishProcesses()	×
188
UNCOV 189	twine_pex, packages_digest, config_files = await concurrently(	×
190	create_venv_pex(**implicitly(twine_subsystem.to_pex_request())),
191	merge_digests(MergeDigests(pkg.digest for pkg in request.packages)),
192	find_config_file(twine_subsystem.config_request()),
193	)
194
195	ca_cert_request = twine_subsystem.ca_certs_digest_request(global_options.ca_certs_path)	×
196	ca_cert = (	×
197	await digest_to_snapshot(**implicitly({ca_cert_request: CreateDigest}))
198	if ca_cert_request
199	else None
200	)
201	ca_cert_digest = (ca_cert.digest,) if ca_cert else ()	×
202
203	input_digest = await merge_digests(	×
204	MergeDigests((packages_digest, config_files.snapshot.digest, *ca_cert_digest))
205	)
206	pex_proc_requests: list[VenvPexProcess] = []	×
207	twine_envs = await concurrently(	×
208	environment_vars_subset(**implicitly({twine_env_request(repo): EnvironmentVarsRequest}))
209	for repo in request.field_set.repositories.value
210	)
211
212	for repo, env in zip(request.field_set.repositories.value, twine_envs):	×
213	pex_proc_requests.append(	×
214	VenvPexProcess(
215	twine_pex,
216	argv=twine_upload_args(twine_subsystem, config_files, repo, dists, ca_cert),
217	input_digest=input_digest,
218	extra_env=twine_env(env, repo),
219	description=repo,
220	)
221	)
222
223	processes = await concurrently(	×
224	setup_venv_pex_process(request, **implicitly()) for request in pex_proc_requests
225	)
226
227	return PublishProcesses(	×
228	PublishPackages(
229	names=dists,
230	process=InteractiveProcess.from_process(process),
231	description=process.description,
232	data=PublishOutputData({"repository": process.description}),
233	)
234	for process in processes
235	)
236
237
238	def rules():	1✔
239	return (	1✔
240	*collect_rules(),
241	*PublishPythonPackageFieldSet.rules(),
242	PythonDistribution.register_plugin_field(PythonRepositoriesField),
243	PythonDistribution.register_plugin_field(SkipTwineUploadField),
244	UnionRule(PreemptiveSkipRequest, PublishPythonPackageSkipRequest),
245	)

pantsbuild / pants / 21374897774

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