21572405674

Committed 01 Feb 2026 11:36PM UTC coverage: 80.273% (-0.05%) from 80.324%

Build # 21572405674

Build Type

push

github

Committed by

web-flow

Commit Message

Remove support for Get (#23062)

Delete the bulk of the code needed to
support `Get` (and `Effect`).

Follow up changes will further trim code
that is no longer needed, but this
is already a big enough step for now.

Run Details

23 of 27 new or added lines in 5 files covered. (85.19%)

49 existing lines in 3 files now uncovered.

78274 of 97510 relevant lines covered (80.27%)

3.1 hits per line

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

89.33

/src/python/pants/engine/internals/rule_visitor.py

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

import ast
import inspect
import itertools
import logging
import sys
from collections.abc import Callable, Iterator, Sequence
from contextlib import contextmanager
from dataclasses import dataclass
from types import ModuleType
from typing import Any, get_type_hints

import typing_extensions

from pants.base.exceptions import RuleTypeError
from pants.engine.internals.selectors import (
    AwaitableConstraints,
    concurrently,
)
from pants.util.memo import memoized
from pants.util.strutil import softwrap
from pants.util.typing import patch_forward_ref

logger = logging.getLogger(__name__)
patch_forward_ref()


def _get_starting_indent(source: str) -> int:
    """Used to remove leading indentation from `source` so ast.parse() doesn't raise an
    exception."""
    if source.startswith(" "):
        return sum(1 for _ in itertools.takewhile(lambda c: c in {" ", b" "}, source))
    return 0


def _node_str(node: Any) -> str:
    if isinstance(node, ast.Name):
        return node.id
    if isinstance(node, ast.Attribute):
        return ".".join([_node_str(node.value), node.attr])
    if isinstance(node, ast.Call):
        return _node_str(node.func)
    if sys.version_info[0:2] < (3, 8):
        if isinstance(node, ast.Str):
            return node.s
    else:
        if isinstance(node, ast.Constant):
            return str(node.value)
    return str(node)


PANTS_RULE_DESCRIPTORS_MODULE_KEY = "__pants_rule_descriptors__"


@dataclass(frozen=True)
class RuleDescriptor:
    """The data we glean about a rule by examining its AST.

    This will be lazily invoked in the first `@rule` decorator in a module. Therefore it will parse
    the AST *before* the module code is fully evaluated, and so the return type may not yet exist as
    a parsed type. So we store it here as a str and look it up later.
    """

    module_name: str
    rule_name: str
    return_type: str

    @property
    def rule_id(self) -> str:
        # TODO: Handle canonical_name/canonical_name_suffix?
        return f"{self.module_name}.{self.rule_name}"


def get_module_scope_rules(module: ModuleType) -> tuple[RuleDescriptor, ...]:
    """Get descriptors for @rules defined at the top level of the given module.

    We discover these top-level rules and rule helpers in the module by examining the AST.
    This means that while executing the `@rule` decorator of a rule1(), the descriptor of a rule2()
    defined later in the module is already known.  This allows rule1() and rule2() to be
    mutually recursive.

    Note that we don't support recursive rules defined dynamically in inner scopes.
    """
    descriptors = getattr(module, PANTS_RULE_DESCRIPTORS_MODULE_KEY, None)
    if descriptors is None:
        descriptors = []
        for node in ast.iter_child_nodes(ast.parse(inspect.getsource(module))):
            if isinstance(node, ast.AsyncFunctionDef) and isinstance(node.returns, ast.Name):
                descriptors.append(RuleDescriptor(module.__name__, node.name, node.returns.id))
        descriptors = tuple(descriptors)
        setattr(module, PANTS_RULE_DESCRIPTORS_MODULE_KEY, descriptors)

    return descriptors


class _TypeStack:
    """The types and rules that a @rule can refer to in its input/outputs, or its awaitables.

    We construct this data through a mix of inspection of types already parsed by Python,
    and descriptors we infer from the AST. This allows us to support mutual recursion between
    rules defined in the same module (the @rule descriptor of the earlier rule can know enough
    about the later rule it calls to set up its own awaitables correctly).

    This logic is necessarily heuristic. It works for well-behaved code, but may be defeated
    by metaprogramming, aliasing, shadowing and so on.
    """

    def __init__(self, func: Callable) -> None:
        self._stack: list[dict[str, Any]] = []
        self.root = sys.modules[func.__module__]

        # We fall back to descriptors last, so that we get parsed objects whenever possible,
        # as those are less susceptible to limitations of the heuristics.
        self.push({descr.rule_name: descr for descr in get_module_scope_rules(self.root)})
        self.push(self.root)
        self._push_function_closures(func)
        # Rule args will be pushed later, as we handle them.

    def __getitem__(self, name: str) -> Any:
        for ns in reversed(self._stack):
            if name in ns:
                return ns[name]
        return self.root.__builtins__.get(name, None)

    def __setitem__(self, name: str, value: Any) -> None:
        self._stack[-1][name] = value

    def _push_function_closures(self, func: Callable) -> None:
        try:
            closurevars = [c for c in inspect.getclosurevars(func) if isinstance(c, dict)]
        except ValueError:
            return

        for closures in closurevars:
            self.push(closures)

    def push(self, frame: object) -> None:
        ns = dict(frame if isinstance(frame, dict) else frame.__dict__)
        self._stack.append(ns)

    def pop(self) -> None:
        assert len(self._stack) > 1
        self._stack.pop()


def _lookup_annotation(obj: Any, attr: str) -> Any:
    """Get type associated with a particular attribute on object. This can get hairy, especially on
    Python <3.10.

    https://docs.python.org/3/howto/annotations.html#accessing-the-annotations-dict-of-an-object-in-python-3-9-and-older
    """
    if hasattr(obj, attr):
        return getattr(obj, attr)
    else:
        try:
            return get_type_hints(obj).get(attr)
        except (NameError, TypeError):
            return None


def _lookup_return_type(func: Callable, check: bool = False) -> Any:
    ret = _lookup_annotation(func, "return")
    typ = typing_extensions.get_origin(ret)
    if isinstance(typ, type):
        args = typing_extensions.get_args(ret)
        if issubclass(typ, (list, set, tuple)):
            return tuple(args)
    if check and ret is None:
        func_file = inspect.getsourcefile(func)
        func_line = func.__code__.co_firstlineno
        raise TypeError(
            f"Failed to look up return type hint for `{func.__name__}` in {func_file}:{func_line}"
        )
    return ret


class _AwaitableCollector(ast.NodeVisitor):
    def __init__(self, func: Callable):
        self.func = func
        source = inspect.getsource(func) or "<string>"
        beginning_indent = _get_starting_indent(source)
        if beginning_indent:
            source = "\n".join(line[beginning_indent:] for line in source.split("\n"))

        self.source_file = inspect.getsourcefile(func) or "<unknown>"

        self.types = _TypeStack(func)
        self.awaitables: list[AwaitableConstraints] = []
        self.visit(ast.parse(source))

    def _format(self, node: ast.AST, msg: str) -> str:
        lineno: str = "<unknown>"
        if isinstance(node, (ast.expr, ast.stmt)):
            lineno = str(node.lineno + self.func.__code__.co_firstlineno - 1)
        return f"{self.source_file}:{lineno}: {msg}"

    def _lookup(self, attr: ast.expr) -> Any:
        names = []
        while isinstance(attr, ast.Attribute):
            names.append(attr.attr)
            attr = attr.value
        # NB: attr could be a constant, like `",".join()`
        id = getattr(attr, "id", None)
        if id is not None:
            names.append(id)

        if not names:
            return attr

        name = names.pop()
        result = self.types[name]
        while result is not None and names:
            result = _lookup_annotation(result, names.pop())
        return result

    def _missing_type_error(self, node: ast.AST, context: str) -> str:
        mod = self.types.root.__name__
        return self._format(
            node,
            softwrap(
                f"""
                Could not resolve type for `{_node_str(node)}` in module {mod}.

                {context}
                """
            ),
        )

    def _check_constraint_arg_type(self, resolved: Any, node: ast.AST) -> type:
        if resolved is None:
            raise RuleTypeError(
                self._missing_type_error(
                    node, context="This may be a limitation of the Pants rule type inference."
                )
            )
        elif not isinstance(resolved, type):
            raise RuleTypeError(
                self._format(
                    node,
                    f"Expected a type, but got: {type(resolved).__name__} {_node_str(resolved)!r}",
                )
            )
        return resolved

    def _get_inputs(self, input_nodes: Sequence[Any]) -> tuple[Sequence[Any], list[Any]]:
        if not input_nodes:
            return input_nodes, []
        if len(input_nodes) != 1:
            return input_nodes, [self._lookup(input_nodes[0])]

        input_constructor = input_nodes[0]
        if isinstance(input_constructor, ast.Call):
            cls_or_func = self._lookup(input_constructor.func)
            try:
                type_ = (
                    _lookup_return_type(cls_or_func, check=True)
                    if not isinstance(cls_or_func, type)
                    else cls_or_func
                )
            except TypeError as e:
                raise RuleTypeError(self._missing_type_error(input_constructor, str(e))) from e
            return [input_constructor.func], [type_]
        elif isinstance(input_constructor, ast.Dict):
            return input_constructor.values, [self._lookup(v) for v in input_constructor.values]
        else:
            return input_nodes, [self._lookup(n) for n in input_nodes]

    def _get_byname_awaitable(
        self, rule_id: str, rule_func: Callable | RuleDescriptor, call_node: ast.Call
    ) -> AwaitableConstraints:
        if isinstance(rule_func, RuleDescriptor):
            # At this point we expect the return type to be defined, so its source code
            # must precede that of the rule invoking the awaitable that returns it.
            output_type = self.types[rule_func.return_type]
        else:
            output_type = _lookup_return_type(rule_func, check=True)

        # To support explicit positional arguments, we record the number passed positionally.
        # TODO: To support keyword arguments, we would additionally need to begin recording the
        # argument names of kwargs. But positional-only callsites can avoid those allocations.
        explicit_args_arity = len(call_node.args)

        input_types: tuple[type, ...]
        if not call_node.keywords:
            input_types = ()
        elif (
            len(call_node.keywords) == 1
            and not call_node.keywords[0].arg
            and isinstance(implicitly_call := call_node.keywords[0].value, ast.Call)
            and self._lookup(implicitly_call.func).__name__ == "implicitly"
        ):
            input_nodes, input_type_nodes = self._get_inputs(implicitly_call.args)
            input_types = tuple(
                self._check_constraint_arg_type(input_type, input_node)
                for input_type, input_node in zip(input_type_nodes, input_nodes)
            )
        else:
            explanation = self._format(
                call_node,
                "Expected an `**implicitly(..)` application as the only keyword input.",
            )
            raise ValueError(
                f"Invalid call. {explanation} failed in a call to {rule_id} in {self.source_file}."
            )

        return AwaitableConstraints(
            rule_id,
            output_type,
            explicit_args_arity,
            input_types,
            # TODO: Extract this from the callee? Currently only intrinsics can be Effects, so need
            # to figure out their new syntax first.
            is_effect=False,
        )

    def visit_Call(self, call_node: ast.Call) -> None:
        func = self._lookup(call_node.func)
        if func is not None:
            if (inspect.isfunction(func) or isinstance(func, RuleDescriptor)) and (
                rule_id := getattr(func, "rule_id", None)
            ) is not None:
                # Is a direct `@rule` call.
                self.awaitables.append(self._get_byname_awaitable(rule_id, func, call_node))
            elif inspect.iscoroutinefunction(func):
                # Is a call to a "rule helper".
                self.awaitables.extend(collect_awaitables(func))

        self.generic_visit(call_node)

    def visit_AsyncFunctionDef(self, rule: ast.AsyncFunctionDef) -> None:
        with self._visit_rule_args(rule.args):
            self.generic_visit(rule)

    def visit_FunctionDef(self, rule: ast.FunctionDef) -> None:
        with self._visit_rule_args(rule.args):
            self.generic_visit(rule)

    @contextmanager
    def _visit_rule_args(self, node: ast.arguments) -> Iterator[None]:
        self.types.push(
            {
                a.arg: self.types[a.annotation.id]
                for a in node.args
                if isinstance(a.annotation, ast.Name)
            }
        )
        try:
            yield
        finally:
            self.types.pop()

    def visit_Assign(self, assign_node: ast.Assign) -> None:
        awaitables_idx = len(self.awaitables)
        self.generic_visit(assign_node)
        collected_awaitables = self.awaitables[awaitables_idx:]
        value = None
        node: ast.AST = assign_node
        while True:
            if isinstance(node, (ast.Assign, ast.Await)):
                node = node.value
                continue
            if isinstance(node, ast.Call):
                f = self._lookup(node.func)
                if f is concurrently:
                    value = tuple(get.output_type for get in collected_awaitables)
                elif f is not None:
                    value = _lookup_return_type(f)
            elif isinstance(node, (ast.Name, ast.Attribute)):
                value = self._lookup(node)
            break

        for tgt in assign_node.targets:
            if isinstance(tgt, ast.Name):
                names = [tgt.id]
                values = [value]
            elif isinstance(tgt, ast.Tuple):
                names = [el.id for el in tgt.elts if isinstance(el, ast.Name)]
                values = value or itertools.cycle([None])  # type: ignore[assignment]
            else:
                # subscript, etc..
                continue
            try:
                for name, value in zip(names, values):
                    self.types[name] = value
            except TypeError as e:
                logger.debug(
                    self._format(
                        node,
                        softwrap(
                            f"""
                            Rule visitor failed to inspect assignment expression for
                            {names} - {values}:

                            {e}
                            """
                        ),
                    )
                )


@memoized
def collect_awaitables(func: Callable) -> list[AwaitableConstraints]:
    return _AwaitableCollector(func).awaitables

1	# Copyright 2022 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	from __future__ import annotations	11✔
4
5	import ast	11✔
6	import inspect	11✔
7	import itertools	11✔
8	import logging	11✔
9	import sys	11✔
10	from collections.abc import Callable, Iterator, Sequence	11✔
11	from contextlib import contextmanager	11✔
12	from dataclasses import dataclass	11✔
13	from types import ModuleType	11✔
14	from typing import Any, get_type_hints	11✔
15
16	import typing_extensions	11✔
17
18	from pants.base.exceptions import RuleTypeError	11✔
19	from pants.engine.internals.selectors import (	11✔
20	AwaitableConstraints,
21	concurrently,
22	)
23	from pants.util.memo import memoized	11✔
24	from pants.util.strutil import softwrap	11✔
25	from pants.util.typing import patch_forward_ref	11✔
26
27	logger = logging.getLogger(__name__)	11✔
28	patch_forward_ref()	11✔
29
30
31	def _get_starting_indent(source: str) -> int:	11✔
32	"""Used to remove leading indentation from `source` so ast.parse() doesn't raise an
33	exception."""
34	if source.startswith(" "):	11✔
35	return sum(1 for _ in itertools.takewhile(lambda c: c in {" ", b" "}, source))	11✔
36	return 0	11✔
37
38
39	def _node_str(node: Any) -> str:	11✔
UNCOV 40	if isinstance(node, ast.Name):	×
UNCOV 41	return node.id	×
UNCOV 42	if isinstance(node, ast.Attribute):	×
UNCOV 43	return ".".join([_node_str(node.value), node.attr])	×
UNCOV 44	if isinstance(node, ast.Call):	×
UNCOV 45	return _node_str(node.func)	×
UNCOV 46	if sys.version_info[0:2] < (3, 8):	×
47	if isinstance(node, ast.Str):	×
48	return node.s	×
49	else:
UNCOV 50	if isinstance(node, ast.Constant):	×
UNCOV 51	return str(node.value)	×
52	return str(node)	×
53
54
55	PANTS_RULE_DESCRIPTORS_MODULE_KEY = "__pants_rule_descriptors__"	11✔
56
57
58	@dataclass(frozen=True)	11✔
59	class RuleDescriptor:	11✔
60	"""The data we glean about a rule by examining its AST.
61
62	This will be lazily invoked in the first `@rule` decorator in a module. Therefore it will parse
63	the AST before the module code is fully evaluated, and so the return type may not yet exist as
64	a parsed type. So we store it here as a str and look it up later.
65	"""
66
67	module_name: str	11✔
68	rule_name: str	11✔
69	return_type: str	11✔
70
71	@property	11✔
72	def rule_id(self) -> str:	11✔
73	# TODO: Handle canonical_name/canonical_name_suffix?
74	return f"{self.module_name}.{self.rule_name}"	11✔
75
76
77	def get_module_scope_rules(module: ModuleType) -> tuple[RuleDescriptor, ...]:	11✔
78	"""Get descriptors for @rules defined at the top level of the given module.
79
80	We discover these top-level rules and rule helpers in the module by examining the AST.
81	This means that while executing the `@rule` decorator of a rule1(), the descriptor of a rule2()
82	defined later in the module is already known. This allows rule1() and rule2() to be
83	mutually recursive.
84
85	Note that we don't support recursive rules defined dynamically in inner scopes.
86	"""
87	descriptors = getattr(module, PANTS_RULE_DESCRIPTORS_MODULE_KEY, None)	11✔
88	if descriptors is None:	11✔
89	descriptors = []	11✔
90	for node in ast.iter_child_nodes(ast.parse(inspect.getsource(module))):	11✔
91	if isinstance(node, ast.AsyncFunctionDef) and isinstance(node.returns, ast.Name):	11✔
92	descriptors.append(RuleDescriptor(module.__name__, node.name, node.returns.id))	11✔
93	descriptors = tuple(descriptors)	11✔
94	setattr(module, PANTS_RULE_DESCRIPTORS_MODULE_KEY, descriptors)	11✔
95
96	return descriptors	11✔
97
98
99	class _TypeStack:	11✔
100	"""The types and rules that a @rule can refer to in its input/outputs, or its awaitables.
101
102	We construct this data through a mix of inspection of types already parsed by Python,
103	and descriptors we infer from the AST. This allows us to support mutual recursion between
104	rules defined in the same module (the @rule descriptor of the earlier rule can know enough
105	about the later rule it calls to set up its own awaitables correctly).
106
107	This logic is necessarily heuristic. It works for well-behaved code, but may be defeated
108	by metaprogramming, aliasing, shadowing and so on.
109	"""
110
111	def __init__(self, func: Callable) -> None:	11✔
112	self._stack: list[dict[str, Any]] = []	11✔
113	self.root = sys.modules[func.__module__]	11✔
114
115	# We fall back to descriptors last, so that we get parsed objects whenever possible,
116	# as those are less susceptible to limitations of the heuristics.
117	self.push({descr.rule_name: descr for descr in get_module_scope_rules(self.root)})	11✔
118	self.push(self.root)	11✔
119	self._push_function_closures(func)	11✔
120	# Rule args will be pushed later, as we handle them.
121
122	def __getitem__(self, name: str) -> Any:	11✔
123	for ns in reversed(self._stack):	11✔
124	if name in ns:	11✔
125	return ns[name]	11✔
126	return self.root.__builtins__.get(name, None)	11✔
127
128	def __setitem__(self, name: str, value: Any) -> None:	11✔
129	self._stack[-1][name] = value	11✔
130
131	def _push_function_closures(self, func: Callable) -> None:	11✔
132	try:	11✔
133	closurevars = [c for c in inspect.getclosurevars(func) if isinstance(c, dict)]	11✔
134	except ValueError:	11✔
135	return	11✔
136
137	for closures in closurevars:	11✔
138	self.push(closures)	11✔
139
140	def push(self, frame: object) -> None:	11✔
141	ns = dict(frame if isinstance(frame, dict) else frame.__dict__)	11✔
142	self._stack.append(ns)	11✔
143
144	def pop(self) -> None:	11✔
145	assert len(self._stack) > 1	11✔
146	self._stack.pop()	11✔
147
148
149	def _lookup_annotation(obj: Any, attr: str) -> Any:	11✔
150	"""Get type associated with a particular attribute on object. This can get hairy, especially on
151	Python <3.10.
152
153	https://docs.python.org/3/howto/annotations.html#accessing-the-annotations-dict-of-an-object-in-python-3-9-and-older
154	"""
155	if hasattr(obj, attr):	11✔
156	return getattr(obj, attr)	11✔
157	else:
158	try:	11✔
159	return get_type_hints(obj).get(attr)	11✔
160	except (NameError, TypeError):	11✔
161	return None	11✔
162
163
164	def _lookup_return_type(func: Callable, check: bool = False) -> Any:	11✔
165	ret = _lookup_annotation(func, "return")	11✔
166	typ = typing_extensions.get_origin(ret)	11✔
167	if isinstance(typ, type):	11✔
168	args = typing_extensions.get_args(ret)	11✔
169	if issubclass(typ, (list, set, tuple)):	11✔
170	return tuple(args)	11✔
171	if check and ret is None:	11✔
UNCOV 172	func_file = inspect.getsourcefile(func)	×
UNCOV 173	func_line = func.__code__.co_firstlineno	×
UNCOV 174	raise TypeError(	×
175	f"Failed to look up return type hint for `{func.__name__}` in {func_file}:{func_line}"
176	)
177	return ret	11✔
178
179
180	class _AwaitableCollector(ast.NodeVisitor):	11✔
181	def __init__(self, func: Callable):	11✔
182	self.func = func	11✔
183	source = inspect.getsource(func) or "<string>"	11✔
184	beginning_indent = _get_starting_indent(source)	11✔
185	if beginning_indent:	11✔
186	source = "\n".join(line[beginning_indent:] for line in source.split("\n"))	11✔
187
188	self.source_file = inspect.getsourcefile(func) or "<unknown>"	11✔
189
190	self.types = _TypeStack(func)	11✔
191	self.awaitables: list[AwaitableConstraints] = []	11✔
192	self.visit(ast.parse(source))	11✔
193
194	def _format(self, node: ast.AST, msg: str) -> str:	11✔
195	lineno: str = "<unknown>"	5✔
196	if isinstance(node, (ast.expr, ast.stmt)):	5✔
197	lineno = str(node.lineno + self.func.__code__.co_firstlineno - 1)	5✔
198	return f"{self.source_file}:{lineno}: {msg}"	5✔
199
200	def _lookup(self, attr: ast.expr) -> Any:	11✔
201	names = []	11✔
202	while isinstance(attr, ast.Attribute):	11✔
203	names.append(attr.attr)	11✔
204	attr = attr.value	11✔
205	# NB: attr could be a constant, like `",".join()`
206	id = getattr(attr, "id", None)	11✔
207	if id is not None:	11✔
208	names.append(id)	11✔
209
210	if not names:	11✔
211	return attr	11✔
212
213	name = names.pop()	11✔
214	result = self.types[name]	11✔
215	while result is not None and names:	11✔
216	result = _lookup_annotation(result, names.pop())	11✔
217	return result	11✔
218
219	def _missing_type_error(self, node: ast.AST, context: str) -> str:	11✔
UNCOV 220	mod = self.types.root.__name__	×
UNCOV 221	return self._format(	×
222	node,
223	softwrap(
224	f"""
225	Could not resolve type for `{_node_str(node)}` in module {mod}.
226
227	{context}
228	"""
229	),
230	)
231
232	def _check_constraint_arg_type(self, resolved: Any, node: ast.AST) -> type:	11✔
233	if resolved is None:	11✔
UNCOV 234	raise RuleTypeError(	×
235	self._missing_type_error(
236	node, context="This may be a limitation of the Pants rule type inference."
237	)
238	)
239	elif not isinstance(resolved, type):	11✔
UNCOV 240	raise RuleTypeError(	×
241	self._format(
242	node,
243	f"Expected a type, but got: {type(resolved).__name__} {_node_str(resolved)!r}",
244	)
245	)
246	return resolved	11✔
247
248	def _get_inputs(self, input_nodes: Sequence[Any]) -> tuple[Sequence[Any], list[Any]]:	11✔
249	if not input_nodes:	11✔
250	return input_nodes, []	11✔
251	if len(input_nodes) != 1:	11✔
UNCOV 252	return input_nodes, [self._lookup(input_nodes[0])]	×
253
254	input_constructor = input_nodes[0]	11✔
255	if isinstance(input_constructor, ast.Call):	11✔
256	cls_or_func = self._lookup(input_constructor.func)	11✔
257	try:	11✔
258	type_ = (	11✔
259	_lookup_return_type(cls_or_func, check=True)
260	if not isinstance(cls_or_func, type)
261	else cls_or_func
262	)
UNCOV 263	except TypeError as e:	×
UNCOV 264	raise RuleTypeError(self._missing_type_error(input_constructor, str(e))) from e	×
265	return [input_constructor.func], [type_]	11✔
266	elif isinstance(input_constructor, ast.Dict):	11✔
267	return input_constructor.values, [self._lookup(v) for v in input_constructor.values]	11✔
268	else:
269	return input_nodes, [self._lookup(n) for n in input_nodes]	11✔
270
271	def _get_byname_awaitable(	11✔
272	self, rule_id: str, rule_func: Callable \| RuleDescriptor, call_node: ast.Call
273	) -> AwaitableConstraints:
274	if isinstance(rule_func, RuleDescriptor):	11✔
275	# At this point we expect the return type to be defined, so its source code
276	# must precede that of the rule invoking the awaitable that returns it.
277	output_type = self.types[rule_func.return_type]	11✔
278	else:
279	output_type = _lookup_return_type(rule_func, check=True)	11✔
280
281	# To support explicit positional arguments, we record the number passed positionally.
282	# TODO: To support keyword arguments, we would additionally need to begin recording the
283	# argument names of kwargs. But positional-only callsites can avoid those allocations.
284	explicit_args_arity = len(call_node.args)	11✔
285
286	input_types: tuple[type, ...]
287	if not call_node.keywords:	11✔
288	input_types = ()	11✔
289	elif (	11✔
290	len(call_node.keywords) == 1
291	and not call_node.keywords[0].arg
292	and isinstance(implicitly_call := call_node.keywords[0].value, ast.Call)
293	and self._lookup(implicitly_call.func).__name__ == "implicitly"
294	):
295	input_nodes, input_type_nodes = self._get_inputs(implicitly_call.args)	11✔
296	input_types = tuple(	11✔
297	self._check_constraint_arg_type(input_type, input_node)
298	for input_type, input_node in zip(input_type_nodes, input_nodes)
299	)
300	else:
NEW 301	explanation = self._format(	×
302	call_node,
303	"Expected an `**implicitly(..)` application as the only keyword input.",
304	)
NEW 305	raise ValueError(	×
306	f"Invalid call. {explanation} failed in a call to {rule_id} in {self.source_file}."
307	)
308
309	return AwaitableConstraints(	11✔
310	rule_id,
311	output_type,
312	explicit_args_arity,
313	input_types,
314	# TODO: Extract this from the callee? Currently only intrinsics can be Effects, so need
315	# to figure out their new syntax first.
316	is_effect=False,
317	)
318
319	def visit_Call(self, call_node: ast.Call) -> None:	11✔
320	func = self._lookup(call_node.func)	11✔
321	if func is not None:	11✔
322	if (inspect.isfunction(func) or isinstance(func, RuleDescriptor)) and (	11✔
323	rule_id := getattr(func, "rule_id", None)
324	) is not None:
325	# Is a direct `@rule` call.
326	self.awaitables.append(self._get_byname_awaitable(rule_id, func, call_node))	11✔
327	elif inspect.iscoroutinefunction(func):	11✔
328	# Is a call to a "rule helper".
329	self.awaitables.extend(collect_awaitables(func))	11✔
330
331	self.generic_visit(call_node)	11✔
332
333	def visit_AsyncFunctionDef(self, rule: ast.AsyncFunctionDef) -> None:	11✔
334	with self._visit_rule_args(rule.args):	11✔
335	self.generic_visit(rule)	11✔
336
337	def visit_FunctionDef(self, rule: ast.FunctionDef) -> None:	11✔
338	with self._visit_rule_args(rule.args):	11✔
339	self.generic_visit(rule)	11✔
340
341	@contextmanager	11✔
342	def _visit_rule_args(self, node: ast.arguments) -> Iterator[None]:	11✔
343	self.types.push(	11✔
344	{
345	a.arg: self.types[a.annotation.id]
346	for a in node.args
347	if isinstance(a.annotation, ast.Name)
348	}
349	)
350	try:	11✔
351	yield	11✔
352	finally:
353	self.types.pop()	11✔
354
355	def visit_Assign(self, assign_node: ast.Assign) -> None:	11✔
356	awaitables_idx = len(self.awaitables)	11✔
357	self.generic_visit(assign_node)	11✔
358	collected_awaitables = self.awaitables[awaitables_idx:]	11✔
359	value = None	11✔
360	node: ast.AST = assign_node	11✔
361	while True:	11✔
362	if isinstance(node, (ast.Assign, ast.Await)):	11✔
363	node = node.value	11✔
364	continue	11✔
365	if isinstance(node, ast.Call):	11✔
366	f = self._lookup(node.func)	11✔
367	if f is concurrently:	11✔
368	value = tuple(get.output_type for get in collected_awaitables)	11✔
369	elif f is not None:	11✔
370	value = _lookup_return_type(f)	11✔
371	elif isinstance(node, (ast.Name, ast.Attribute)):	11✔
372	value = self._lookup(node)	11✔
373	break	11✔
374
375	for tgt in assign_node.targets:	11✔
376	if isinstance(tgt, ast.Name):	11✔
377	names = [tgt.id]	11✔
378	values = [value]	11✔
379	elif isinstance(tgt, ast.Tuple):	11✔
380	names = [el.id for el in tgt.elts if isinstance(el, ast.Name)]	11✔
381	values = value or itertools.cycle([None]) # type: ignore[assignment]	11✔
382	else:
383	# subscript, etc..
384	continue	11✔
385	try:	11✔
386	for name, value in zip(names, values):	11✔
387	self.types[name] = value	11✔
388	except TypeError as e:	5✔
389	logger.debug(	5✔
390	self._format(
391	node,
392	softwrap(
393	f"""
394	Rule visitor failed to inspect assignment expression for
395	{names} - {values}:
396
397	{e}
398	"""
399	),
400	)
401	)
402
403
404	@memoized	11✔
405	def collect_awaitables(func: Callable) -> list[AwaitableConstraints]:	11✔
406	return _AwaitableCollector(func).awaitables	11✔

pantsbuild / pants / 21572405674

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