18517631058

Committed 15 Oct 2025 04:18AM UTC coverage: 69.207% (-11.1%) from 80.267%

Build # 18517631058

Build Type

Pull #22745

github

Committed by

web-flow

Commit Message

Merge 642a76ca1 into 99919310e

Pull Request Pull Request #22745: [windows] Add windows support in the stdio crate.

Run Details

53815 of 77759 relevant lines covered (69.21%)

2.42 hits per line

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

0.0

/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.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(),
            }
        )

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

pantsbuild / pants / 18517631058

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