21733710241

Committed 06 Feb 2026 12:18AM UTC coverage: 80.284%. Remained the same

Build # 21733710241

Build Type

push

github

Committed by

web-flow

Commit Message

Skip Preemptive Python (#23053)

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

Run Details

28 of 39 new or added lines in 2 files covered. (71.79%)

2 existing lines in 2 files now uncovered.

78544 of 97833 relevant lines covered (80.28%)

3.36 hits per line

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

53.76

/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 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 (
    CheckSkipRequest,
    CheckSkipResult,
    PublishFieldSet,
    PublishOutputData,
    PublishPackages,
    PublishProcesses,
    PublishRequest,
)
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) -> PythonDistCheckSkipRequest | None:
        return PythonDistCheckSkipRequest(publish_fs=self, package_fs=package_fs)

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


class PythonDistCheckSkipRequest(CheckSkipRequest[PublishPythonPackageFieldSet]):
    pass


@rule
async def check_if_skip_upload(
    request: PythonDistCheckSkipRequest, twine_subsystem: TwineSubsystem
) -> CheckSkipResult:
    if twine_subsystem.skip:
        reason = f"(by `[{TwineSubsystem.name}].skip = 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 CheckSkipResult.no_skip()
    name = (
        request.package_fs.provides.value.kwargs.get("name", "<unknown python artifact>")
        if isinstance(request.package_fs, PythonDistributionFieldSet)
        else "<unknown artifact>"
    )
    return CheckSkipResult.skip(
        names=[name],
        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(CheckSkipRequest, PythonDistCheckSkipRequest),
    )

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

pantsbuild / pants / 21733710241

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