21264706899

Committed 22 Jan 2026 09:00PM UTC coverage: 80.255% (+1.6%) from 78.666%

Build # 21264706899

Build Type

Pull #23031

github

Committed by

web-flow

Commit Message

Merge 8385604a3 into d250c80fe

Pull Request Pull Request #23031: Enable publish without package

Run Details

32 of 60 new or added lines in 6 files covered. (53.33%)

2 existing lines in 2 files now uncovered.

78788 of 98172 relevant lines covered (80.26%)

3.36 hits per line

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

51.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.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 (
    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.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 get_output_data(self) -> PublishOutputData:
        return PublishOutputData(
            {
                "publisher": "twine",
                **super().get_output_data(),
            }
        )

    def package_before_publish(self, package_fs: PackageFieldSet) -> bool:
        return not self.skip_twine.value

    # I'd rather opt out early here, so we don't build unnecessarily, however the error feedback is
    # misleading and not very helpful in that case.
    #
    # @classmethod
    # def opt_out(cls, tgt: Target) -> bool:
    #     return not tgt[PythonRepositoriesField].value


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 twine_subsystem.skip or not dists:
        return PublishProcesses()

    # Too verbose to provide feedback as to why some packages were skipped?
    skip = None
    if request.field_set.skip_twine.value:
        skip = f"(by `{request.field_set.skip_twine.alias}` on {request.field_set.address})"
    elif not request.field_set.repositories.value:
        # I'd rather have used the opt_out mechanism on the field set, but that gives no hint as to
        # why the target was not applicable.
        skip = f"(no `{request.field_set.repositories.alias}` specified for {request.field_set.address})"

    if skip:
        return PublishProcesses(
            [
                PublishPackages(
                    names=dists,
                    description=skip,
                ),
            ]
        )

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

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

pantsbuild / pants / 21264706899

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