18913303678

Committed 29 Oct 2025 03:29PM UTC coverage: 80.004% (-0.3%) from 80.283%

Build # 18913303678

Build Type

push

github

Committed by

web-flow

Commit Message

Updating 3rd party lockfiles for PBS script and MyPy (#22796)

Also small bumps to fastapi and starlette, which updated over the last few days. Starlette updated for features, and then a security vulnerability.

Fastapi bumped just for starlette.

Run Details

88 of 93 new or added lines in 73 files covered. (94.62%)

221 existing lines in 15 files now uncovered.

77334 of 96663 relevant lines covered (80.0%)

5.2 hits per line

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

68.67

/src/python/pants/core/goals/repl.py

# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations

import os
from abc import ABC
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass
from typing import ClassVar

from pants.core.environments.rules import _warn_on_non_local_environments
from pants.engine.addresses import Addresses
from pants.engine.console import Console
from pants.engine.env_vars import CompleteEnvironmentVars
from pants.engine.environment import EnvironmentName
from pants.engine.fs import Digest
from pants.engine.goal import Goal, GoalSubsystem
from pants.engine.intrinsics import run_interactive_process
from pants.engine.process import InteractiveProcess
from pants.engine.rules import collect_rules, goal_rule, implicitly, rule
from pants.engine.target import FilteredTargets, Target
from pants.engine.unions import UnionMembership, union
from pants.option.option_types import ArgsListOption, BoolOption, StrOption
from pants.util.frozendict import FrozenDict
from pants.util.memo import memoized_property
from pants.util.strutil import softwrap


@union(in_scope_types=[EnvironmentName])
@dataclass(frozen=True)
class ReplImplementation(ABC):
    """A REPL implementation for a specific language or runtime.

    Proxies from the top-level `repl` goal to an actual implementation.
    """

    name: ClassVar[str]
    supports_args: ClassVar[bool]

    targets: Sequence[Target]

    def in_chroot(self, relpath: str) -> str:
        return os.path.join("{chroot}", relpath)

    @memoized_property
    def addresses(self) -> Addresses:
        return Addresses(t.address for t in self.targets)


class ReplSubsystem(GoalSubsystem):
    name = "repl"
    help = "Open a REPL with the specified code loadable."

    @classmethod
    def activated(cls, union_membership: UnionMembership) -> bool:
        return ReplImplementation in union_membership

    shell = StrOption(
        default=None,
        help="Override the automatically-detected REPL program for the target(s) specified.",
    )
    args = ArgsListOption(
        example="-i helloworld/main.py",
        tool_name="the repl program",
        passthrough=True,
        extra_help="Currently supported only for the ipython shell.",
    )
    restartable = BoolOption(
        default=False,
        help="True if the REPL should be restarted if its inputs have changed.",
    )


class Repl(Goal):
    subsystem_cls = ReplSubsystem
    environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY


@dataclass(frozen=True)
class ReplRequest:
    digest: Digest
    args: tuple[str, ...]
    extra_env: FrozenDict[str, str]
    immutable_input_digests: FrozenDict[str, Digest]
    append_only_caches: FrozenDict[str, str]
    run_in_workspace: bool

    def __init__(
        self,
        *,
        digest: Digest,
        args: Iterable[str],
        extra_env: Mapping[str, str] | None = None,
        immutable_input_digests: Mapping[str, Digest] | None = None,
        append_only_caches: Mapping[str, str] | None = None,
        run_in_workspace: bool = True,
    ) -> None:
        object.__setattr__(self, "digest", digest)
        object.__setattr__(self, "args", tuple(args))
        object.__setattr__(self, "extra_env", FrozenDict(extra_env or {}))
        object.__setattr__(
            self, "immutable_input_digests", FrozenDict(immutable_input_digests or {})
        )
        object.__setattr__(self, "append_only_caches", FrozenDict(append_only_caches or {}))
        object.__setattr__(self, "run_in_workspace", run_in_workspace)


@rule(polymorphic=True)
async def get_repl_request(
    impl: ReplImplementation,
    environment_name: EnvironmentName,
) -> ReplRequest:
    raise NotImplementedError()


@goal_rule
async def run_repl(
    console: Console,
    repl_subsystem: ReplSubsystem,
    specified_targets: FilteredTargets,
    union_membership: UnionMembership,
    complete_env: CompleteEnvironmentVars,
) -> Repl:
    await _warn_on_non_local_environments(specified_targets, "the `repl` goal")

    # TODO: When we support multiple languages, detect the default repl to use based
    #  on the targets.  For now we default to the python repl.
    repl_shell_name = repl_subsystem.shell or "python"
    implementations = {impl.name: impl for impl in union_membership[ReplImplementation]}
    repl_implementation_cls = implementations.get(repl_shell_name)
    if repl_implementation_cls is None:
        available = sorted(implementations.keys())
        console.print_stderr(
            softwrap(
                f"""
                {repr(repl_shell_name)} is not a registered REPL. Available REPLs (which may
                be specified through the option `--repl-shell`): {available}
                """
            )
        )
        return Repl(-1)

    if repl_subsystem.args and not repl_implementation_cls.supports_args:
        console.print_stderr(
            softwrap(
                f"""
                REPL goal does not support passing args to a {repr(repl_shell_name)} shell.
                """
            )
        )
        return Repl(-1)

    repl_impl = repl_implementation_cls(targets=specified_targets)
    request = await get_repl_request(**implicitly({repl_impl: ReplImplementation}))

    env = {**complete_env, **request.extra_env}

    result = await run_interactive_process(
        InteractiveProcess(
            argv=(*request.args, *repl_subsystem.args),
            env=env,
            input_digest=request.digest,
            run_in_workspace=request.run_in_workspace,
            restartable=repl_subsystem.restartable,
            immutable_input_digests=request.immutable_input_digests,
            append_only_caches=request.append_only_caches,
        ),
    )
    return Repl(result.exit_code)


def rules():
    return collect_rules()

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	16✔
4
5	import os	16✔
6	from abc import ABC	16✔
7	from collections.abc import Iterable, Mapping, Sequence	16✔
8	from dataclasses import dataclass	16✔
9	from typing import ClassVar	16✔
10
11	from pants.core.environments.rules import _warn_on_non_local_environments	16✔
12	from pants.engine.addresses import Addresses	16✔
13	from pants.engine.console import Console	16✔
14	from pants.engine.env_vars import CompleteEnvironmentVars	16✔
15	from pants.engine.environment import EnvironmentName	16✔
16	from pants.engine.fs import Digest	16✔
17	from pants.engine.goal import Goal, GoalSubsystem	16✔
18	from pants.engine.intrinsics import run_interactive_process	16✔
19	from pants.engine.process import InteractiveProcess	16✔
20	from pants.engine.rules import collect_rules, goal_rule, implicitly, rule	16✔
21	from pants.engine.target import FilteredTargets, Target	16✔
22	from pants.engine.unions import UnionMembership, union	16✔
23	from pants.option.option_types import ArgsListOption, BoolOption, StrOption	16✔
24	from pants.util.frozendict import FrozenDict	16✔
25	from pants.util.memo import memoized_property	16✔
26	from pants.util.strutil import softwrap	16✔
27
28
29	@union(in_scope_types=[EnvironmentName])	16✔
30	@dataclass(frozen=True)	16✔
31	class ReplImplementation(ABC):	16✔
32	"""A REPL implementation for a specific language or runtime.
33
34	Proxies from the top-level `repl` goal to an actual implementation.
35	"""
36
37	name: ClassVar[str]	16✔
38	supports_args: ClassVar[bool]	16✔
39
40	targets: Sequence[Target]	16✔
41
42	def in_chroot(self, relpath: str) -> str:	16✔
43	return os.path.join("{chroot}", relpath)	×
44
45	@memoized_property	16✔
46	def addresses(self) -> Addresses:	16✔
47	return Addresses(t.address for t in self.targets)	×
48
49
50	class ReplSubsystem(GoalSubsystem):	16✔
51	name = "repl"	16✔
52	help = "Open a REPL with the specified code loadable."	16✔
53
54	@classmethod	16✔
55	def activated(cls, union_membership: UnionMembership) -> bool:	16✔
56	return ReplImplementation in union_membership	×
57
58	shell = StrOption(	16✔
59	default=None,
60	help="Override the automatically-detected REPL program for the target(s) specified.",
61	)
62	args = ArgsListOption(	16✔
63	example="-i helloworld/main.py",
64	tool_name="the repl program",
65	passthrough=True,
66	extra_help="Currently supported only for the ipython shell.",
67	)
68	restartable = BoolOption(	16✔
69	default=False,
70	help="True if the REPL should be restarted if its inputs have changed.",
71	)
72
73
74	class Repl(Goal):	16✔
75	subsystem_cls = ReplSubsystem	16✔
76	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY	16✔
77
78
79	@dataclass(frozen=True)	16✔
80	class ReplRequest:	16✔
81	digest: Digest	16✔
82	args: tuple[str, ...]	16✔
83	extra_env: FrozenDict[str, str]	16✔
84	immutable_input_digests: FrozenDict[str, Digest]	16✔
85	append_only_caches: FrozenDict[str, str]	16✔
86	run_in_workspace: bool	16✔
87
88	def __init__(	16✔
89	self,
90	*,
91	digest: Digest,
92	args: Iterable[str],
93	extra_env: Mapping[str, str] \| None = None,
94	immutable_input_digests: Mapping[str, Digest] \| None = None,
95	append_only_caches: Mapping[str, str] \| None = None,
96	run_in_workspace: bool = True,
97	) -> None:
98	object.__setattr__(self, "digest", digest)	×
99	object.__setattr__(self, "args", tuple(args))	×
100	object.__setattr__(self, "extra_env", FrozenDict(extra_env or {}))	×
101	object.__setattr__(	×
102	self, "immutable_input_digests", FrozenDict(immutable_input_digests or {})
103	)
104	object.__setattr__(self, "append_only_caches", FrozenDict(append_only_caches or {}))	×
105	object.__setattr__(self, "run_in_workspace", run_in_workspace)	×
106
107
108	@rule(polymorphic=True)	16✔
109	async def get_repl_request(	16✔
110	impl: ReplImplementation,
111	environment_name: EnvironmentName,
112	) -> ReplRequest:
113	raise NotImplementedError()	×
114
115
116	@goal_rule	16✔
117	async def run_repl(	16✔
118	console: Console,
119	repl_subsystem: ReplSubsystem,
120	specified_targets: FilteredTargets,
121	union_membership: UnionMembership,
122	complete_env: CompleteEnvironmentVars,
123	) -> Repl:
124	await _warn_on_non_local_environments(specified_targets, "the `repl` goal")	×
125
126	# TODO: When we support multiple languages, detect the default repl to use based
127	# on the targets. For now we default to the python repl.
128	repl_shell_name = repl_subsystem.shell or "python"	×
129	implementations = {impl.name: impl for impl in union_membership[ReplImplementation]}	×
NEW 130	repl_implementation_cls = implementations.get(repl_shell_name)	×
UNCOV 131	if repl_implementation_cls is None:	×
132	available = sorted(implementations.keys())	×
133	console.print_stderr(	×
134	softwrap(
135	f"""
136	{repr(repl_shell_name)} is not a registered REPL. Available REPLs (which may
137	be specified through the option `--repl-shell`): {available}
138	"""
139	)
140	)
141	return Repl(-1)	×
142
143	if repl_subsystem.args and not repl_implementation_cls.supports_args:	×
144	console.print_stderr(	×
145	softwrap(
146	f"""
147	REPL goal does not support passing args to a {repr(repl_shell_name)} shell.
148	"""
149	)
150	)
151	return Repl(-1)	×
152
153	repl_impl = repl_implementation_cls(targets=specified_targets)	×
154	request = await get_repl_request(**implicitly({repl_impl: ReplImplementation}))	×
155
156	env = {complete_env, request.extra_env}	×
157
158	result = await run_interactive_process(	×
159	InteractiveProcess(
160	argv=(request.args, repl_subsystem.args),
161	env=env,
162	input_digest=request.digest,
163	run_in_workspace=request.run_in_workspace,
164	restartable=repl_subsystem.restartable,
165	immutable_input_digests=request.immutable_input_digests,
166	append_only_caches=request.append_only_caches,
167	),
168	)
169	return Repl(result.exit_code)	×
170
171
172	def rules():	16✔
173	return collect_rules()	16✔

pantsbuild / pants / 18913303678

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