21494992949

Committed 29 Jan 2026 09:15PM UTC coverage: 80.245%. First build

Build # 21494992949

Build Type

Pull #23053

github

Committed by

web-flow

Commit Message

Merge 16a8312ae into 78e2689de

Pull Request Pull Request #23053: Skip Preemptive Python

Run Details

71 of 124 new or added lines in 5 files covered. (57.26%)

78854 of 98267 relevant lines covered (80.24%)

3.09 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 (
    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)
            if isinstance(package_fs, PythonDistributionFieldSet)
            else None
        )

    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()
    return CheckSkipResult.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(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	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	CheckSkipRequest,
21	CheckSkipResult,
22	PublishFieldSet,
23	PublishOutputData,
24	PublishPackages,
25	PublishProcesses,
26	PublishRequest,
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(self, package_fs: PackageFieldSet) -> PythonDistCheckSkipRequest \| None:	1✔
NEW 81	return (	×
82	PythonDistCheckSkipRequest(publish_fs=self, package_fs=package_fs)
83	if isinstance(package_fs, PythonDistributionFieldSet)
84	else None
85	)
86
87	def get_output_data(self) -> PublishOutputData:	1✔
88	return PublishOutputData(	×
89	{
90	"publisher": "twine",
91	**super().get_output_data(),
92	}
93	)
94
95
96	class PythonDistCheckSkipRequest(CheckSkipRequest[PublishPythonPackageFieldSet]):	1✔
97	pass	1✔
98
99
100	@rule	1✔
101	async def check_if_skip_upload(	1✔
102	request: PythonDistCheckSkipRequest, twine_subsystem: TwineSubsystem
103	) -> CheckSkipResult:
NEW 104	if twine_subsystem.skip:	×
NEW 105	reason = f"(by `[{TwineSubsystem.name}].skip = True`)"	×
NEW 106	elif request.publish_fs.skip_twine.value:	×
NEW 107	reason = f"(by `{request.publish_fs.skip_twine.alias}` on {request.address})"	×
NEW 108	elif not request.publish_fs.repositories.value:	×
NEW 109	reason = f"(no `{request.publish_fs.repositories.alias}` specified for {request.address})"	×
110	else:
NEW 111	return CheckSkipResult.no_skip()	×
NEW 112	return CheckSkipResult.skip(	×
113	names=[
114	cast(PythonDistributionFieldSet, request.package_fs).provides.value.kwargs.get(
115	"name", "<unknown python artifact>"
116	)
117	],
118	description=reason,
119	data=request.publish_fs.get_output_data(),
120	)
121
122
123	def twine_upload_args(	1✔
124	twine_subsystem: TwineSubsystem,
125	config_files: ConfigFiles,
126	repo: str,
127	dists: tuple[str, ...],
128	ca_cert: Snapshot \| None,
129	) -> tuple[str, ...]:
130	args = ["upload", "--non-interactive"]	×
131
132	if ca_cert and ca_cert.files:	×
133	args.append(f"--cert={ca_cert.files[0]}")	×
134
135	if config_files.snapshot.files:	×
136	args.append(f"--config-file={config_files.snapshot.files[0]}")	×
137
138	args.extend(twine_subsystem.args)	×
139
140	if repo.startswith("@"):	×
141	# Named repository from the config file.
142	args.append(f"--repository={repo[1:]}")	×
143	else:
144	args.append(f"--repository-url={repo}")	×
145
146	args.extend(dists)	×
147	return tuple(args)	×
148
149
150	def twine_env_suffix(repo: str) -> str:	1✔
151	return f"_{repo[1:]}".replace("-", "_").upper() if repo.startswith("@") else ""	×
152
153
154	def twine_env_request(repo: str) -> EnvironmentVarsRequest:	1✔
155	suffix = twine_env_suffix(repo)	×
156	env_vars = [	×
157	"TWINE_USERNAME",
158	"TWINE_PASSWORD",
159	"TWINE_REPOSITORY_URL",
160	]
161	req = EnvironmentVarsRequest(env_vars + [f"{var}{suffix}" for var in env_vars])	×
162	return req	×
163
164
165	def twine_env(env: EnvironmentVars, repo: str) -> EnvironmentVars:	1✔
166	suffix = twine_env_suffix(repo)	×
167	return EnvironmentVars(	×
168	{key.rsplit(suffix, maxsplit=1)[0] if suffix else key: value for key, value in env.items()}
169	)
170
171
172	@rule	1✔
173	async def twine_upload(	1✔
174	request: PublishPythonPackageRequest,
175	twine_subsystem: TwineSubsystem,
176	global_options: GlobalOptions,
177	) -> PublishProcesses:
178	dists = tuple(	×
179	artifact.relpath
180	for pkg in request.packages
181	for artifact in pkg.artifacts
182	if artifact.relpath
183	)
184
NEW 185	if not dists:	×
186	return PublishProcesses()	×
187
188	twine_pex, packages_digest, config_files = await concurrently(	×
189	create_venv_pex(**implicitly(twine_subsystem.to_pex_request())),
190	merge_digests(MergeDigests(pkg.digest for pkg in request.packages)),
191	find_config_file(twine_subsystem.config_request()),
192	)
193
194	ca_cert_request = twine_subsystem.ca_certs_digest_request(global_options.ca_certs_path)	×
195	ca_cert = (	×
196	await digest_to_snapshot(**implicitly({ca_cert_request: CreateDigest}))
197	if ca_cert_request
198	else None
199	)
200	ca_cert_digest = (ca_cert.digest,) if ca_cert else ()	×
201
202	input_digest = await merge_digests(	×
203	MergeDigests((packages_digest, config_files.snapshot.digest, *ca_cert_digest))
204	)
205	pex_proc_requests: list[VenvPexProcess] = []	×
206	twine_envs = await concurrently(	×
207	environment_vars_subset(**implicitly({twine_env_request(repo): EnvironmentVarsRequest}))
208	for repo in request.field_set.repositories.value
209	)
210
211	for repo, env in zip(request.field_set.repositories.value, twine_envs):	×
212	pex_proc_requests.append(	×
213	VenvPexProcess(
214	twine_pex,
215	argv=twine_upload_args(twine_subsystem, config_files, repo, dists, ca_cert),
216	input_digest=input_digest,
217	extra_env=twine_env(env, repo),
218	description=repo,
219	)
220	)
221
222	processes = await concurrently(	×
223	setup_venv_pex_process(request, **implicitly()) for request in pex_proc_requests
224	)
225
226	return PublishProcesses(	×
227	PublishPackages(
228	names=dists,
229	process=InteractiveProcess.from_process(process),
230	description=process.description,
231	data=PublishOutputData({"repository": process.description}),
232	)
233	for process in processes
234	)
235
236
237	def rules():	1✔
238	return (	1✔
239	*collect_rules(),
240	*PublishPythonPackageFieldSet.rules(),
241	PythonDistribution.register_plugin_field(PythonRepositoriesField),
242	PythonDistribution.register_plugin_field(SkipTwineUploadField),
243	UnionRule(CheckSkipRequest, PythonDistCheckSkipRequest),
244	)

pantsbuild / pants / 21494992949

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