19529437518

Committed 20 Nov 2025 07:44AM UTC coverage: 78.884% (-1.4%) from 80.302%

Build # 19529437518

Build Type

push

github

Committed by

web-flow

Commit Message

nfpm.native_libs: Add RPM package depends from packaged pex_binaries (#22899)

## PR Series Overview

This is the second in a series of PRs that introduces a new backend:
`pants.backend.npm.native_libs`
Initially, the backend will be available as:
`pants.backend.experimental.nfpm.native_libs`

I proposed this new backend (originally named `bindeps`) in discussion
#22396.

This backend will inspect ELF bin/lib files (like `lib*.so`) in packaged
contents (for this PR series, only in `pex_binary` targets) to identify
package dependency metadata and inject that metadata on the relevant
`nfpm_deb_package` or `nfpm_rpm_package` targets. Effectively, it will
provide an approximation of these native packager features:
- `rpm`: `rpmdeps` + `elfdeps`
- `deb`: `dh_shlibdeps` + `dpkg-shlibdeps` (These substitute
`${shlibs:Depends}` in debian control files have)

### Goal: Host-agnostic package builds

This pants backend is designed to be host-agnostic, like
[nFPM](https://nfpm.goreleaser.com/).

Native packaging tools are often restricted to a single release of a
single distro. Unlike native package builders, this new pants backend
does not use any of those distro-specific or distro-release-specific
utilities or local package databases. This new backend should be able to
run (help with building deb and rpm packages) anywhere that pants can
run (MacOS, rpm linux distros, deb linux distros, other linux distros,
docker, ...).

### Previous PRs in series

- #22873

## PR Overview

This PR adds rules in `nfpm.native_libs` to add package dependency
metadata to `nfpm_rpm_package`. The 2 new rules are:

- `inject_native_libs_dependencies_in_package_fields`:

    - An implementation of the polymorphic rule `inject_nfpm_package_fields`.
      This rule is low priority (`priority = 2`) so that in-repo plugins can
      override/augment what it injects. (See #22864)

    - Rule logic overview:
        - find any pex_binaries that will be packaged in an `nfpm_rpm_package`
   ... (continued)

Run Details

96 of 118 new or added lines in 3 files covered. (81.36%)

910 existing lines in 53 files now uncovered.

73897 of 93678 relevant lines covered (78.88%)

3.21 hits per line

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

85.66

/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 os
import sys
import warnings
from collections.abc import Callable, Iterator, Sequence
from contextlib import contextmanager
from dataclasses import dataclass
from functools import partial
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 (
    Awaitable,
    AwaitableConstraints,
    Effect,
    GetParseError,
    MultiGet,
)
from pants.util.docutil import doc_url
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


def _returns_awaitable(func: Any) -> bool:
    if not callable(func):
        return False
    ret = _lookup_return_type(func)
    if not isinstance(ret, tuple):
        ret = (ret,)
    return any(issubclass(r, Awaitable) for r in ret if isinstance(r, type))


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_legacy_awaitable(self, call_node: ast.Call, is_effect: bool) -> AwaitableConstraints:
        get_args = call_node.args
        parse_error = partial(GetParseError, get_args=get_args, source_file_name=self.source_file)

        if len(get_args) not in (1, 2, 3):
            # TODO: fix parse error message formatting... (TODO: create ticket)
            raise parse_error(
                self._format(
                    call_node,
                    f"Expected one to three arguments, but got {len(get_args)} arguments.",
                )
            )

        output_node = get_args[0]
        output_type = self._lookup(output_node)

        input_nodes, input_types = self._get_inputs(get_args[1:])

        return AwaitableConstraints(
            None,
            self._check_constraint_arg_type(output_type, output_node),
            0,
            tuple(
                self._check_constraint_arg_type(input_type, input_node)
                for input_type, input_node in zip(input_types, input_nodes)
            ),
            is_effect,
        )

    def _get_byname_awaitable(
        self, rule_id: str, rule_func: Callable | RuleDescriptor, call_node: ast.Call
    ) -> AwaitableConstraints:
        parse_error = partial(
            GetParseError, get_args=call_node.args, source_file_name=self.source_file
        )

        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:
            raise parse_error(
                self._format(
                    call_node,
                    "Expected an `**implicitly(..)` application as the only keyword input.",
                )
            )

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

pantsbuild / pants / 19529437518

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