-
Notifications
You must be signed in to change notification settings - Fork 56
/
analyzer.py
1764 lines (1509 loc) · 77.2 KB
/
analyzer.py
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
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""The AST visitor."""
import ast
import logging
import symtable
from typing import Union
from .anutils import (
ExecuteInInnerScope,
Scope,
UnresolvedSuperCallError,
format_alias,
get_ast_node_name,
get_module_name,
resolve_method_resolution_order,
sanitize_exprs,
tail,
)
from .node import Flavor, Node
# TODO: add Cython support (strip type annotations in a preprocess step, then treat as Python)
# TODO: built-in functions (range(), enumerate(), zip(), iter(), ...):
# add to a special scope "built-in" in analyze_scopes() (or ignore altogether)
# TODO: support Node-ifying ListComp et al, List, Tuple
# TODO: make the analyzer smarter (see individual TODOs below)
# Note the use of the term "node" for two different concepts:
#
# - AST nodes (the "node" argument of CallGraphVisitor.visit_*())
#
# - The Node class that mainly stores auxiliary information about AST nodes,
# for the purposes of generating the call graph.
#
# Namespaces also get a Node (with no associated AST node).
# These tables were useful for porting the visitor to Python 3:
#
# https://docs.python.org/2/library/compiler.html#module-compiler.ast
# https://docs.python.org/3/library/ast.html#abstract-grammar
#
class CallGraphVisitor(ast.NodeVisitor):
"""A visitor that can be walked over a Python AST, and will derive
information about the objects in the AST and how they use each other.
A single CallGraphVisitor object can be run over several ASTs (from a
set of source files). The resulting information is the aggregate from
all files. This way use information between objects in different files
can be gathered."""
def __init__(self, filenames, root: str = None, logger=None):
self.logger = logger or logging.getLogger(__name__)
# full module names for all given files
self.module_to_filename = {} # inverse mapping for recording which file each AST node came from
for filename in filenames:
mod_name = get_module_name(filename)
self.module_to_filename[mod_name] = filename
self.filenames = filenames
self.root = root
# data gathered from analysis
self.defines_edges = {}
self.uses_edges = {}
self.nodes = {} # Node name: list of Node objects (in possibly different namespaces)
self.scopes = {} # fully qualified name of namespace: Scope object
self.class_base_ast_nodes = {} # pass 1: class Node: list of AST nodes
self.class_base_nodes = {} # pass 2: class Node: list of Node objects (local bases, no recursion)
self.mro = {} # pass 2: class Node: list of Node objects in Python's MRO order
# current context for analysis
self.module_name = None
self.filename = None
self.name_stack = [] # for building namespace name, node naming
self.scope_stack = [] # the Scope objects currently in scope
self.class_stack = [] # Nodes for class definitions currently in scope
self.context_stack = [] # for detecting which FunctionDefs are methods
self.last_value = None
# Analyze.
self.process()
def process(self):
"""Analyze the set of files, twice so that any forward-references are picked up."""
for pas in range(2):
for filename in self.filenames:
self.logger.info("========== pass %d, file '%s' ==========" % (pas + 1, filename))
self.process_one(filename)
if pas == 0:
self.resolve_base_classes() # must be done only after all files seen
self.postprocess()
def process_one(self, filename):
"""Analyze the specified Python source file."""
if filename not in self.filenames:
raise ValueError(
"Filename '%s' has not been preprocessed (was not given to __init__, which got %s)"
% (filename, self.filenames)
)
with open(filename, "rt", encoding="utf-8") as f:
content = f.read()
self.filename = filename
self.module_name = get_module_name(filename, root=self.root)
self.analyze_scopes(content, filename) # add to the currently known scopes
self.visit(ast.parse(content, filename))
self.module_name = None
self.filename = None
def resolve_base_classes(self):
"""Resolve base classes from AST nodes to Nodes.
Run this between pass 1 and pass 2 to pick up inherited methods.
Currently, this can parse ast.Names and ast.Attributes as bases.
"""
self.logger.debug("Resolving base classes")
assert len(self.scope_stack) == 0 # only allowed between passes
for node in self.class_base_ast_nodes: # Node: list of AST nodes
self.class_base_nodes[node] = []
for ast_node in self.class_base_ast_nodes[node]:
# perform the lookup in the scope enclosing the class definition
self.scope_stack.append(self.scopes[node.namespace])
if isinstance(ast_node, ast.Name):
baseclass_node = self.get_value(ast_node.id)
elif isinstance(ast_node, ast.Attribute):
_, baseclass_node = self.get_attribute(ast_node) # don't care about obj, just grab attr
else: # give up
baseclass_node = None
self.scope_stack.pop()
if isinstance(baseclass_node, Node) and baseclass_node.namespace is not None:
self.class_base_nodes[node].append(baseclass_node)
self.logger.debug("All base classes (non-recursive, local level only): %s" % self.class_base_nodes)
self.logger.debug("Resolving method resolution order (MRO) for all analyzed classes")
self.mro = resolve_method_resolution_order(self.class_base_nodes, self.logger)
self.logger.debug("Method resolution order (MRO) for all analyzed classes: %s" % self.mro)
def postprocess(self):
"""Finalize the analysis."""
# Compared to the original Pyan, the ordering of expand_unknowns() and
# contract_nonexistents() has been switched.
#
# It seems the original idea was to first convert any unresolved, but
# specific, references to the form *.name, and then expand those to see
# if they match anything else. However, this approach has the potential
# to produce a lot of spurious uses edges (for unrelated functions with
# a name that happens to match).
#
# Now that the analyzer is (very slightly) smarter about resolving
# attributes and imports, we do it the other way around: we only expand
# those references that could not be resolved to any known name, and
# then remove any references pointing outside the analyzed file set.
self.expand_unknowns()
self.resolve_imports()
self.contract_nonexistents()
self.cull_inherited()
self.collapse_inner()
###########################################################################
# visitor methods
# In visit_*(), the "node" argument refers to an AST node.
# Python docs:
# https://docs.python.org/3/library/ast.html#abstract-grammar
def resolve_imports(self):
"""
resolve relative imports and remap nodes
"""
# first find all imports and map to themselves. we will then remap those that are currently pointing
# to duplicates or into the void
imports_to_resolve = {n for items in self.nodes.values() for n in items if n.flavor == Flavor.IMPORTEDITEM}
# map real definitions
import_mapping = {}
while len(imports_to_resolve) > 0:
from_node = imports_to_resolve.pop()
if from_node in import_mapping:
continue
to_uses = self.uses_edges.get(from_node, set([from_node]))
assert len(to_uses) == 1
to_node = to_uses.pop() # resolve alias
# resolve namespace and get module
if to_node.namespace == "":
module_node = to_node
else:
assert from_node.name == to_node.name
module_node = self.get_node("", to_node.namespace)
module_uses = self.uses_edges.get(module_node)
if module_uses is not None:
# check if in module item exists and if yes, map to it
for candidate_to_node in module_uses:
if candidate_to_node.name == from_node.name:
to_node = candidate_to_node
import_mapping[from_node] = to_node
if to_node.flavor == Flavor.IMPORTEDITEM and from_node is not to_node: # avoid self-recursion
imports_to_resolve.add(to_node)
break
# set previously undefined nodes to defined
# go through undefined attributes
attribute_import_mapping = {}
for nodes in self.nodes.values():
for node in nodes:
if not node.defined and node.flavor == Flavor.ATTRIBUTE:
# try to resolve namespace and find imported item mapping
for from_node, to_node in import_mapping.items():
if (
f"{from_node.namespace}.{from_node.name}" == node.namespace
and from_node.flavor == Flavor.IMPORTEDITEM
):
# use define edges as potential candidates
for candidate_to_node in self.defines_edges[to_node]: #
if candidate_to_node.name == node.name:
attribute_import_mapping[node] = candidate_to_node
break
import_mapping.update(attribute_import_mapping)
# remap nodes based on import mapping
self.nodes = {name: [import_mapping.get(n, n) for n in items] for name, items in self.nodes.items()}
self.uses_edges = {
import_mapping.get(from_node, from_node): {import_mapping.get(to_node, to_node) for to_node in to_nodes}
for from_node, to_nodes in self.uses_edges.items()
if len(to_nodes) > 0
}
self.defines_edges = {
import_mapping.get(from_node, from_node): {import_mapping.get(to_node, to_node) for to_node in to_nodes}
for from_node, to_nodes in self.defines_edges.items()
if len(to_nodes) > 0
}
def filter(self, node: Union[None, Node] = None, namespace: Union[str, None] = None, max_iter: int = 1000):
"""
filter callgraph nodes that related to `node` or are in `namespace`
Args:
node: pyan node for which related nodes should be found, if none, filter only for namespace
namespace: namespace to search in (name of top level module),
if None, determines namespace from `node`
max_iter: maximum number of iterations and nodes to iterate
Returns:
self
"""
# filter the nodes to avoid cluttering the callgraph with irrelevant information
filtered_nodes = self.get_related_nodes(node, namespace=namespace, max_iter=max_iter)
self.nodes = {name: [node for node in nodes if node in filtered_nodes] for name, nodes in self.nodes.items()}
self.uses_edges = {
node: {n for n in nodes if n in filtered_nodes}
for node, nodes in self.uses_edges.items()
if node in filtered_nodes
}
self.defines_edges = {
node: {n for n in nodes if n in filtered_nodes}
for node, nodes in self.defines_edges.items()
if node in filtered_nodes
}
return self
def get_related_nodes(
self, node: Union[None, Node] = None, namespace: Union[str, None] = None, max_iter: int = 1000
) -> set:
"""
get nodes that related to `node` or are in `namespace`
Args:
node: pyan node for which related nodes should be found, if none, filter only for namespace
namespace: namespace to search in (name of top level module),
if None, determines namespace from `node`
max_iter: maximum number of iterations and nodes to iterate
Returns:
set: set of nodes related to `node` including `node` itself
"""
# check if searching through all nodes is necessary
if node is None:
queue = []
if namespace is None:
new_nodes = {n for items in self.nodes.values() for n in items}
else:
new_nodes = {
n
for items in self.nodes.values()
for n in items
if n.namespace is not None and namespace in n.namespace
}
else:
new_nodes = set()
if namespace is None:
namespace = node.namespace.strip(".").split(".", 1)[0]
queue = [node]
# use queue system to search through nodes
# essentially add a node to the queue and then search all connected nodes which are in turn added to the queue
# until the queue itself is empty or the maximum limit of max_iter searches have been hit
i = max_iter
while len(queue) > 0:
item = queue.pop()
if item not in new_nodes:
new_nodes.add(item)
i -= 1
if i < 0:
break
queue.extend(
[
n
for n in self.uses_edges.get(item, [])
if n in self.uses_edges and n not in new_nodes and namespace in n.namespace
]
)
queue.extend(
[
n
for n in self.defines_edges.get(item, [])
if n in self.defines_edges and n not in new_nodes and namespace in n.namespace
]
)
return new_nodes
def visit_Module(self, node):
self.logger.debug("Module %s, %s" % (self.module_name, self.filename))
# Modules live in the top-level namespace, ''.
module_node = self.get_node("", self.module_name, node, flavor=Flavor.MODULE)
self.associate_node(module_node, node, filename=self.filename)
ns = self.module_name
self.name_stack.append(ns)
self.scope_stack.append(self.scopes[ns])
self.context_stack.append("Module %s" % (ns))
self.generic_visit(node) # visit the **children** of node
self.context_stack.pop()
self.scope_stack.pop()
self.name_stack.pop()
self.last_value = None
if self.add_defines_edge(module_node, None):
self.logger.info("Def Module %s" % node)
def visit_ClassDef(self, node):
self.logger.debug("ClassDef %s, %s:%s" % (node.name, self.filename, node.lineno))
from_node = self.get_node_of_current_namespace()
ns = from_node.get_name()
to_node = self.get_node(ns, node.name, node, flavor=Flavor.CLASS)
if self.add_defines_edge(from_node, to_node):
self.logger.info("Def from %s to Class %s" % (from_node, to_node))
# The graph Node may have been created earlier by a FromImport,
# in which case its AST node points to the site of the import.
#
# Change the AST node association of the relevant graph Node
# to this AST node (the definition site) to get the correct
# source line number information in annotated output.
#
self.associate_node(to_node, node, self.filename)
# Bind the name specified by the AST node to the graph Node
# in the current scope.
#
self.set_value(node.name, to_node)
self.class_stack.append(to_node)
self.name_stack.append(node.name)
inner_ns = self.get_node_of_current_namespace().get_name()
self.scope_stack.append(self.scopes[inner_ns])
self.context_stack.append("ClassDef %s" % (node.name))
self.class_base_ast_nodes[to_node] = []
for b in node.bases:
# gather info for resolution of inherited attributes in pass 2 (see get_attribute())
self.class_base_ast_nodes[to_node].append(b)
# mark uses from a derived class to its bases (via names appearing in a load context).
self.visit(b)
for stmt in node.body:
self.visit(stmt)
self.context_stack.pop()
self.scope_stack.pop()
self.name_stack.pop()
self.class_stack.pop()
def visit_FunctionDef(self, node):
self.logger.debug("FunctionDef %s, %s:%s" % (node.name, self.filename, node.lineno))
# To begin with:
#
# - Analyze decorators. They belong to the surrounding scope,
# so we must analyze them before entering the function scope.
#
# - Determine whether this definition is for a function, an (instance)
# method, a static method or a class method.
#
# - Grab the name representing "self", if this is either an instance
# method or a class method. (For a class method, it represents cls,
# but Pyan only cares about types, not instances.)
#
self_name, flavor = self.analyze_functiondef(node)
# Now we can create the Node.
#
from_node = self.get_node_of_current_namespace()
ns = from_node.get_name()
to_node = self.get_node(ns, node.name, node, flavor=flavor)
if self.add_defines_edge(from_node, to_node):
self.logger.info("Def from %s to Function %s" % (from_node, to_node))
# Same remarks as for ClassDef above.
#
self.associate_node(to_node, node, self.filename)
self.set_value(node.name, to_node)
# Enter the function scope
#
self.name_stack.append(node.name)
inner_ns = self.get_node_of_current_namespace().get_name()
self.scope_stack.append(self.scopes[inner_ns])
self.context_stack.append("FunctionDef %s" % (node.name))
# Capture which names correspond to function args.
#
self.generate_args_nodes(node.args, inner_ns)
# self_name is just an ordinary name in the method namespace, except
# that its value is implicitly set by Python when the method is called.
#
# Bind self_name in the function namespace to its initial value,
# i.e. the current class. (Class, because Pyan cares only about
# object types, not instances.)
#
# After this point, self_name behaves like any other name.
#
if self_name is not None:
class_node = self.get_current_class()
self.scopes[inner_ns].defs[self_name] = class_node
self.logger.info('Method def: setting self name "%s" to %s' % (self_name, class_node))
# record bindings of args to the given default values, if present
self.analyze_arguments(node.args)
# Analyze the function body
#
for stmt in node.body:
self.visit(stmt)
# Exit the function scope
#
self.context_stack.pop()
self.scope_stack.pop()
self.name_stack.pop()
def visit_AsyncFunctionDef(self, node):
self.visit_FunctionDef(node) # TODO: alias for now; tag async functions in output in a future version?
def visit_Lambda(self, node):
# TODO: avoid lumping together all lambdas in the same namespace.
self.logger.debug("Lambda, %s:%s" % (self.filename, node.lineno))
with ExecuteInInnerScope(self, "lambda"):
inner_ns = self.get_node_of_current_namespace().get_name()
self.generate_args_nodes(node.args, inner_ns)
self.analyze_arguments(node.args)
self.visit(node.body) # single expr
def generate_args_nodes(self, ast_args, inner_ns):
"""Capture which names correspond to function args.
In the function scope, set them to a nonsense Node,
to prevent leakage of identifiers of matching name
from the enclosing scope (due to the local value being None
until we set it to this nonsense Node).
ast_args: node.args from a FunctionDef or Lambda
inner_ns: namespace of the function or lambda, for scope lookup
"""
sc = self.scopes[inner_ns]
# As the name of the nonsense node, we can use any string that
# is not a valid Python identifier.
#
# It has no sensible flavor, so we leave its flavor unspecified.
nonsense_node = self.get_node(inner_ns, "^^^argument^^^", None)
# args, vararg (*args), kwonlyargs, kwarg (**kwargs)
for a in ast_args.args: # positional
sc.defs[a.arg] = nonsense_node
if ast_args.vararg is not None: # *args if present
sc.defs[ast_args.vararg] = nonsense_node
for a in ast_args.kwonlyargs: # any after *args or *
sc.defs[a.arg] = nonsense_node
if ast_args.kwarg is not None: # **kwargs if present
sc.defs[ast_args.kwarg] = nonsense_node
def analyze_arguments(self, ast_args):
"""Analyze an arguments node of the AST.
Record bindings of args to the given default values, if present.
Used for analyzing FunctionDefs and Lambdas."""
# https://greentreesnakes.readthedocs.io/en/latest/nodes.html?highlight=functiondef#arguments
if ast_args.defaults:
n = len(ast_args.defaults)
for tgt, val in zip(ast_args.args[-n:], ast_args.defaults):
targets = sanitize_exprs(tgt)
values = sanitize_exprs(val)
self.analyze_binding(targets, values)
if ast_args.kw_defaults:
n = len(ast_args.kw_defaults)
for tgt, val in zip(ast_args.kwonlyargs, ast_args.kw_defaults):
if val is not None:
targets = sanitize_exprs(tgt)
values = sanitize_exprs(val)
self.analyze_binding(targets, values)
def visit_Import(self, node):
self.logger.debug("Import %s, %s:%s" % ([format_alias(x) for x in node.names], self.filename, node.lineno))
# TODO: add support for relative imports (path may be like "....something.something")
# https://www.python.org/dev/peps/pep-0328/#id10
for import_item in node.names: # the names are modules
self.analyze_module_import(import_item, node)
def visit_ImportFrom(self, node):
self.logger.debug(
"ImportFrom: from %s import %s, %s:%s"
% (node.module, [format_alias(x) for x in node.names], self.filename, node.lineno)
)
# Pyan needs to know the package structure, and how the program
# being analyzed is actually going to be invoked (!), to be able to
# resolve relative imports correctly.
#
# As a solution, we register imports here and later, when all files have been parsed, resolve them.
from_node = self.get_node_of_current_namespace()
if node.module is None: # resolve relative imports 'None' such as "from . import foo"
self.logger.debug(
"ImportFrom (original) from %s import %s, %s:%s"
% ("." * node.level, [format_alias(x) for x in node.names], self.filename, node.lineno)
)
tgt_level = node.level
current_module_namespace = self.module_name.rsplit(".", tgt_level)[0]
tgt_name = current_module_namespace
self.logger.debug(
"ImportFrom (resolved): from %s import %s, %s:%s"
% (tgt_name, [format_alias(x) for x in node.names], self.filename, node.lineno)
)
elif node.level != 0: # resolve from ..module import foo
self.logger.debug(
"ImportFrom (original): from %s import %s, %s:%s"
% (node.module, [format_alias(x) for x in node.names], self.filename, node.lineno)
)
tgt_level = node.level
current_module_namespace = self.module_name.rsplit(".", tgt_level)[0]
tgt_name = current_module_namespace + "." + node.module
self.logger.debug(
"ImportFrom (resolved): from %s import %s, %s:%s"
% (tgt_name, [format_alias(x) for x in node.names], self.filename, node.lineno)
)
else:
tgt_name = node.module # normal from module.submodule import foo
# link each import separately
for alias in node.names:
# check if import is module
if tgt_name + "." + alias.name in self.module_to_filename:
to_node = self.get_node("", tgt_name + "." + alias.name, node, flavor=Flavor.MODULE)
else:
to_node = self.get_node(tgt_name, alias.name, node, flavor=Flavor.IMPORTEDITEM)
# if there is alias, add extra edge between alias and node
if alias.asname is not None:
alias_name = alias.asname
else:
alias_name = alias.name
self.set_value(alias_name, to_node) # set node to be discoverable in module
self.logger.info("From setting name %s to %s" % (alias_name, to_node))
self.logger.debug("Use from %s to ImportFrom %s" % (from_node, to_node))
if self.add_uses_edge(from_node, to_node):
self.logger.info("New edge added for Use from %s to ImportFrom %s" % (from_node, to_node))
def analyze_module_import(self, import_item, ast_node):
"""Analyze a names AST node inside an Import or ImportFrom AST node.
This handles the case where the objects being imported are modules.
import_item: an item of ast_node.names
ast_node: for recording source location information
"""
src_name = import_item.name # what is being imported
# mark the use site
#
# where it is being imported to, i.e. the **user**
from_node = self.get_node_of_current_namespace()
# the thing **being used** (under the asname, if any)
mod_node = self.get_node("", src_name, ast_node, flavor=Flavor.MODULE)
# if there is alias, add extra edge between alias and node
if import_item.asname is not None:
alias_name = import_item.asname
else:
alias_name = mod_node.name
self.add_uses_edge(from_node, mod_node)
self.logger.info("New edge added for Use import %s in %s" % (mod_node, from_node))
self.set_value(alias_name, mod_node) # set node to be discoverable in module
self.logger.info("From setting name %s to %s" % (alias_name, mod_node))
# Edmund Horner's original post has info on what this fixed in Python 2.
# https://ejrh.wordpress.com/2012/01/31/call-graphs-in-python-part-2/
#
# Essentially, this should make '.'.join(...) see str.join.
# Pyan3 currently handles that in resolve_attribute() and get_attribute().
#
# Python 3.4 does not have ast.Constant, but 3.6 does.
# TODO: actually test this with Python 3.6 or later.
#
def visit_Constant(self, node):
self.logger.debug("Constant %s, %s:%s" % (node.value, self.filename, node.lineno))
t = type(node.value)
ns = self.get_node_of_current_namespace().get_name()
tn = t.__name__
self.last_value = self.get_node(ns, tn, node, flavor=Flavor.ATTRIBUTE)
# attribute access (node.ctx determines whether set (ast.Store) or get (ast.Load))
def visit_Attribute(self, node):
objname = get_ast_node_name(node.value)
self.logger.debug(
"Attribute %s of %s in context %s, %s:%s" % (node.attr, objname, type(node.ctx), self.filename, node.lineno)
)
# TODO: self.last_value is a hack. Handle names in store context (LHS)
# in analyze_binding(), so that visit_Attribute() only needs to handle
# the load context (i.e. detect uses of the name).
#
if isinstance(node.ctx, ast.Store):
new_value = self.last_value
try:
if self.set_attribute(node, new_value):
self.logger.info("setattr %s on %s to %s" % (node.attr, objname, new_value))
except UnresolvedSuperCallError:
# Trying to set something belonging to an unresolved super()
# of something; just ignore this attempt to setattr.
return
elif isinstance(node.ctx, ast.Load):
try:
obj_node, attr_node = self.get_attribute(node)
except UnresolvedSuperCallError:
# Avoid adding a wildcard if the lookup failed due to an
# unresolved super() in the attribute chain.
return
# Both object and attr known.
if isinstance(attr_node, Node):
self.logger.info("getattr %s on %s returns %s" % (node.attr, objname, attr_node))
# add uses edge
from_node = self.get_node_of_current_namespace()
self.logger.debug("Use from %s to %s" % (from_node, attr_node))
if self.add_uses_edge(from_node, attr_node):
self.logger.info("New edge added for Use from %s to %s" % (from_node, attr_node))
# remove resolved wildcard from current site to <Node *.attr>
if attr_node.namespace is not None:
self.remove_wild(from_node, attr_node, node.attr)
self.last_value = attr_node
# Object known, but attr unknown. Create node and add a uses edge.
#
# TODO: this is mainly useful for imports. Should probably disallow
# creating new attribute nodes for other undefined attrs of known objs.
#
# E.g.
#
# import math # create <Node math>
# math.sin # create <Node math.sin> (instead of <Node *.sin> even though math.py is not analyzed)
#
# This sometimes creates silly nodes such as (when analyzing Pyan itself)
# <Node pyan.analyzer.CallGraphVisitor.defines_edges.name.namespace>
# but these are harmless, as they are considered undefined and
# will not be visualized.
#
elif isinstance(obj_node, Node) and obj_node.namespace is not None:
tgt_name = node.attr
from_node = self.get_node_of_current_namespace()
ns = obj_node.get_name() # fully qualified namespace **of attr**
to_node = self.get_node(ns, tgt_name, node, flavor=Flavor.ATTRIBUTE)
self.logger.debug(
f"Use from {from_node} to {to_node} (target obj {obj_node} known but target attr "
f"{node.attr} not resolved; maybe fwd ref or unanalyzed import)"
)
if self.add_uses_edge(from_node, to_node):
self.logger.info(
"New edge added for Use from {from_node} to {to_node} (target obj {obj_node} known but "
f"target attr {node.attr} not resolved; maybe fwd ref or unanalyzed import)"
)
# remove resolved wildcard from current site to <Node *.attr>
self.remove_wild(from_node, obj_node, node.attr)
self.last_value = to_node
# pass on
else:
self.visit(node.value)
# name access (node.ctx determines whether set (ast.Store) or get (ast.Load))
def visit_Name(self, node):
self.logger.debug("Name %s in context %s, %s:%s" % (node.id, type(node.ctx), self.filename, node.lineno))
# TODO: self.last_value is a hack. Handle names in store context (LHS)
# in analyze_binding(), so that visit_Name() only needs to handle
# the load context (i.e. detect uses of the name).
if isinstance(node.ctx, ast.Store):
# when we get here, self.last_value has been set by visit_Assign()
self.set_value(node.id, self.last_value)
# A name in a load context is a use of the object the name points to.
elif isinstance(node.ctx, ast.Load):
tgt_name = node.id
to_node = self.get_value(tgt_name) # resolves "self" if needed
current_class = self.get_current_class()
if current_class is None or to_node is not current_class: # add uses edge only if not pointing to "self"
# TODO if the name is a local variable (i.e. in the innermost scope), and
# has no known value, then don't try to create a Node for it.
if not isinstance(to_node, Node):
# namespace=None means we don't know the namespace yet
to_node = self.get_node(None, tgt_name, node, flavor=Flavor.UNKNOWN)
from_node = self.get_node_of_current_namespace()
self.logger.debug("Use from %s to Name %s" % (from_node, to_node))
if self.add_uses_edge(from_node, to_node):
self.logger.info("New edge added for Use from %s to Name %s" % (from_node, to_node))
self.last_value = to_node
def visit_Assign(self, node):
# - chaining assignments like "a = b = c" produces multiple targets
# - tuple unpacking works as a separate mechanism on top of that (see analyze_binding())
#
if len(node.targets) > 1:
self.logger.debug("Assign (chained with %d outputs)" % (len(node.targets)))
# TODO: support lists, dicts, sets (so that we can recognize calls to their methods)
# TODO: begin with supporting empty lists, dicts, sets
# TODO: need to be more careful in sanitizing; currently destroys a bare list
values = sanitize_exprs(node.value) # values is the same for each set of targets
for targets in node.targets:
targets = sanitize_exprs(targets)
self.logger.debug(
"Assign %s %s, %s:%s"
% (
[get_ast_node_name(x) for x in targets],
[get_ast_node_name(x) for x in values],
self.filename,
node.lineno,
)
)
self.analyze_binding(targets, values)
def visit_AnnAssign(self, node): # PEP 526, Python 3.6+
target = sanitize_exprs(node.target)
self.last_value = None
if node.value is not None:
value = sanitize_exprs(node.value)
# issue #62: value may be an empty list, so it doesn't always have any elements
# even after `sanitize_exprs`.
self.logger.debug(
"AnnAssign %s %s, %s:%s"
% (get_ast_node_name(target[0]), get_ast_node_name(value), self.filename, node.lineno)
)
self.analyze_binding(target, value)
else: # just a type declaration
self.logger.debug(
"AnnAssign %s <no value>, %s:%s" % (get_ast_node_name(target[0]), self.filename, node.lineno)
)
self.last_value = None
self.visit(target[0])
# TODO: use the type annotation from node.annotation?
# http://greentreesnakes.readthedocs.io/en/latest/nodes.html#AnnAssign
def visit_AugAssign(self, node):
targets = sanitize_exprs(node.target)
values = sanitize_exprs(node.value) # values is the same for each set of targets
self.logger.debug(
"AugAssign %s %s %s, %s:%s"
% (
[get_ast_node_name(x) for x in targets],
type(node.op),
[get_ast_node_name(x) for x in values],
self.filename,
node.lineno,
)
)
# TODO: maybe no need to handle tuple unpacking in AugAssign? (but simpler to use the same implementation)
self.analyze_binding(targets, values)
# for() is also a binding form.
#
# (Without analyzing the bindings, we would get an unknown node for any
# use of the loop counter(s) in the loop body. This would have confusing
# consequences in the expand_unknowns() step, if the same name is
# in use elsewhere.)
#
def visit_For(self, node):
self.logger.debug("For-loop, %s:%s" % (self.filename, node.lineno))
targets = sanitize_exprs(node.target)
values = sanitize_exprs(node.iter)
self.analyze_binding(targets, values)
for stmt in node.body:
self.visit(stmt)
for stmt in node.orelse:
self.visit(stmt)
def visit_AsyncFor(self, node):
self.visit_For(node) # TODO: alias for now; tag async for in output in a future version?
def visit_ListComp(self, node):
self.logger.debug("ListComp, %s:%s" % (self.filename, node.lineno))
self.analyze_comprehension(node, "listcomp")
def visit_SetComp(self, node):
self.logger.debug("SetComp, %s:%s" % (self.filename, node.lineno))
self.analyze_comprehension(node, "setcomp")
def visit_DictComp(self, node):
self.logger.debug("DictComp, %s:%s" % (self.filename, node.lineno))
self.analyze_comprehension(node, "dictcomp", field1="key", field2="value")
def visit_GeneratorExp(self, node):
self.logger.debug("GeneratorExp, %s:%s" % (self.filename, node.lineno))
self.analyze_comprehension(node, "genexpr")
def analyze_comprehension(self, node, label, field1="elt", field2=None):
# The outermost iterator is evaluated in the current scope;
# everything else in the new inner scope.
#
# See function symtable_handle_comprehension() in
# https://github.com/python/cpython/blob/master/Python/symtable.c
# For how it works, see
# https://stackoverflow.com/questions/48753060/what-are-these-extra-symbols-in-a-comprehensions-symtable
# For related discussion, see
# https://bugs.python.org/issue10544
gens = node.generators # tuple of ast.comprehension
outermost = gens[0]
moregens = gens[1:] if len(gens) > 1 else []
outermost_iters = sanitize_exprs(outermost.iter)
outermost_targets = sanitize_exprs(outermost.target)
for expr in outermost_iters:
self.visit(expr) # set self.last_value (to something and hope for the best)
with ExecuteInInnerScope(self, label):
for expr in outermost_targets:
self.visit(expr) # use self.last_value
self.last_value = None
for expr in outermost.ifs:
self.visit(expr)
# TODO: there's also an is_async field we might want to use in a future version of Pyan.
for gen in moregens:
targets = sanitize_exprs(gen.target)
values = sanitize_exprs(gen.iter)
self.analyze_binding(targets, values)
for expr in gen.ifs:
self.visit(expr)
self.visit(getattr(node, field1)) # e.g. node.elt
if field2:
self.visit(getattr(node, field2))
def visit_Call(self, node):
self.logger.debug("Call %s, %s:%s" % (get_ast_node_name(node.func), self.filename, node.lineno))
# visit args to detect uses
for arg in node.args:
self.visit(arg)
for kw in node.keywords:
self.visit(kw.value)
# see if we can predict the result
try:
result_node = self.resolve_builtins(node)
except UnresolvedSuperCallError:
result_node = None
if isinstance(result_node, Node): # resolved result
self.last_value = result_node
from_node = self.get_node_of_current_namespace()
to_node = result_node
self.logger.debug("Use from %s to %s (via resolved call to built-ins)" % (from_node, to_node))
if self.add_uses_edge(from_node, to_node):
self.logger.info(
"New edge added for Use from %s to %s (via resolved call to built-ins)" % (from_node, to_node)
)
else: # generic function call
# Visit the function name part last, so that inside a binding form,
# it will be left standing as self.last_value.
self.visit(node.func)
# If self.last_value matches a known class i.e. the call was of the
# form MyClass(), add a uses edge to MyClass.__init__().
#
# We need to do this manually, because there is no text "__init__"
# at the call site.
#
# In this lookup to self.class_base_ast_nodes we don't care about
# the AST nodes; the keys just conveniently happen to be the Nodes
# of known classes.
#
if self.last_value in self.class_base_ast_nodes:
from_node = self.get_node_of_current_namespace()
class_node = self.last_value
to_node = self.get_node(class_node.get_name(), "__init__", None, flavor=Flavor.METHOD)
self.logger.debug("Use from %s to %s (call creates an instance)" % (from_node, to_node))
if self.add_uses_edge(from_node, to_node):
self.logger.info(
"New edge added for Use from %s to %s (call creates an instance)" % (from_node, to_node)
)
def visit_With(self, node):
self.logger.debug("With (context manager), %s:%s" % (self.filename, node.lineno))
def add_uses_enter_exit_of(graph_node):
# add uses edges to __enter__ and __exit__ methods of given Node
if isinstance(graph_node, Node):
from_node = self.get_node_of_current_namespace()
withed_obj_node = graph_node
self.logger.debug("Use from %s to With %s" % (from_node, withed_obj_node))
for methodname in ("__enter__", "__exit__"):
to_node = self.get_node(withed_obj_node.get_name(), methodname, None, flavor=Flavor.METHOD)
if self.add_uses_edge(from_node, to_node):
self.logger.info("New edge added for Use from %s to %s" % (from_node, to_node))
for withitem in node.items:
expr = withitem.context_expr
vars = withitem.optional_vars
# XXX: we currently visit expr twice (again in analyze_binding()) if vars is not None
self.last_value = None
self.visit(expr)
add_uses_enter_exit_of(self.last_value)
self.last_value = None
if vars is not None:
# bind optional_vars
#
# TODO: For now, we support only the following (most common) case:
# - only one binding target, vars is ast.Name
# (not ast.Tuple or something else)
# - the variable will point to the object that was with'd
# (i.e. we assume the object's __enter__() method
# to finish with "return self")
#
if isinstance(vars, ast.Name):
self.analyze_binding(sanitize_exprs(vars), sanitize_exprs(expr))
else:
self.visit(vars) # just capture any uses on the With line itself
for stmt in node.body:
self.visit(stmt)
###########################################################################
# Analysis helpers
def analyze_functiondef(self, ast_node):
"""Analyze a function definition.
Visit decorators, and if this is a method definition, capture the name
of the first positional argument to denote "self", like Python does.
Return (self_name, flavor), where self_name the name representing self,
or None if not applicable; and flavor is a Flavor, specifically one of
FUNCTION, METHOD, STATICMETHOD or CLASSMETHOD."""
if not isinstance(ast_node, (ast.AsyncFunctionDef, ast.FunctionDef)):
raise TypeError("Expected ast.FunctionDef; got %s" % (type(ast_node)))
# Visit decorators
self.last_value = None
deco_names = []
for deco in ast_node.decorator_list: