File size: 8,901 Bytes
036a2db
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
"""
symbols.py — Attribute ownership extraction for resolving self.attr.method() chains.

Given a class with `self.router = APIRouter()`, this module figures out that
`self.router` has type `APIRouter`, enabling cross-file method resolution.
"""

import ast
from typing import Dict, Optional, Tuple, List


def _iter_statements(body):
    """Yield statements in source order, recursing into if/for/while/with/try."""
    for stmt in body:
        yield stmt

        if isinstance(stmt, (ast.If, ast.For, ast.AsyncFor, ast.While)):
            yield from _iter_statements(stmt.body)
            yield from _iter_statements(stmt.orelse)

        elif isinstance(stmt, (ast.With, ast.AsyncWith)):
            yield from _iter_statements(stmt.body)

        elif isinstance(stmt, ast.Try):
            yield from _iter_statements(stmt.body)
            for handler in stmt.handlers:
                yield from _iter_statements(handler.body)
            yield from _iter_statements(stmt.orelse)
            yield from _iter_statements(stmt.finalbody)


def _extract_annotation_name(annotation) -> Optional[Tuple[Optional[str], str]]:
    """
    Returns (qualifier, bare_name) or None.

    Router          ->  (None, "Router")
    Optional[Router]->  (None, "Router")       <- unwrap subscript
    routing.Router  ->  ("routing", "Router")
    """
    if isinstance(annotation, ast.Name):
        return (None, annotation.id)

    if isinstance(annotation, ast.Attribute) and isinstance(annotation.value, ast.Name):
        return (annotation.value.id, annotation.attr)

    # Optional[Router], List[Router], etc.  — unwrap one level
    if isinstance(annotation, ast.Subscript):
        return _extract_annotation_name(annotation.slice)

    return None


def _extract_call_owner(value) -> Optional[Tuple[Optional[str], str]]:
    """
    Returns (qualifier, bare_name) or None.

    APIRouter()          ->  (None,      "APIRouter")
    routing.APIRouter()  ->  ("routing", "APIRouter")
    make_helper()        ->  (None,      "make_helper")
    """
    if not isinstance(value, ast.Call):
        return None

    func = value.func

    if isinstance(func, ast.Name):
        return (None, func.id)

    if isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name):
        return (func.value.id, func.attr)

    if isinstance(func, ast.Attribute):
        return (None, func.attr)

    return None


def _extract_assign_owner(value, local_var_types=None, param_types=None):
    """
    Handles non-Call RHS cases that still indicate a type:
      self.x = router or Router()   -> BoolOp -> try each operand
      self.x = router               -> bare Name -> no type info (return None)
    """
    if local_var_types is None:
        local_var_types = {}
    if param_types is None:
        param_types = {}

    if isinstance(value, ast.Call):
        return _extract_call_owner(value)

    if isinstance(value, ast.Name):
        if value.id in param_types:
            return param_types[value.id]
        if value.id in local_var_types:
            return local_var_types[value.id]
        return None

    if isinstance(value, ast.BoolOp):
        call_result = None
        name_result = None
        for operand in value.values:
            ref = _extract_assign_owner(operand, local_var_types, param_types)
            if ref is None:
                continue
            if isinstance(operand, ast.Call):
                call_result = call_result or ref
            else:
                name_result = name_result or ref
        return call_result or name_result

    return None


def extract_local_var_types(
    fn_node,
    param_types: Optional[Dict[str, Tuple[Optional[str], str]]] = None,
) -> Dict[str, Tuple[Optional[str], str]]:
    """
    Track local variable -> type assignments within a SINGLE function body,
    free function or method alike:

        h = Helper()              -> {"h": (None, "Helper")}
        r: routing.Router = ...   -> {"r": ("routing", "Router")}
        x = make_helper()         -> chases factory functions the same way
                                      attribute ownership tracking does
                                      (resolution of the factory return type
                                      itself happens later, in graph_builder)

    This is the free-function counterpart to the self.attr tracking that
    extract_attribute_ownerships does for class bodies. It exists because
    a huge fraction of real code instantiates a class in a local variable
    inside a plain function (not as a self.attr) and immediately calls a
    method on it -- e.g.:

        def run():
            h = Handler()
            return h.process()

    Without this, `h.process()` can never resolve: there's nowhere that
    records "h" has type "Handler". extract_attribute_ownerships alone
    can't help, because it only ever looks inside ast.ClassDef bodies.

    Returns: {local_var_name: (qualifier, bare_type_name)}
    """
    if param_types is None:
        param_types = {}

    local_var_types: Dict[str, Tuple[Optional[str], str]] = {}

    for stmt in _iter_statements(fn_node.body):

        if isinstance(stmt, ast.AnnAssign):
            target = stmt.target
            if isinstance(target, ast.Name):
                ref = _extract_annotation_name(stmt.annotation)
                if ref:
                    local_var_types[target.id] = ref
            continue

        if isinstance(stmt, ast.Assign):
            for target in stmt.targets:
                if isinstance(target, ast.Name):
                    ref = _extract_assign_owner(
                        stmt.value, local_var_types, param_types
                    )
                    if ref:
                        local_var_types[target.id] = ref
            continue

    return local_var_types


def extract_param_types(fn_node) -> Dict[str, Tuple[Optional[str], str]]:
    """
    Map annotated parameter names to their (qualifier, bare_name) type, e.g.

        def f(router: routing.Router, name: str): ...

    -> {"router": ("routing", "Router")}   (unannotated/non-type params skipped)
    """
    param_types: Dict[str, Tuple[Optional[str], str]] = {}
    for arg in fn_node.args.args:
        if arg.annotation is not None:
            ref = _extract_annotation_name(arg.annotation)
            if ref:
                param_types[arg.arg] = ref
    return param_types


def extract_attribute_ownerships(tree) -> Dict[str, Tuple[Optional[str], str]]:
    """
    Extracts mappings like:
        self.router: routing.Router = routing.Router()
        self.router = Router()
        self.router2 = router            # typed constructor param
        h = Helper(); self.helper = h    # local var tracking
        self.router3 = router or Router()
        self.opt_router: Optional[Router] = None

    Returns: {"FastAPI.router": (qualifier, type_name), ...}
    """
    ownerships: Dict[str, Tuple[Optional[str], str]] = {}

    for node in tree.body:
        if not isinstance(node, ast.ClassDef):
            continue

        class_name = node.name

        for item in node.body:
            if not isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue

            param_types = extract_param_types(item)

            local_var_types = {}

            for stmt in _iter_statements(item.body):

                if isinstance(stmt, ast.AnnAssign):
                    target = stmt.target
                    if (
                        isinstance(target, ast.Attribute)
                        and isinstance(target.value, ast.Name)
                        and target.value.id == "self"
                    ):
                        ref = _extract_annotation_name(stmt.annotation)
                        if ref:
                            ownerships[f"{class_name}.{target.attr}"] = ref
                    continue

                if isinstance(stmt, ast.Assign):
                    for target in stmt.targets:
                        if (
                            isinstance(target, ast.Attribute)
                            and isinstance(target.value, ast.Name)
                            and target.value.id == "self"
                        ):
                            ref = _extract_assign_owner(
                                stmt.value, local_var_types, param_types
                            )
                            if ref:
                                ownerships[f"{class_name}.{target.attr}"] = ref

                        elif isinstance(target, ast.Name):
                            ref = _extract_assign_owner(
                                stmt.value, local_var_types, param_types
                            )
                            if ref:
                                local_var_types[target.id] = ref
                    continue

    return ownerships