# Licensed under the GPL: https://www.gnu.org/licenses/old-licenses/gpl-2.0.html # For details: https://github.com/pylint-dev/pylint/blob/main/LICENSE # Copyright (c) https://github.com/pylint-dev/pylint/blob/main/CONTRIBUTORS.txt """Visitor doing some post-processing on the astroid tree. Try to resolve definitions (namespace) dictionary, relationship... """ from __future__ import annotations import collections import os import traceback from abc import ABC, abstractmethod from typing import Callable, Optional import astroid from astroid import nodes from pylint import constants from pylint.pyreverse import utils _WrapperFuncT = Callable[[Callable[[str], nodes.Module], str], Optional[nodes.Module]] def _astroid_wrapper( func: Callable[[str], nodes.Module], modname: str ) -> nodes.Module | None: print(f"parsing {modname}...") try: return func(modname) except astroid.exceptions.AstroidBuildingException as exc: print(exc) except Exception: # pylint: disable=broad-except traceback.print_exc() return None class IdGeneratorMixIn: """Mixin adding the ability to generate integer uid.""" def __init__(self, start_value: int = 0) -> None: self.id_count = start_value def init_counter(self, start_value: int = 0) -> None: """Init the id counter.""" self.id_count = start_value def generate_id(self) -> int: """Generate a new identifier.""" self.id_count += 1 return self.id_count class Project: """A project handle a set of modules / packages.""" def __init__(self, name: str = ""): self.name = name self.uid: int | None = None self.path: str = "" self.modules: list[nodes.Module] = [] self.locals: dict[str, nodes.Module] = {} self.__getitem__ = self.locals.__getitem__ self.__iter__ = self.locals.__iter__ self.values = self.locals.values self.keys = self.locals.keys self.items = self.locals.items def add_module(self, node: nodes.Module) -> None: self.locals[node.name] = node self.modules.append(node) def get_module(self, name: str) -> nodes.Module: return self.locals[name] def get_children(self) -> list[nodes.Module]: return self.modules def __repr__(self) -> str: return f"" class Linker(IdGeneratorMixIn, utils.LocalsVisitor): """Walk on the project tree and resolve relationships. According to options the following attributes may be added to visited nodes: * uid, a unique identifier for the node (on astroid.Project, astroid.Module, astroid.Class and astroid.locals_type). Only if the linker has been instantiated with tag=True parameter (False by default). * Function a mapping from locals names to their bounded value, which may be a constant like a string or an integer, or an astroid node (on astroid.Module, astroid.Class and astroid.Function). * instance_attrs_type as locals_type but for klass member attributes (only on astroid.Class) * associations_type as instance_attrs_type but for association relationships * aggregations_type as instance_attrs_type but for aggregations relationships """ def __init__(self, project: Project, tag: bool = False) -> None: IdGeneratorMixIn.__init__(self) utils.LocalsVisitor.__init__(self) # tag nodes or not self.tag = tag # visited project self.project = project self.associations_handler = AggregationsHandler() self.associations_handler.set_next(OtherAssociationsHandler()) def visit_project(self, node: Project) -> None: """Visit a pyreverse.utils.Project node. * optionally tag the node with a unique id """ if self.tag: node.uid = self.generate_id() for module in node.modules: self.visit(module) def visit_module(self, node: nodes.Module) -> None: """Visit an astroid.Module node. * set the locals_type mapping * set the depends mapping * optionally tag the node with a unique id """ if hasattr(node, "locals_type"): return node.locals_type = collections.defaultdict(list) node.depends = [] if self.tag: node.uid = self.generate_id() def visit_classdef(self, node: nodes.ClassDef) -> None: """Visit an astroid.Class node. * set the locals_type and instance_attrs_type mappings * optionally tag the node with a unique id """ if hasattr(node, "locals_type"): return node.locals_type = collections.defaultdict(list) if self.tag: node.uid = self.generate_id() # resolve ancestors for baseobj in node.ancestors(recurs=False): specializations = getattr(baseobj, "specializations", []) specializations.append(node) baseobj.specializations = specializations # resolve instance attributes node.instance_attrs_type = collections.defaultdict(list) node.aggregations_type = collections.defaultdict(list) node.associations_type = collections.defaultdict(list) for assignattrs in tuple(node.instance_attrs.values()): for assignattr in assignattrs: if not isinstance(assignattr, nodes.Unknown): self.associations_handler.handle(assignattr, node) self.handle_assignattr_type(assignattr, node) def visit_functiondef(self, node: nodes.FunctionDef) -> None: """Visit an astroid.Function node. * set the locals_type mapping * optionally tag the node with a unique id """ if hasattr(node, "locals_type"): return node.locals_type = collections.defaultdict(list) if self.tag: node.uid = self.generate_id() def visit_assignname(self, node: nodes.AssignName) -> None: """Visit an astroid.AssignName node. handle locals_type """ # avoid double parsing done by different Linkers.visit # running over the same project: if hasattr(node, "_handled"): return node._handled = True if node.name in node.frame(future=True): frame = node.frame(future=True) else: # the name has been defined as 'global' in the frame and belongs # there. frame = node.root() if not hasattr(frame, "locals_type"): # If the frame doesn't have a locals_type yet, # it means it wasn't yet visited. Visit it now # to add what's missing from it. if isinstance(frame, nodes.ClassDef): self.visit_classdef(frame) elif isinstance(frame, nodes.FunctionDef): self.visit_functiondef(frame) else: self.visit_module(frame) current = frame.locals_type[node.name] frame.locals_type[node.name] = list(set(current) | utils.infer_node(node)) @staticmethod def handle_assignattr_type(node: nodes.AssignAttr, parent: nodes.ClassDef) -> None: """Handle an astroid.assignattr node. handle instance_attrs_type """ current = set(parent.instance_attrs_type[node.attrname]) parent.instance_attrs_type[node.attrname] = list( current | utils.infer_node(node) ) def visit_import(self, node: nodes.Import) -> None: """Visit an astroid.Import node. resolve module dependencies """ context_file = node.root().file for name in node.names: relative = astroid.modutils.is_relative(name[0], context_file) self._imported_module(node, name[0], relative) def visit_importfrom(self, node: nodes.ImportFrom) -> None: """Visit an astroid.ImportFrom node. resolve module dependencies """ basename = node.modname context_file = node.root().file if context_file is not None: relative = astroid.modutils.is_relative(basename, context_file) else: relative = False for name in node.names: if name[0] == "*": continue # analyze dependencies fullname = f"{basename}.{name[0]}" if fullname.find(".") > -1: try: fullname = astroid.modutils.get_module_part(fullname, context_file) except ImportError: continue if fullname != basename: self._imported_module(node, fullname, relative) def compute_module(self, context_name: str, mod_path: str) -> bool: """Should the module be added to dependencies ?""" package_dir = os.path.dirname(self.project.path) if context_name == mod_path: return False # astroid does return a boolean but is not typed correctly yet return astroid.modutils.module_in_path(mod_path, (package_dir,)) # type: ignore[no-any-return] def _imported_module( self, node: nodes.Import | nodes.ImportFrom, mod_path: str, relative: bool ) -> None: """Notify an imported module, used to analyze dependencies.""" module = node.root() context_name = module.name if relative: mod_path = f"{'.'.join(context_name.split('.')[:-1])}.{mod_path}" if self.compute_module(context_name, mod_path): # handle dependencies if not hasattr(module, "depends"): module.depends = [] mod_paths = module.depends if mod_path not in mod_paths: mod_paths.append(mod_path) class AssociationHandlerInterface(ABC): @abstractmethod def set_next( self, handler: AssociationHandlerInterface ) -> AssociationHandlerInterface: pass @abstractmethod def handle(self, node: nodes.AssignAttr, parent: nodes.ClassDef) -> None: pass class AbstractAssociationHandler(AssociationHandlerInterface): """ Chain of Responsibility for handling types of association, useful to expand in the future if we want to add more distinct associations. Every link of the chain checks if it's a certain type of association. If no association is found it's set as a generic association in `associations_type`. The default chaining behavior is implemented inside the base handler class. """ _next_handler: AssociationHandlerInterface def set_next( self, handler: AssociationHandlerInterface ) -> AssociationHandlerInterface: self._next_handler = handler return handler @abstractmethod def handle(self, node: nodes.AssignAttr, parent: nodes.ClassDef) -> None: if self._next_handler: self._next_handler.handle(node, parent) class AggregationsHandler(AbstractAssociationHandler): def handle(self, node: nodes.AssignAttr, parent: nodes.ClassDef) -> None: if isinstance(node.parent, (nodes.AnnAssign, nodes.Assign)) and isinstance( node.parent.value, astroid.node_classes.Name ): current = set(parent.aggregations_type[node.attrname]) parent.aggregations_type[node.attrname] = list( current | utils.infer_node(node) ) else: super().handle(node, parent) class OtherAssociationsHandler(AbstractAssociationHandler): def handle(self, node: nodes.AssignAttr, parent: nodes.ClassDef) -> None: current = set(parent.associations_type[node.attrname]) parent.associations_type[node.attrname] = list(current | utils.infer_node(node)) def project_from_files( files: list[str], func_wrapper: _WrapperFuncT = _astroid_wrapper, project_name: str = "no name", black_list: tuple[str, ...] = constants.DEFAULT_IGNORE_LIST, ) -> Project: """Return a Project from a list of files or modules.""" # build the project representation astroid_manager = astroid.MANAGER project = Project(project_name) for something in files: if not os.path.exists(something): fpath = astroid.modutils.file_from_modpath(something.split(".")) elif os.path.isdir(something): fpath = os.path.join(something, "__init__.py") else: fpath = something ast = func_wrapper(astroid_manager.ast_from_file, fpath) if ast is None: continue project.path = project.path or ast.file project.add_module(ast) base_name = ast.name # recurse in package except if __init__ was explicitly given if ast.package and something.find("__init__") == -1: # recurse on others packages / modules if this is a package for fpath in astroid.modutils.get_module_files( os.path.dirname(ast.file), black_list ): ast = func_wrapper(astroid_manager.ast_from_file, fpath) if ast is None or ast.name == base_name: continue project.add_module(ast) return project