diff --git a/.github/workflows/deploy-docs.yml b/.github/workflows/deploy-docs.yml index 951c4af..03f61cd 100644 --- a/.github/workflows/deploy-docs.yml +++ b/.github/workflows/deploy-docs.yml @@ -18,33 +18,33 @@ jobs: runs-on: ubuntu-latest steps: - - name: checkout - uses: actions/checkout@v3 - with: - fetch-depth: 0 - - - name: Setup Mamba - uses: mamba-org/provision-with-micromamba@v15 - with: - environment-file: false - - - name: Build environment - run: | - micromamba create --name TEST python=3 pip --file requirements-dev.txt --channel conda-forge - micromamba activate TEST - python -m pip install -e . --force-reinstall - - - name: Build documentation - run: | - set -e - micromamba activate TEST - pushd doc - make clean html linkcheck - popd - - - name: Deploy - if: github.event_name == 'release' || github.event_name == 'push' - uses: peaceiris/actions-gh-pages@v3 - with: - github_token: ${{ secrets.GITHUB_TOKEN }} - publish_dir: doc/_build/html + - name: checkout + uses: actions/checkout@v3 + with: + fetch-depth: 0 + + - name: Setup Mamba + uses: mamba-org/provision-with-micromamba@v15 + with: + environment-file: false + + - name: Build environment + run: | + micromamba create --name TEST python=3 pip --file requirements-dev.txt --channel conda-forge + micromamba activate TEST + python -m pip install -e . --force-reinstall + + - name: Build documentation + run: | + set -e + micromamba activate TEST + pushd doc + make clean html linkcheck + popd + + - name: Deploy + if: github.event_name == 'release' || github.event_name == 'push' + uses: peaceiris/actions-gh-pages@v3 + with: + github_token: ${{ secrets.GITHUB_TOKEN }} + publish_dir: doc/_build/html diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 9344aae..2914c53 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -14,7 +14,7 @@ jobs: runs-on: ${{ matrix.os }} strategy: matrix: - python-version: ["3.6", "3.7", "3.8", "3.9", "3.10", "3.11"] + python-version: ["3.7", "3.8", "3.9", "3.10", "3.11"] os: [windows-latest, ubuntu-latest, macos-latest] fail-fast: false @@ -29,8 +29,8 @@ jobs: - name: Setup Micromamba uses: mamba-org/provision-with-micromamba@v15 with: - environment-file: false - + environment-file: false + - name: Python ${{ matrix.python-version }} run: | micromamba create --name TEST python=${{ matrix.python-version }} pip --file requirements-dev.txt --channel conda-forge @@ -47,4 +47,3 @@ jobs: micromamba activate TEST coverage report -m codecov - diff --git a/depfinder/cli.py b/depfinder/cli.py index 9d80fec..a65afd1 100644 --- a/depfinder/cli.py +++ b/depfinder/cli.py @@ -29,22 +29,21 @@ from __future__ import absolute_import, division, print_function from argparse import ArgumentParser -from collections import defaultdict import logging import os from pprint import pprint import itertools import pdb import sys +from typing import Dict, Iterable, Set import yaml from . import main from .inspection import parse_file -from .main import (simple_import_search, notebook_path_to_dependencies, - sanitize_deps) +from .main import simple_import_search, notebook_path_to_dependencies, sanitize_deps -logger = logging.getLogger('depfinder') +logger = logging.getLogger("depfinder") class InvalidSelection(RuntimeError): @@ -68,81 +67,92 @@ def _init_parser(): nargs="?", ) p.add_argument( - '-y', - '--yaml', - action='store_true', + "-y", + "--yaml", + action="store_true", default=False, - help=("Output in syntactically valid yaml when true. Defaults to " - "%(default)s")) + help=( + "Output in syntactically valid yaml when true. Defaults to " "%(default)s" + ), + ) p.add_argument( - '-V', - '--version', - action='store_true', + "-V", + "--version", + action="store_true", default=False, - help="Print out the version of depfinder and exit" + help="Print out the version of depfinder and exit", ) p.add_argument( - '--no-remap', - action='store_true', + "--no-remap", + action="store_true", default=False, - help=("Do not remap the names of the imported libraries to their " - "proper conda name") + help=( + "Do not remap the names of the imported libraries to their " + "proper conda name" + ), ) p.add_argument( - '-v', - '--verbose', - action='store_true', + "-v", + "--verbose", + action="store_true", default=False, - help="Enable debug level logging info from depfinder" + help="Enable debug level logging info from depfinder", ) p.add_argument( - '-q', - '--quiet', - action='store_true', + "-q", + "--quiet", + action="store_true", default=False, - help="Turn off all logging from depfinder" + help="Turn off all logging from depfinder", ) p.add_argument( - '-k', '--key', + "-k", + "--key", action="append", default=[], - help=("Select some or all of the output keys. Valid options are " - "'required', 'optional', 'builtin', 'relative', 'all'. Defaults " - "to 'all'") + help=( + "Select some or all of the output keys. Valid options are " + "'required', 'optional', 'builtin', 'relative', 'all'. Defaults " + "to 'all'" + ), ) p.add_argument( - '--conda', + "--conda", action="store_true", default=False, - help=("Format output so it can be passed as an argument to conda " - "install or conda create") + help=( + "Format output so it can be passed as an argument to conda " + "install or conda create" + ), ) p.add_argument( - '--pdb', + "--pdb", action="store_true", help="Enable PDB debugging on exception", default=False, ) p.add_argument( - '--ignore', - default='', - help="Comma separated list of file patterns not to inspect" + "--ignore", + default="", + help="Comma separated list of file patterns not to inspect", ) p.add_argument( - '--strict', + "--strict", default=False, action="store_true", - help=("Immediately raise an Exception if any files fail to parse. Defaults to off.") + help=( + "Immediately raise an Exception if any files fail to parse. Defaults to off." + ), ) p.add_argument( - '--custom-namespaces', - default='', + "--custom-namespaces", + default="", help=( "A comma-separated list of custom namespace packages. " "Listing namespaces here will enable depfinder to properly report " "dependencies when a namespace is in use (e.g., depfinder will report " "both foo.bar and foo.baz instead of foo when run with --custom-namespaces=foo)." - ) + ), ) return p @@ -151,14 +161,17 @@ def cli(): p = _init_parser() args = p.parse_args() if args.verbose and args.quiet: - msg = ("You have enabled both verbose mode (--verbose or -v) and " - "quiet mode (-q or --quiet). Please pick one. Exiting...") + msg = ( + "You have enabled both verbose mode (--verbose or -v) and " + "quiet mode (-q or --quiet). Please pick one. Exiting..." + ) raise InvalidSelection(msg) if args.pdb: # set the pdb_hook as the except hook for all exceptions - def pdb_hook(exctype, value, traceback): - pdb.post_mortem(traceback) + def pdb_hook(exctype, value, traceback): # type: ignore + pdb.post_mortem(traceback) # type: ignore + sys.excepthook = pdb_hook cs = args.custom_namespaces.split(",") @@ -173,7 +186,7 @@ def pdb_hook(exctype, value, traceback): loglevel = logging.DEBUG stream_handler = logging.StreamHandler() stream_handler.setLevel(loglevel) - f = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' + f = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" formatter = logging.Formatter(f) stream_handler.setFormatter(formatter) logger.setLevel(loglevel) @@ -183,6 +196,7 @@ def pdb_hook(exctype, value, traceback): # handle the case where the user just cares about the version. Print # version and exit from . import __version__ + print(__version__) return 0 @@ -192,11 +206,9 @@ def pdb_hook(exctype, value, traceback): logger.warning("positional argument `file_or_directory` not provided.") raise RuntimeError("positional argument `file_or_directory` is required") keys = args.key - if keys == []: - keys = None - logger.debug('keys: %s', keys) + logger.debug("keys: %s", keys) - def dump_deps(deps, keys): + def dump_deps(deps: Dict[str, Set[str]], keys: Iterable[str]): """ Helper function to print the dependencies to the console. @@ -205,34 +217,45 @@ def dump_deps(deps, keys): deps : dict Dictionary of dependencies that were found """ - if keys is None: + if not keys: keys = list(deps.keys()) - deps = {k: list(v) for k, v in deps.items() if k in keys} + sorted_deps = {k: sorted(v) for k, v in deps.items() if k in keys} if args.yaml: - print(yaml.dump(deps, default_flow_style=False)) + print(yaml.dump(sorted_deps, default_flow_style=False)) elif args.conda: - list_of_deps = [item for sublist in itertools.chain(deps.values()) - for item in sublist] - print(' '.join(list_of_deps)) + list_of_deps = [ + item + for sublist in itertools.chain( + deps.get("required", set()), deps["questionable"] + ) + for item in sublist + ] + print(" ".join(list_of_deps)) else: pprint(deps) if os.path.isdir(file_or_dir): - logger.debug("Treating {} as a directory and recursively searching " - "it for python files".format(file_or_dir)) + logger.debug( + "Treating {} as a directory and recursively searching " + "it for python files".format(file_or_dir) + ) # directories are a little easier from the purpose of the API call. # print the dependencies to the console and then exit - ignore = args.ignore.split(',') + ignore = args.ignore.split(",") deps = simple_import_search( - file_or_dir, remap=not args.no_remap, - ignore=ignore, custom_namespaces=cs, + file_or_dir, + remap=not args.no_remap, + ignore=ignore, + custom_namespaces=cs, ) dump_deps(deps, keys) return 0 elif os.path.isfile(file_or_dir): - if file_or_dir.endswith('ipynb'): - logger.debug("Treating {} as a jupyter notebook and searching " - "all of its code cells".format(file_or_dir)) + if file_or_dir.endswith("ipynb"): + logger.debug( + "Treating {} as a jupyter notebook and searching " + "all of its code cells".format(file_or_dir) + ) deps = notebook_path_to_dependencies( file_or_dir, remap=not args.no_remap, @@ -242,24 +265,22 @@ def dump_deps(deps, keys): # print the dependencies to the console and then exit dump_deps(sanitized, keys) return 0 - elif file_or_dir.endswith('.py'): - logger.debug("Treating {} as a single python file" - "".format(file_or_dir)) - mod, path, import_finder = parse_file(file_or_dir, custom_namespaces=cs) - mods = defaultdict(set) - for k, v in import_finder.describe().items(): - mods[k].update(v) - deps = {k: sorted(list(v)) for k, v in mods.items() if v} + elif file_or_dir.endswith(".py"): + logger.debug("Treating {} as a single python file" "".format(file_or_dir)) + _, _, import_finder = parse_file(file_or_dir, custom_namespaces=cs) + mods = import_finder.describe() - sanitized = sanitize_deps(deps) + sanitized = sanitize_deps(mods) # print the dependencies to the console and then exit dump_deps(sanitized, keys) return 0 else: # Any file with a suffix that is not ".ipynb" or ".py" will not # be parsed correctly - msg = ("depfinder is only configured to work with jupyter " - "notebooks and python source code files. It is anticipated " - "that the file {} will not work with depfinder" - "".format(file_or_dir)) + msg = ( + "depfinder is only configured to work with jupyter " + "notebooks and python source code files. It is anticipated " + "that the file {} will not work with depfinder" + "".format(file_or_dir) + ) raise RuntimeError(msg) diff --git a/depfinder/inspection.py b/depfinder/inspection.py index 8d3f713..253b28e 100644 --- a/depfinder/inspection.py +++ b/depfinder/inspection.py @@ -32,30 +32,45 @@ import ast import logging import os -import sys from collections import defaultdict -from typing import Union +from typing import Dict, Iterable, List, Set, Tuple + +from pydantic import BaseModel from .stdliblist import builtin_modules from .utils import ( AST_QUESTIONABLE, + ImportType, namespace_packages, - SKETCHY_TYPES_TABLE, + ast_types_to_str, + ast_import_types, + ImportMetadata, ) -logger = logging.getLogger('depfinder') +logger = logging.getLogger("depfinder.inspection") -PACKAGE_NAME = None +current_package_name: str = "" STRICT_CHECKING = False -def get_top_level_import_name(name, custom_namespaces=None): +class FoundModules(BaseModel): + required: Set[str] + questionable: Set[str] + builtin: Set[str] + relative: Set[str] + + +def get_top_level_import_name(name: str, custom_namespaces: Iterable[str] = ()) -> str: num_dot = name.count(".") custom_namespaces = custom_namespaces or [] - if name in namespace_packages or name in custom_namespaces or name in builtin_modules: + if ( + name in namespace_packages + or name in custom_namespaces + or name in builtin_modules + ): return name elif any( ((num_dot - nsp.count(".")) == 1) and name.startswith(nsp + ".") @@ -65,12 +80,11 @@ def get_top_level_import_name(name, custom_namespaces=None): # foo.bar return name else: - if '.' not in name: + if "." not in name: return name else: return get_top_level_import_name( - name.rsplit('.', 1)[0], - custom_namespaces=custom_namespaces + name.rsplit(".", 1)[0], custom_namespaces=custom_namespaces ) @@ -92,20 +106,23 @@ class ImportFinder(ast.NodeVisitor): """ - def __init__(self, filename='', custom_namespaces=None): + def __init__(self, filename: str = "", custom_namespaces: Iterable[str] = ()): self.filename = filename - self.required_modules = set() - self.sketchy_modules = set() - self.builtin_modules = set() - self.relative_modules = set() - self.imports = [] - self.import_froms = [] - self.total_imports = defaultdict(dict) - self.sketchy_nodes = {} - self.custom_namespaces = custom_namespaces or [] + self.required_modules: Set[str] = set() + self.questionable_modules: Set[str] = set() + self.builtin_modules: Set[str] = set() + self.relative_modules: Set[str] = set() + self.imports: List[ast.AST] = [] + self.import_froms: List[ast.AST] = [] + self.total_imports_new: List[ImportMetadata] = list() + self.total_imports: Dict[ + str, Dict[Tuple[str, int], ImportMetadata] + ] = defaultdict(dict) + self.sketchy_nodes: Dict[ast.AST, ast.AST] = {} + self.custom_namespaces: Iterable[str] = custom_namespaces or () super(ImportFinder, self).__init__() - def visit(self, node): + def visit(self, node: ast.AST): """Recursively visit all ast nodes. Look for Import and ImportFrom nodes. Classify them as being imports @@ -131,7 +148,7 @@ def visit_Import(self, node: ast.Import): an ast.Import node is something like 'import bar' - If ImportCatcher is inside of a try block then the import that has just + If ImportFinder is inside of a try block then the import that has just been encountered will be added to the `sketchy_modules` instance attribute. Otherwise the module will be added to the `required_modules` instance attribute @@ -139,23 +156,27 @@ def visit_Import(self, node: ast.Import): self.imports.append(node) self._add_to_total_imports(node) - mods = set([ - get_top_level_import_name(name.name, custom_namespaces=self.custom_namespaces) - for name in node.names - ]) - for mod in mods: - self._add_import_node(mod) + imports: Set[str] = set() + for name in node.names: + import_name = get_top_level_import_name( + name.name, custom_namespaces=self.custom_namespaces + ) + imports.add(import_name) + + for import_name in imports: + self._add_import_node(import_name) def visit_ImportFrom(self, node: ast.ImportFrom): """Executes when an ast.ImportFrom node is encountered an ast.ImportFrom node is something like 'from foo import bar' - If ImportCatcher is inside of a try block then the import that has just + If ImportFinder is inside of a try block then the import that has just been encountered will be added to the `sketchy_modules` instance attribute. Otherwise the module will be added to the `required_modules` instance attribute """ + logger.debug(f"node={node}, node.lineno={node.lineno}") self.import_froms.append(node) if node.module is None: # this is a relative import like 'from . import bar' @@ -176,30 +197,53 @@ def visit_ImportFrom(self, node: ast.ImportFrom): ) self._add_import_node(node_name) - def _add_to_total_imports(self, node: Union[ast.Import, ast.ImportFrom]): - import_metadata = {} + def _add_to_total_imports(self, node: ast_import_types): + logger.debug(f"_add_to_total_imports node={node}, node.lineno={node.lineno}") + + import_metadata = ImportMetadata() try: - import_metadata.update({'exact_line': ast.unparse(node)}) + import_metadata.exact_line = ast.unparse(node) except AttributeError: + # what are the circumstances where we hit this exception? pass - import_metadata.update({v: False for v in SKETCHY_TYPES_TABLE.values()}) - import_metadata.update({SKETCHY_TYPES_TABLE[node.__class__]: True for node in self.sketchy_nodes}) - names = set() + # import_metadata.update({v: False for v in SKETCHY_TYPES_TABLE.values()}) + # For all of the sketchy nodes, update the import metadata to "True" for + # each of the node types that our import is found within. e.g., if the + # import is found within a try block and a function definition, then + # set import_metadata.try = True and import_metadata.function = True + + for sketchy_node in self.sketchy_nodes: + logger.debug(f"sketchy_node={sketchy_node}") + import_metadata.__setattr__(ast_types_to_str[sketchy_node.__class__], True) + if isinstance(node, ast.Import): - _names = set(name.name for name in node.names) - import_metadata['import'] = _names - names.update(_names) + # import nodes can have multiple imports, e.g. + # import foo, bar, baz + names: Set[str] = set() + for node_alias in node.names: + names.add(node_alias.name) + import_metadata.imported_modules = names + import_metadata.import_type = ImportType.import_normal elif isinstance(node, ast.ImportFrom): - import_metadata['import_from'] = {node.module} - names.add(node.module) + import_metadata.import_type = ImportType.import_from + if node.module is not None: + import_metadata.imported_modules = {node.module} else: - raise NotImplementedError(f"Expected ast.Import or ast.ImportFrom this is {type(node)}") - - for name in names: - self.total_imports[name].update({(self.filename, node.lineno): import_metadata}) + # defensive coding. This will only be hit if a new + # `visit_*` method is added to this class + raise NotImplementedError( + f"Expected ast.Import or ast.ImportFrom this is {type(node)}" + ) + import_metadata.lineno = node.lineno + import_metadata.filename = self.filename + self.total_imports_new.append(import_metadata) + for import_name in import_metadata.imported_modules: + self.total_imports[import_name].update( + {(self.filename, node.lineno): import_metadata} + ) - def _add_import_node(self, node_name): + def _add_import_node(self, node_name: str): # see if the module is a builtin if node_name in builtin_modules: self.builtin_modules.add(node_name) @@ -207,19 +251,41 @@ def _add_import_node(self, node_name): # see if we are in a try block if self.sketchy_nodes: - self.sketchy_modules.add(node_name) + self.questionable_modules.add(node_name) return # if none of the above cases are true, it is likely that this # ImportFrom node occurs at the top level of the module self.required_modules.add(node_name) - def describe(self): + def describe_pydantic(self) -> FoundModules: + """Return the found imports + + Returns + ------- + FoundModules : + 'required': The modules that were encountered outside of a + try/except block + 'questionable': The modules that were encountered inside of a + try/except block + 'relative': The modules that were imported via relative import + syntax + 'builtin' : The modules that are part of the standard library + """ + found_modules = FoundModules( + required=self.required_modules, + relative=self.relative_modules, + questionable=self.questionable_modules, + builtin=self.builtin_modules, + ) + return found_modules + + def describe(self) -> Dict[str, Set[str]]: """Return the found imports Returns ------- - dict : + FoundModules : 'required': The modules that were encountered outside of a try/except block 'questionable': The modules that were encountered inside of a @@ -228,20 +294,23 @@ def describe(self): syntax 'builtin' : The modules that are part of the standard library """ - desc = { - 'required': self.required_modules, - 'relative': self.relative_modules, - 'questionable': self.sketchy_modules, - 'builtin': self.builtin_modules + deps = { + "required": self.required_modules, + "relative": self.relative_modules, + "questionable": self.questionable_modules, + "builtin": self.builtin_modules, } - desc = {k: v for k, v in desc.items() if v} - return desc + + deps = {k: v for k, v in deps.items() if v} + return deps def __repr__(self): - return 'ImportCatcher: %s' % repr(self.describe()) + return "ImportFinder: %s" % repr(self.describe()) -def get_imported_libs(code, filename='', custom_namespaces=None): +def get_imported_libs( + code: str, filename: str = "", custom_namespaces: Iterable[str] = () +) -> ImportFinder: """Given a code snippet, return a list of the imported libraries Parameters @@ -251,8 +320,8 @@ def get_imported_libs(code, filename='', custom_namespaces=None): Returns ------- - ImportCatcher - The ImportCatcher is the object in `depfinder` that contains all the + ImportFinder + The ImportFinder is the object in `depfinder` that contains all the information regarding which imports were found where. You will most likely be interested in calling the describe() function on this return value. @@ -269,15 +338,16 @@ def get_imported_libs(code, filename='', custom_namespaces=None): 'required': {'stdlib_list'}} """ # skip ipython notebook lines - code = '\n'.join([line for line in code.split('\n') - if not line.startswith('%')]) + code = "\n".join([line for line in code.split("\n") if not line.startswith("%")]) tree = ast.parse(code) import_finder = ImportFinder(filename=filename, custom_namespaces=custom_namespaces) import_finder.visit(tree) return import_finder -def parse_file(python_file, custom_namespaces=None): +def parse_file( + python_file: str, custom_namespaces: Iterable[str] = () +) -> Tuple[str, str, ImportFinder]: """Parse a single python file Parameters @@ -287,40 +357,41 @@ def parse_file(python_file, custom_namespaces=None): Returns ------- - catchers : tuple - Yields tuples of (module_name, full_path_to_module, ImportCatcher) + tuple + Yields tuples of (module_name, full_path_to_module, ImportFinder) """ - global PACKAGE_NAME - if PACKAGE_NAME is None: - PACKAGE_NAME = os.path.basename(python_file).split('.')[0] - logger.debug("Setting PACKAGE_NAME global variable to {}" - "".format(PACKAGE_NAME)) + global current_package_name + if not current_package_name: + # this might have a potential bug if the filename has more than one "." + current_package_name = os.path.basename(python_file).split(".")[0] + logger.debug( + "Setting PACKAGE_NAME global variable to {}" "".format(current_package_name) + ) # Try except block added for adal package which has a BOM at the beginning, # requiring a different encoding to load properly try: - with open(python_file, 'r') as f: + with open(python_file, "r") as f: code = f.read() - catcher = get_imported_libs( + import_finder = get_imported_libs( code, filename=python_file, custom_namespaces=custom_namespaces ) except SyntaxError: - with open(python_file, 'r', encoding='utf-8-sig') as f: + with open(python_file, "r", encoding="utf-8-sig") as f: code = f.read() - catcher = get_imported_libs( + import_finder = get_imported_libs( code, filename=python_file, custom_namespaces=custom_namespaces ) - catcher.total_imports = dict(catcher.total_imports) - mod_name = os.path.split(python_file)[:-3] - return mod_name, python_file, catcher + import_finder.total_imports = dict(import_finder.total_imports) + return current_package_name, python_file, import_finder -def iterate_over_library(path_to_source_code, custom_namespaces=None): +def iterate_over_library( + path_to_source_code: str, custom_namespaces: Iterable[str] = () +) -> Iterable[Tuple[str, str, ImportFinder]]: """Helper function to recurse into a library and find imports in .py files. This allows the user to apply filters on the user-side to exclude imports based on their file names. - `conda-skeletor `_ - makes heavy use of this function Parameters ---------- @@ -328,24 +399,27 @@ def iterate_over_library(path_to_source_code, custom_namespaces=None): Yields ------- - catchers : tuple - Yields tuples of (module_name, full_path_to_module, ImportCatcher) + tuple + Yields tuples of (module_name, full_path_to_module, ImportFinder) """ - global PACKAGE_NAME + global current_package_name global STRICT_CHECKING - if PACKAGE_NAME is None: - PACKAGE_NAME = os.path.basename(path_to_source_code).split('.')[0] - logger.debug("Setting PACKAGE_NAME global variable to {}" - "".format(PACKAGE_NAME)) - skipped_files = [] - all_files = [] - for parent, folders, files in os.walk(path_to_source_code): + if not current_package_name: + current_package_name = os.path.basename(path_to_source_code).split(".")[0] + logger.debug( + "Setting PACKAGE_NAME global variable to {}" "".format(current_package_name) + ) + skipped_files: List[str] = [] + all_files: List[str] = [] + for parent, _, files in os.walk(path_to_source_code): for f in files: - if f.endswith('.py'): + if f.endswith(".py"): full_file_path = os.path.join(parent, f) all_files.append(full_file_path) try: - yield parse_file(full_file_path, custom_namespaces=custom_namespaces) + yield parse_file( + full_file_path, custom_namespaces=custom_namespaces + ) except Exception: logger.exception("Could not parse file: {}".format(full_file_path)) skipped_files.append(full_file_path) @@ -355,4 +429,6 @@ def iterate_over_library(path_to_source_code, custom_namespaces=None): logger.warn("%s: %s" % (str(idx), f)) if skipped_files and STRICT_CHECKING: - raise RuntimeError("Some files failed to parse. See logs for full stack traces.") + raise RuntimeError( + "Some files failed to parse. See logs for full stack traces." + ) diff --git a/depfinder/main.py b/depfinder/main.py index 99976d0..d1ce96d 100644 --- a/depfinder/main.py +++ b/depfinder/main.py @@ -35,16 +35,26 @@ import logging from collections import defaultdict from fnmatch import fnmatch +from typing import Any, Dict, Iterable, List, Set, Tuple + +from pydantic import BaseModel from .inspection import iterate_over_library, get_imported_libs -from .utils import pkg_data +from .utils import ImportMetadata, pkg_data + +from .stdliblist import builtin_modules as _builtin_modules -logger = logging.getLogger('depfinder') +logger = logging.getLogger("depfinder.main") STRICT_CHECKING = False -def simple_import_search(path_to_source_code, remap=True, ignore=None, custom_namespaces=None): +def simple_import_search( + path_to_source_code: str, + remap: bool = True, + ignore: Iterable[str] = (), + custom_namespaces: Iterable[str] = (), +) -> Dict[str, Set[str]]: """Return all imported modules in all .py files in `path_to_source_code` Parameters @@ -63,7 +73,7 @@ def simple_import_search(path_to_source_code, remap=True, ignore=None, custom_na ------- dict The list of all imported modules, sorted according to the keys listed - in the docstring of depfinder.ImportCatcher.describe() + in the docstring of depfinder.ImportFinder.describe() Examples -------- @@ -84,22 +94,27 @@ def simple_import_search(path_to_source_code, remap=True, ignore=None, custom_na 'stdlib_list', 'test_with_code']} """ - all_deps = defaultdict(set) - catchers = iterate_over_library(path_to_source_code, custom_namespaces=custom_namespaces) - for mod, path, catcher in catchers: + all_deps: Dict[str, Set[str]] = defaultdict(set) + import_finders = iterate_over_library( + path_to_source_code, custom_namespaces=custom_namespaces + ) + for _, path, import_finder in import_finders: # if ignore provided skip things which match the ignore pattern if ignore and any(fnmatch(path, i) for i in ignore): continue - for k, v in catcher.describe().items(): + for k, v in import_finder.describe().items(): all_deps[k].update(v) - all_deps = {k: sorted(list(v)) for k, v in all_deps.items() if v} if remap: return sanitize_deps(all_deps) return all_deps -def notebook_path_to_dependencies(path_to_notebook, remap=True, custom_namespaces=None): +def notebook_path_to_dependencies( + path_to_notebook: str, + remap: bool = True, + custom_namespaces: Iterable[str] = (), +) -> Dict[str, Set[str]]: """Helper function that turns a jupyter notebook into a list of dependencies Parameters @@ -129,31 +144,42 @@ def notebook_path_to_dependencies(path_to_notebook, remap=True, custom_namespace """ try: from IPython.core.inputsplitter import IPythonInputSplitter + transform = IPythonInputSplitter(line_input_checker=False).transform_cell except: - transform = lambda code: code + logger.warning( + "Could not import IPython. Jupyter notebook parsing will work better with IPython installed" + ) + + def transform(code: str) -> str: + # no-op, just return the code. match the transform_cell function from IPython + return code - nb = json.load(io.open(path_to_notebook, encoding='utf8')) - codeblocks = [''.join(cell['source']) for cell in nb['cells'] - if cell['cell_type'] == 'code'] - all_deps = defaultdict(set) + # Could also do this with nbconvert, i think? But this basically achieves the same thing + nb = json.load(io.open(path_to_notebook, encoding="utf8")) + codeblocks: List[str] = [ + "".join(cell["source"]) for cell in nb["cells"] if cell["cell_type"] == "code" + ] + all_deps: Dict[str, Set[str]] = defaultdict(set) for codeblock in codeblocks: codeblock = transform(codeblock) # TODO this may fail on py2/py3 syntax when running in the other runtime. # May want to consider updating some error handling around that case. # Will wait until that use case surfaces before modifying - deps_dict = get_imported_libs(codeblock, custom_namespaces=custom_namespaces).describe() + import_finder = get_imported_libs( + codeblock, custom_namespaces=custom_namespaces + ) + deps_dict = import_finder.describe() for k, v in deps_dict.items(): all_deps[k].update(v) - all_deps = {k: sorted(list(v)) for k, v in all_deps.items()} if remap: - return sanitize_deps(all_deps) + all_deps = sanitize_deps(all_deps) return all_deps -def sanitize_deps(deps_dict): +def sanitize_deps(dependencies: Dict[str, Set[str]]) -> Dict[str, Set[str]]: """ Helper function that takes the output of `notebook_path_to_dependencies` or `simple_import_search` and turns normalizes the import names to be @@ -169,36 +195,47 @@ def sanitize_deps(deps_dict): If remap is True: Sanitized `deps_dict` If remap is False: `deps_dict` """ - from .inspection import PACKAGE_NAME - new_deps_dict = {} - list_of_possible_fakes = set([v for val in pkg_data['_FAKE_PACKAGES'].values() for v in val]) - for k, packages_list in deps_dict.items(): + from .inspection import current_package_name + new_deps_dict: Dict[str, Set[str]] = {} + list_of_possible_fakes = set( + [v for val in pkg_data["_FAKE_PACKAGES"].values() for v in val] + ) + for k, packages_list in dependencies.items(): pkgs = copy.copy(packages_list) new_deps_dict[k] = set() for pkg in pkgs: # drop fake packages if pkg in list_of_possible_fakes: - logger.debug("Ignoring {} from the list of imports. It is " - "installed as part of another package. Set the " - "`--no-remap` cli flag if you want to disable " - "this".format(pkg)) + logger.debug( + "Ignoring {} from the list of imports. It is " + "installed as part of another package. Set the " + "`--no-remap` cli flag if you want to disable " + "this".format(pkg) + ) continue - if pkg == PACKAGE_NAME: - logger.debug("Ignoring {} from the list of imports. It is " - "the name of the package that we are trying to " - "find the dependencies for. Set the `--no-remap` " - "cli flag if you want to disable this.".format(pkg)) + if pkg == current_package_name: + logger.debug( + "Ignoring {} from the list of imports. It is " + "the name of the package that we are trying to " + "find the dependencies for. Set the `--no-remap` " + "cli flag if you want to disable this.".format(pkg) + ) continue - pkg_to_add = pkg_data['_PACKAGE_MAPPING'].get(pkg, pkg) + pkg_to_add = pkg_data["_PACKAGE_MAPPING"].get(pkg, pkg) if pkg != pkg_to_add: logger.debug("Renaming {} to {}".format(pkg, pkg_to_add)) new_deps_dict[k].add(pkg_to_add) - new_deps_dict = {k: sorted(list(v)) for k, v in new_deps_dict.items() if v} + new_deps_dict = {k: v for k, v in new_deps_dict.items() if v} return new_deps_dict -def simple_import_search_conda_forge_import_map(path_to_source_code, builtins=None, ignore=None, custom_namespaces=None): +def simple_import_search_conda_forge_import_map( + path_to_source_code: str, + builtins: Iterable[str] = _builtin_modules, + ignore: Iterable[str] = (), + custom_namespaces: Iterable[str] = (), +): """Return all conda-forge packages used in all .py files in `path_to_source_code` Parameters @@ -209,7 +246,7 @@ def simple_import_search_conda_forge_import_map(path_to_source_code, builtins=No ignore : list, optional String pattern which if matched causes the file to not be inspected custom_namespaces : list of str or None - If not None, then resulting package outputs will list everying under these + If not None, then resulting package outputs will list everything under these namespaces (e.g., for packages foo.bar and foo.baz, the outputs are foo.bar and foo.baz instead of foo if custom_namespaces=["foo"]). @@ -217,7 +254,7 @@ def simple_import_search_conda_forge_import_map(path_to_source_code, builtins=No ------- dict The list of all imported modules, sorted according to the keys listed - in the docstring of depfinder.ImportCatcher.describe() + in the docstring of depfinder.ImportFinder.describe() Examples -------- @@ -239,24 +276,40 @@ def simple_import_search_conda_forge_import_map(path_to_source_code, builtins=No 'test_with_code']} """ # run depfinder on source code - if ignore is None: - ignore = [] - total_imports_list = [] - for _, _, c in iterate_over_library(path_to_source_code, custom_namespaces=custom_namespaces): - total_imports_list.append(c.total_imports) - total_imports = defaultdict(dict) + class FoundImports(BaseModel): + + import_metadata_type = Dict[str, Set[ImportMetadata]] + total_imports_list: List[import_metadata_type] = [] + for _, _, import_finder in iterate_over_library( + path_to_source_code, custom_namespaces=custom_namespaces + ): + total_imports_list.append(import_finder.total_imports) + + total_imports: import_metadata_type = defaultdict(dict) + for total_import in total_imports_list: - for name, md in total_import.items(): - total_imports[name].update(md) + for import_name, md in total_import.items(): + total_imports[import_name].update(md) from .reports import report_conda_forge_names_from_import_map + imports, _, _ = report_conda_forge_names_from_import_map( total_imports, builtin_modules=builtins, ignore=ignore ) return {k: sorted(list(v)) for k, v in imports.items()} -def simple_import_to_pkg_map(path_to_source_code, builtins=None, ignore=None, custom_namespaces=None): - """Provide the map beteen all the imports and their possible packages +class TotalImports(BaseModel): + import_name: str + metadata: Dict[str, Any] + + +def simple_import_to_pkg_map( + path_to_source_code, + builtins=_builtin_modules, + ignore=None, + custom_namespaces=None +): + """Provide the map between all the imports and their possible packages Parameters ---------- @@ -281,13 +334,16 @@ def simple_import_to_pkg_map(path_to_source_code, builtins=None, ignore=None, cu if ignore is None: ignore = [] total_imports_list = [] - for _, _, c in iterate_over_library(path_to_source_code, custom_namespaces=custom_namespaces): + for _, _, c in iterate_over_library( + path_to_source_code, custom_namespaces=custom_namespaces + ): total_imports_list.append(c.total_imports) total_imports = defaultdict(dict) for total_import in total_imports_list: for name, md in total_import.items(): total_imports[name].update(md) from .reports import report_conda_forge_names_from_import_map + _, _, import_to_artifact = report_conda_forge_names_from_import_map( total_imports, builtin_modules=builtins, ignore=ignore ) diff --git a/depfinder/reports.py b/depfinder/reports.py index bc748f2..4aadfbe 100644 --- a/depfinder/reports.py +++ b/depfinder/reports.py @@ -35,43 +35,87 @@ from concurrent.futures.thread import ThreadPoolExecutor from fnmatch import fnmatch from functools import lru_cache +from typing import Dict, Iterable, List, Set, Tuple +from pydantic import BaseModel +import pydantic +from conda_forge_metadata.autotick_bot.import_to_pkg import map_import_to_package import requests from .stdliblist import builtin_modules as _builtin_modules -from .utils import SKETCHY_TYPES_TABLE +from .utils import ImportMetadata, ast_types_to_str -logger = logging.getLogger('depfinder') +logger = logging.getLogger("depfinder") @lru_cache() def _import_map_num_letters(): + """ + Conda-forge, in libcfgraph, has a ton of json files that are grouped by the first N letters of the import name. the value of N is maintained in the file "import_maps_meta.json. so we download that from libcfgraph and use it to determine how many letters to use in the import map file name we request from libcfgraph. + """ req = requests.get( - 'https://raw.githubusercontent.com/regro/libcfgraph/master' - '/import_maps_meta.json') + "https://raw.githubusercontent.com/regro/libcfgraph/master/import_maps_meta.json" + ) req.raise_for_status() - return int(req.json()['num_letters']) + return int(req.json()["num_letters"]) @lru_cache() -def _import_map_cache(import_first_letters): +def _import_map_cache(import_first_letters) -> Dict[str, Set[str]]: + """Get the packages that supply the import provided""" req = requests.get( - f'https://raw.githubusercontent.com/regro/libcfgraph' - f'/master/import_maps/{import_first_letters.lower()}.json') + f"https://raw.githubusercontent.com/regro/libcfgraph" + f"/master/import_maps/{import_first_letters.lower()}.json" + ) if not req.ok: - print('Request to {req_url} failed'.format(req_url=req.url)) + print("Request to {req_url} failed".format(req_url=req.url)) return {} - return {k: set(v['elements']) for k, v in req.json().items()} + return {k: set(v["elements"]) for k, v in req.json().items()} + +class FileListingMeta(BaseModel): + n_files: int +@lru_cache() +def _artifact_to_pkg_cache() -> Dict[str,str]: + file_listing_meta_raw = requests.get( + "https://raw.githubusercontent.com/regro/libcfgraph/master/.file_listing_meta.json" + ).text + file_listing_meta = FileListingMeta.parse_raw(file_listing_meta_raw) + files_from_libcfgraph = (f'.file_listing_{i}.json' for i in range(file_listing_meta.n_files)) + artifact_to_pkg_name: Dict[str, str] = {} + for filename in files_from_libcfgraph: + file_listing: List[str] = requests.get( + f"https://raw.githubusercontent.com/regro/libcfgraph/master/{filename}" + ).json() + # Each row looks something like: + # "artifacts/cryptography/conda-forge/osx-64/cryptography-39.0.0-py38ha6c3189_0.json", + # ARTIFACT_TO_PKG ultimately results in something like + # cryptography-39.0.0-py38ha6c3189_0: cryptography + for entry in file_listing: + artifacts, pkg_name, channel, platform, build_string = entry.split("/") + build_string, suffix = build_string.rsplit('.', maxsplit=1) + # TODO: upstream this to libcfgraph so we just request it, so we reduce bandwidth requirements + artifact_to_pkg_name[build_string] = pkg_name + + return artifact_to_pkg_name -FILE_LISTING = requests.get('https://raw.githubusercontent.com/regro/libcfgraph/master/.file_listing.json').json() -# TODO: upstream this to libcfgraph so we just request it, so we reduce bandwidth requirements -ARTIFACT_TO_PKG = {v.split('/')[-1].rsplit('.', 1)[0]: v.split('/')[1] for v in FILE_LISTING if 'artifacts' in v} +# hubs_auth is a ranked list of the conda-forge packages that are the hubs_auths = requests.get( - 'https://raw.githubusercontent.com/regro/cf-graph-countyfair/master/ranked_hubs_authorities.json').json() + "https://raw.githubusercontent.com/regro/cf-graph-countyfair/master/ranked_hubs_authorities.json" +).json() + +class PackageExtraction(BaseModel): + import_name: str + inferred_import_name: str + supplying_artifacts: Set[str] + supplying_pkgs: Set[str] + most_likely_pkg: str -def extract_pkg_from_import(name): + +def extract_pkg_from_import( + name: str, +) -> PackageExtraction: """Provide the name of the package that matches with the import provided, with the maps between the imports and artifacts and packages that matches @@ -88,23 +132,46 @@ def extract_pkg_from_import(name): original_name = name while True: try: - fllt = name[:min(len(name), num_letters)] + fllt = name[: min(len(name), num_letters)] import_map = _import_map_cache(fllt) supplying_artifacts = import_map[name] except KeyError: - if '.' not in name: - return original_name, {}, {} - name = name.rsplit('.', 1)[0] + if "." not in name: + return PackageExtraction( + import_name=original_name, + inferred_import_name=name, + supplying_artifacts=set(), + supplying_pkgs=set(), + most_likely_pkg=original_name, + ) + name = name.rsplit(".", 1)[0] pass else: break - import_to_artifact = {name: supplying_artifacts} + + # import_to_artifact = {name: supplying_artifacts} # TODO: launder supplying_pkgs through centrality scoring so we have one thing # but keep the rest for the more detailed reports - supplying_pkgs = {ARTIFACT_TO_PKG[k] for k in supplying_artifacts} - import_to_pkg = {name: supplying_pkgs} + artifact_to_pkg = _artifact_to_pkg_cache() + supplying_pkgs = {artifact_to_pkg[k] for k in supplying_artifacts} + # import_to_pkg = {name: supplying_pkgs} + # get the "most authoritative" provider of the import from the sorted hubs_auth list from conda-forge + ret = next(iter(k for k in hubs_auths if k in supplying_pkgs), original_name) + + report = PackageExtraction( + import_name=original_name, + inferred_import_name=name, + supplying_artifacts=supplying_artifacts, + supplying_pkgs=supplying_pkgs, + most_likely_pkg=ret, + ) - return next(iter(k for k in hubs_auths if k in supplying_pkgs), original_name), import_to_artifact, import_to_pkg + # return ( + # ret, + # import_to_artifact, + # import_to_pkg, + # ) + return report def recursively_search_for_name(name, module_names): @@ -112,35 +179,64 @@ def recursively_search_for_name(name, module_names): if name in module_names: return name else: - if '.' in name: - name = name.rsplit('.', 1)[0] + if "." in name: + name = name.rsplit(".", 1)[0] else: return False +class Report(BaseModel): + required: Set[str] = set() + questionable: Set[str] = set() + builtin: Set[str] = set() + questionable_no_match: Set[str] = set() + required_no_match: Set[str] = set() -def report_conda_forge_names_from_import_map(total_imports, builtin_modules=None, ignore=None): - if ignore is None: - ignore = [] - if builtin_modules is None: - builtin_modules = _builtin_modules - report_keys = ['required', 'questionable', 'builtin', 'questionable no match', 'required no match'] - report = {k: set() for k in report_keys} + +def report_conda_forge_names_from_import_map( + total_imports: Dict[str, Dict[Tuple[str, int], ImportMetadata]], + builtin_modules: Iterable[str] = _builtin_modules, + ignore: Iterable[str] = (), +): + report_keys = [ + "required", + "questionable", + "builtin", + "questionable no match", + "required no match", + ] + report: Dict[str, Set[str]] = {k: set() for k in report_keys} import_to_pkg = {k: {} for k in report_keys} import_to_artifact = {k: {} for k in report_keys} futures = {} + logger.debug(f'{builtin_modules=}') + with ThreadPoolExecutor() as pool: for name, md in total_imports.items(): - if all([any(fnmatch(filename, ignore_element) for ignore_element in ignore) for filename, _ in md]): + if all( + [ + any(fnmatch(filename, ignore_element) for ignore_element in ignore) + for filename, _ in md + ] + ): continue elif recursively_search_for_name(name, builtin_modules): - report['builtin'].add(name) + report["builtin"].add(name) + logger.debug(f"{name} is builtin") continue - future = pool.submit(extract_pkg_from_import, name) + + logger.debug(f"Mapping {name} to `map_import_to_package`") + future = pool.submit(map_import_to_package, name) futures[future] = md for future in as_completed(futures): md = futures[future] - most_likely_pkg, _import_to_artifact, _import_to_pkg = future.result() + # most_likely_pkg, _import_to_artifact, _import_to_pkg = future.result() + extraction: PackageExtraction = future.result() + most_likely_pkg = extraction.most_likely_pkg + _import_to_artifact = { + extraction.inferred_import_name: extraction.supplying_artifacts + } + _import_to_pkg = {extraction.inferred_import_name: extraction.supplying_pkgs} for (filename, lineno), import_metadata in md.items(): # Make certain to throw out imports, since an import can happen multiple times @@ -148,18 +244,20 @@ def report_conda_forge_names_from_import_map(total_imports, builtin_modules=None # but is questionable for a regular file if any(fnmatch(filename, ignore_element) for ignore_element in ignore): continue - if any(import_metadata.get(v, False) for v in SKETCHY_TYPES_TABLE.values()): + if any( + import_metadata.__getattribute__(v) for v in ast_types_to_str.values() + ): # if we couldn't find any artifacts to represent this then it doesn't exist in our maps if not _import_to_artifact: - report_key = 'questionable no match' + report_key = "questionable no match" else: - report_key = 'questionable' + report_key = "questionable" else: # if we couldn't find any artifacts to represent this then it doesn't exist in our maps if not _import_to_artifact: - report_key = 'required no match' + report_key = "required no match" else: - report_key = 'required' + report_key = "required" report[report_key].add(most_likely_pkg) import_to_pkg[report_key].update(_import_to_pkg) diff --git a/depfinder/stdliblist.py b/depfinder/stdliblist.py index bfd4b88..95f04d6 100644 --- a/depfinder/stdliblist.py +++ b/depfinder/stdliblist.py @@ -1,17 +1,20 @@ import sys import logging -logger = logging.getLogger('depfinder') +logger = logging.getLogger("depfinder") MAJOR, MINOR = sys.version_info.major, sys.version_info.minor if MAJOR == 3 and MINOR >= 10: - builtin_modules = list(set(list(sys.stdlib_module_names) + list(sys.builtin_module_names))) + builtin_modules: tuple[str] = tuple( + set(list(sys.stdlib_module_names) + list(sys.builtin_module_names)) + ) else: try: from stdlib_list import stdlib_list - pyver = '%s.%s' % (MAJOR, MINOR) - builtin_modules = stdlib_list(pyver) + + pyver = "%s.%s" % (MAJOR, MINOR) + builtin_modules = tuple(stdlib_list(pyver)) del pyver except ImportError: - logger.exception('stdlib-list required for python <= 3.9') - raise \ No newline at end of file + logger.exception("stdlib-list required for python <= 3.9") + raise diff --git a/depfinder/utils.py b/depfinder/utils.py index e582c3c..1588499 100644 --- a/depfinder/utils.py +++ b/depfinder/utils.py @@ -1,35 +1,66 @@ from __future__ import print_function, division, absolute_import import ast -import logging +from enum import Enum +import enum import pkgutil -import sys +from typing import Any, Dict, List, Set, Tuple, Type, Union +from pydantic import BaseModel, create_model import requests import yaml from .stdliblist import builtin_modules -SKETCHY_TYPES_TABLE = {} +class ImportType(Enum): + import_normal = ast.Import + import_from = ast.ImportFrom + unset = enum.auto() + + +class ImportMetadata(BaseModel): + # this group of vars are set to true when an import occurs within + # one of their node types + ast_try = False + ast_match_case = False + ast_function_def = False + ast_async_function_def = False + ast_if = False + ast_while = False + ast_for = False + ast_async_for = False + exact_line = "" + import_type = ImportType.unset + imported_modules: Set[str] = set() + lineno = -1 + filename = "unset" + + +ast_type = Type[ast.AST] +ast_import_types = Union[ast.Import, ast.ImportFrom] +ast_try: List[ast_type] +ast_match: List[ast_type] +ast_types_to_str: Dict[ast_type, str] = {} try: # python 3 - AST_TRY = [ast.Try] - SKETCHY_TYPES_TABLE[ast.Try] = 'try' + ast_try = [ast.Try] + ast_types_to_str[ast.Try] = "ast_try" except AttributeError: # python 2.7 - AST_TRY = [ast.TryExcept, ast.TryFinally] - SKETCHY_TYPES_TABLE[ast.TryExcept] = 'try' - SKETCHY_TYPES_TABLE[ast.TryFinally] = 'try' + # honestly could probably drop this section soon + ast_try = [ast.TryExcept, ast.TryFinally] # type: ignore + ast_types_to_str[ast.TryExcept] = "ast_try" # type: ignore + ast_types_to_str[ast.TryFinally] = "ast_try" # type: ignore try: # python 3.10+ - AST_MATCH = [ast.match_case] - SKETCHY_TYPES_TABLE[ast.match_case] = 'match' + ast_match = [ast.match_case] # type: ignore + ast_types_to_str[ast.match_case] = "ast_match_case" # type: ignore except AttributeError: # match/case does not exist before 3.10 - AST_MATCH = [] + ast_match = [] # this AST_QUESTIONABLE list comprises the various ways an import can be weird @@ -38,22 +69,32 @@ # 3. part of an if/elif/else # 4. inside a loop # 5. (for Python 3.10+) inside a match/case -AST_QUESTIONABLE = tuple(AST_TRY + AST_MATCH + [ - ast.FunctionDef, - ast.AsyncFunctionDef, - ast.If, - ast.While, - ast.For, - ast.AsyncFor, -]) -SKETCHY_TYPES_TABLE[ast.FunctionDef] = 'function' -SKETCHY_TYPES_TABLE[ast.AsyncFunctionDef] = 'async-function' -SKETCHY_TYPES_TABLE[ast.If] = 'if' -SKETCHY_TYPES_TABLE[ast.While] = 'while' -SKETCHY_TYPES_TABLE[ast.For] = 'for' -SKETCHY_TYPES_TABLE[ast.AsyncFor] = 'async-for' -del AST_TRY -del AST_MATCH +AST_QUESTIONABLE: Tuple[ast_type] = tuple( + ast_try + + ast_match + + [ + ast.FunctionDef, + ast.AsyncFunctionDef, + ast.If, + ast.While, + ast.For, + ast.AsyncFor, + ] +) +ast_types_to_str.update( + { + ast.FunctionDef: "ast_function_def", + ast.AsyncFunctionDef: "ast_async_function_def", + ast.If: "ast_if", + ast.While: "ast_while", + ast.For: "ast_for", + ast.AsyncFor: "ast_async_for", + } +) + + +del ast_try +del ast_match try: # Try and use the C extensions because they're faster @@ -64,17 +105,21 @@ yaml_loader = yaml.SafeLoader pkg_data = yaml.load( - pkgutil.get_data(__name__, 'pkg_data/pkg_data.yml').decode(), + pkgutil.get_data(__name__, "pkg_data/pkg_data.yml").decode(), Loader=yaml_loader, ) -req = requests.get('https://raw.githubusercontent.com/regro/cf-graph-countyfair/master/mappings/pypi/name_mapping.yaml') +req = requests.get( + "https://raw.githubusercontent.com/regro/cf-graph-countyfair/master/mappings/pypi/name_mapping.json" +) if req.status_code == 200: - mapping_list = yaml.load(req.text, Loader=yaml_loader) + mapping_list = req.json() else: mapping_list = yaml.load( - pkgutil.get_data(__name__, 'pkg_data/name_mapping.yml').decode(), + pkgutil.get_data(__name__, "pkg_data/name_mapping.yml").decode(), Loader=yaml_loader, ) -namespace_packages = {pkg['import_name'] for pkg in mapping_list if '.' in pkg['import_name']} +namespace_packages = { + pkg["import_name"] for pkg in mapping_list if "." in pkg["import_name"] +} diff --git a/requirements.txt b/requirements.txt index 30e1887..47d8185 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,5 @@ pyyaml stdlib-list; python_version < "3.10" -requests \ No newline at end of file +requests +pydantic +conda-forge-metadata diff --git a/setup.cfg b/setup.cfg index 3587cb7..67a9731 100644 --- a/setup.cfg +++ b/setup.cfg @@ -20,6 +20,7 @@ install_requires = pyyaml stdlib-list; python_version < "3.10" requests + pydantic python_requires = >=2.7 packages = find: diff --git a/test.py b/test.py index 1024942..8cdf0f2 100644 --- a/test.py +++ b/test.py @@ -1,5 +1,4 @@ -from __future__ import (unicode_literals, print_function, division, - absolute_import) +from __future__ import unicode_literals, print_function, division, absolute_import import contextlib import itertools @@ -17,9 +16,16 @@ import depfinder from depfinder import cli, main, inspection, parse_file -from depfinder.main import simple_import_search_conda_forge_import_map, simple_import_to_pkg_map -from depfinder.reports import report_conda_forge_names_from_import_map, extract_pkg_from_import, \ - recursively_search_for_name, _builtin_modules +from depfinder.main import ( + simple_import_search_conda_forge_import_map, + simple_import_to_pkg_map, +) +from depfinder.reports import ( + report_conda_forge_names_from_import_map, + extract_pkg_from_import, + recursively_search_for_name, + _builtin_modules, +) random.seed(12345) @@ -31,9 +37,9 @@ complex_imports = [ - {'targets': - {'questionable': ['atom', 'chemist', 'molecule', 'physicist']}, - 'code': """ + { + "targets": {"questionable": ["atom", "chemist", "molecule", "physicist"]}, + "code": """ try: import molecule except ImportError: @@ -41,10 +47,11 @@ else: import chemist finally: - import physicist""" - }, - {'targets': {'required': ['foo'], 'builtin': ['os']}, - 'code': """ + import physicist""", + }, + { + "targets": {"required": ["foo"], "builtin": ["os"]}, + "code": """ import foo try: import os @@ -132,60 +139,56 @@ async def async_function(): elif that: import harpo else: - import chico""" - }, + import chico""", + }, ] simple_imports = [ - {'targets': {'required': ['foo']}, - 'code': 'import foo'}, - {'targets': {'required': ['bar', 'foo']}, - 'code': 'import foo, bar'}, - {'targets': {'required': ['numpy']}, - 'code': 'import numpy'}, - {'targets': {'required': ['matplotlib']}, - 'code': 'from matplotlib import pyplot'}, + {"targets": {"required": ["foo"]}, "code": "import foo"}, + {"targets": {"required": ["bar", "foo"]}, "code": "import foo, bar"}, + {"targets": {"required": ["numpy"]}, "code": "import numpy"}, + {"targets": {"required": ["matplotlib"]}, "code": "from matplotlib import pyplot"}, # Hit the fake packages code block in main.sanitize_deps() - {'targets': {'required': ['numpy']}, - 'code': 'from numpy import warnings as npwarn'}, + { + "targets": {"required": ["numpy"]}, + "code": "from numpy import warnings as npwarn", + }, ] relative_imports = [ - {'targets': {}, - 'code': 'from . import bar'}, - {'targets': {'relative': ['bar']}, - 'code': 'from .bar import baz'}, - {'targets': {'relative': ['bar']}, - 'code': 'from ..bar import baz'}, + {"targets": {}, "code": "from . import bar"}, + {"targets": {"relative": ["bar"]}, "code": "from .bar import baz"}, + {"targets": {"relative": ["bar"]}, "code": "from ..bar import baz"}, ] -@pytest.fixture(scope='module') + +@pytest.fixture(scope="module") def using_stdlib_list(): try: import stdlib_list + return True except ImportError: return False + def test_nested_namespace_builtins(using_stdlib_list): if using_stdlib_list: - expected = {'builtin': ['concurrent.futures']} + expected = {"builtin": ["concurrent.futures"]} else: - expected = {'builtin': ['concurrent']} - code = 'import concurrent.futures' + expected = {"builtin": ["concurrent"]} + code = "import concurrent.futures" - - - test_object = Initter({'targets': expected, 'code': code}) + test_object = Initter({"targets": expected, "code": code}) imports = main.get_imported_libs(test_object.code) assert imports.describe() == test_object.targets class Initter(object): def __init__(self, artifact): - targets = artifact.get('targets', {}) + targets = artifact.get("targets", {}) self.targets = {k: set(v) for k, v in targets.items()} - self.code = artifact['code'] + self.code = artifact["code"] def test_imports(): @@ -206,22 +209,23 @@ def test_for_smoke(): """Do not validate the output of the functions, just make sure that calling them does not make depfinder blow up """ - deps = list(main.iterate_over_library('.')) + deps = list(main.iterate_over_library(".")) assert deps is not None assert str(deps) is not None assert repr(deps) is not None # hit the simple api - assert main.simple_import_search('.') is not None + assert main.simple_import_search(".") is not None ### NOTEBOOK TESTING CODE ### + @contextlib.contextmanager def write_notebook(cells): nb = v4.new_notebook() - nb['cells'] = [v4.new_code_cell(code_cell) for code_cell in cells] - fname = tempfile.NamedTemporaryFile(suffix='.ipynb').name - with open(fname, 'w') as f: + nb["cells"] = [v4.new_code_cell(code_cell) for code_cell in cells] + fname = tempfile.NamedTemporaryFile(suffix=".ipynb").name + with open(fname, "w") as f: f.write(v4.writes(nb)) try: @@ -234,18 +238,19 @@ def test_notebook_remapping(): code = "import mpl_toolkits" with write_notebook([code]) as fname: deps = main.notebook_path_to_dependencies(fname, remap=False) - assert {'required': ['mpl_toolkits']} == deps - assert {} == main.notebook_path_to_dependencies(fname) + assert deps == {"required": {"mpl_toolkits"}} + deps_remap = main.notebook_path_to_dependencies(fname, remap=True) + assert deps_remap == {} -@pytest.mark.parametrize("import_list_dict", [complex_imports, - simple_imports, - relative_imports]) +@pytest.mark.parametrize( + "import_list_dict", [complex_imports, simple_imports, relative_imports] +) def tester(import_list_dict, capsys): # http://nbviewer.ipython.org/gist/fperez/9716279 for import_dict in import_list_dict: - cell_code = [import_dict['code']] - target = import_dict['targets'] + cell_code = [import_dict["code"]] + target = import_dict["targets"] with write_notebook(cell_code) as fname: # parse the notebook! assert set(target) == set(main.notebook_path_to_dependencies(fname)) @@ -261,16 +266,18 @@ def test_multiple_code_cells(capsys): # http://nbviewer.ipython.org/gist/fperez/9716279 code_for_cells = [] for import_dict in import_list_dict: - code_for_cells.append(import_dict['code']) - target = import_dict['targets'] + code_for_cells.append(import_dict["code"]) + target = import_dict["targets"] for k, v in target.items(): targets[k].update(set(v)) - # turn targets into a dict of sorted lists - targets = {k: sorted(list(v)) for k, v in targets.items()} with write_notebook(code_for_cells) as fname: # parse the notebook! - assert targets == main.notebook_path_to_dependencies(fname) + deps = main.notebook_path_to_dependencies(fname) + assert targets["required"] == deps["required"] + assert targets["questionable"] == deps["questionable"] + assert targets["builtin"] == deps["builtin"] + assert targets["relative"] == deps["relative"] # check the notebook cli _run_cli(path_to_check=fname) stdout, stderr = capsys.readouterr() @@ -279,6 +286,7 @@ def test_multiple_code_cells(capsys): ### CLI TESTING CODE ### + def _process_args(path_to_check, extra_flags): """ Parameters @@ -302,10 +310,10 @@ def _process_args(path_to_check, extra_flags): def _subprocess_cli(path_to_check=None, extra_flags=None): path_to_check, extra_flags = _process_args(path_to_check, extra_flags) p = subprocess.Popen( - ['depfinder', path_to_check] + extra_flags, + ["depfinder", path_to_check] + extra_flags, env=dict(os.environ), stderr=subprocess.PIPE, - stdout=subprocess.PIPE + stdout=subprocess.PIPE, ) stdout, stderr = p.communicate() @@ -318,7 +326,7 @@ def _run_cli(path_to_check=None, extra_flags=None): Helper function to run depfinder in its cli mode """ path_to_check, extra_flags = _process_args(path_to_check, extra_flags) - sys.argv = ['depfinder', path_to_check] + extra_flags + sys.argv = ["depfinder", path_to_check] + extra_flags cli.cli() return None @@ -331,13 +339,12 @@ def known_flags(): flags = [flag for flag in flags if flag] # now flatten the nested list flags = [flag for flag_twins in flags for flag in flag_twins] - flags.remove('-k') - flags.remove('--key') - flags.remove('--pdb') - flags.remove('--ignore') - flags.remove('--custom-namespaces') - flags.extend(['-k all', '-k required', '-k optional', '-k builtin', - '-k relative']) + flags.remove("-k") + flags.remove("--key") + flags.remove("--pdb") + flags.remove("--ignore") + flags.remove("--custom-namespaces") + flags.extend(["-k all", "-k required", "-k optional", "-k builtin", "-k relative"]) return flags @@ -347,10 +354,10 @@ def flags(): @pytest.mark.parametrize( - 'flags', + "flags", itertools.chain( [random.sample(known_flags(), i) for i in range(1, len(known_flags()))] - ) + ), ) def test_cli_with_random_flags(flags): """ @@ -366,8 +373,8 @@ def test_cli_with_random_flags(flags): # The only thing that I know of that will exit with a nonzero status # is if you try to combine quiet mode and verbose mode. This handles # that case - quiet = {'-q', '--quiet'} - verbose = {'-v', '--verbose'} + quiet = {"-q", "--quiet"} + verbose = {"-v", "--verbose"} flags = set(flags) assert flags & quiet != set() and flags & verbose != set() return @@ -376,9 +383,11 @@ def test_cli_with_random_flags(flags): @pytest.mark.parametrize( - 'path, req', - ((dirname(depfinder.__file__), None), - (join(dirname(depfinder.__file__), 'main.py'), set())) + "path, req", + ( + (dirname(depfinder.__file__), None), + (join(dirname(depfinder.__file__), "main.py"), set(("pydantic",))), + ), ) def test_cli(path, req, capsys): """ @@ -387,26 +396,34 @@ def test_cli(path, req, capsys): """ main.PACKAGE_NAME = None old_argv = sys.argv - sys.argv = ['depfinder'] + sys.argv = ["depfinder"] _run_cli(path_to_check=path) sys.argv = old_argv # read stdout and stderr with pytest's built-in capturing mechanism stdout, stderr = capsys.readouterr() - print('stdout\n{}'.format(stdout)) - print('stderr\n{}'.format(stderr)) + print("stdout\n{}".format(stdout)) + print("stderr\n{}".format(stderr)) if req is None: - dependencies_file = join(dirname(dirname(depfinder.__file__)), - 'requirements.txt') - dependencies = set([dep for dep in open(dependencies_file, 'r').read().split('\n') if not dep.startswith("stdlib")]) + dependencies_file = join( + dirname(dirname(depfinder.__file__)), "requirements.txt" + ) + dependencies = set( + [ + dep + for dep in open(dependencies_file, "r").read().split("\n") + if not dep.startswith("stdlib") + ] + ) else: dependencies = req - assert dependencies == set(eval(stdout).get('required', set())) + assert dependencies == set(eval(stdout).get("required", set())) def test_known_fail_cli(tmpdir): - tmpfile = os.path.join(str(tmpdir), 'bad_file.txt') + tmpfile = os.path.join(str(tmpdir), "bad_file.txt") import this - with open(tmpfile, 'w') as f: + + with open(tmpfile, "w") as f: f.write("".join([this.d.get(this.c, this.c) for this.c in this.s])) with pytest.raises(RuntimeError): @@ -415,17 +432,15 @@ def test_known_fail_cli(tmpdir): def test_known_fail_cli2(): with pytest.raises(cli.InvalidSelection): - _run_cli(extra_flags=['-q', '-v']) + _run_cli(extra_flags=["-q", "-v"]) @pytest.mark.parametrize( - 'path', - (dirname(depfinder.__file__), - join(dirname(depfinder.__file__), 'main.py')) + "path", (dirname(depfinder.__file__), join(dirname(depfinder.__file__), "main.py")) ) def test_individual_args(path, flags): for flag in flags: - if flag in ['-h', '--help']: + if flag in ["-h", "--help"]: # skip the help messages since they cause the system to exit continue _run_cli(path_to_check=path, extra_flags=[flag]) @@ -435,55 +450,70 @@ def test_individual_args(path, flags): def test_fake_packages(): fake_import = "import mpl_toolkits" imports = main.get_imported_libs(fake_import) - assert imports.describe() == {'required': {'mpl_toolkits'}} + assert imports.describe() == {"required": {"mpl_toolkits"}} assert main.sanitize_deps(imports.describe()) == {} def test_get_top_level_import(): - name = 'this.that.something' + name = "this.that.something" top_level_name = inspection.get_top_level_import_name(name) - assert top_level_name == 'this' + assert top_level_name == "this" - name = 'google.cloud.storage.something' + name = "google.cloud.storage.something" top_level_name = inspection.get_top_level_import_name(name) - assert top_level_name == 'google.cloud.storage' + assert top_level_name == "google.cloud.storage" def test_report_conda_forge_names_from_import_map(): - m, f, c = parse_file(join(dirname(depfinder.__file__), 'utils.py')) - report, import_to_artifact, import_to_pkg = report_conda_forge_names_from_import_map(c.total_imports) - assert report['required'] == {'pyyaml', 'requests'} + m, f, c = parse_file(join(dirname(depfinder.__file__), "utils.py")) + ( + report, + import_to_artifact, + import_to_pkg, + ) = report_conda_forge_names_from_import_map(c.total_imports) + assert report["required"] == {"pyyaml", "requests", "pydantic"} def test_report_conda_forge_names_from_import_map_ignore(): - m, f, c = parse_file(join(dirname(depfinder.__file__), 'inspection.py')) - report, import_to_artifact, import_to_pkg = report_conda_forge_names_from_import_map(c.total_imports, - ignore=['*insp*']) - assert report['required'] == set() + m, f, c = parse_file(join(dirname(depfinder.__file__), "inspection.py")) + ( + report, + import_to_artifact, + import_to_pkg, + ) = report_conda_forge_names_from_import_map(c.total_imports, ignore=["*insp*"]) + assert report["required"] == set() def test_simple_import_search_conda_forge_import_map(): path_to_source = dirname(depfinder.__file__) - expected_result = sorted(list({"pyyaml", "requests"})) + expected_result = sorted(list({"pyyaml", "requests", "pydantic"})) report = simple_import_search_conda_forge_import_map(path_to_source) - assert report['required'] == expected_result + assert report["required"] == expected_result -@pytest.mark.parametrize('import_name, expected_result', [ - ('six.moves', 'six'), - ('win32com.shell', 'pywin32'), - ('win32com', 'pywin32'), - # this comes from cython but doesn't seem to be a real pkg - ('refnanny.hi', 'refnanny.hi') -]) -def test_extract_pkg_from_import_for_complex_imports(import_name, expected_result): - result, _, _ = extract_pkg_from_import(import_name) - assert result == expected_result +@pytest.mark.parametrize( + "import_name, expected_result", + [ + ("six.moves", "six"), + ("win32com.shell", "pywin32"), + ("win32com", "pywin32"), + # this comes from cython but doesn't seem to be a real pkg + ("refnanny.hi", "refnanny.hi"), + ], +) +def test_extract_pkg_from_import_for_complex_imports( + import_name: str, expected_result: str +): + extraction = extract_pkg_from_import(import_name) + assert extraction.most_likely_pkg == expected_result -@pytest.mark.parametrize('import_name, expected_result', [ - ('six.moves', False), -]) +@pytest.mark.parametrize( + "import_name, expected_result", + [ + ("six.moves", False), + ], +) def test_search_for_name(import_name, expected_result): builtin_name_maybe = recursively_search_for_name(import_name, _builtin_modules) assert builtin_name_maybe == expected_result @@ -492,16 +522,30 @@ def test_search_for_name(import_name, expected_result): def test_simple_import_to_pkg_map(): path_to_source = dirname(depfinder.__file__) import_to_artifact = simple_import_to_pkg_map(path_to_source) - expected_result = {'builtin': {}, - 'questionable': {'stdlib_list': {'stdlib-list'}, 'IPython.core.inputsplitter': {'ipython', 'autovizwidget'}}, - 'questionable no match': {}, - 'required': {'requests': {'apache-libcloud', - 'arm_pyart', - 'autovizwidget', - 'dbxfs', - 'google-api-core', - 'google-cloud-bigquery-storage-core', - 'requests'}, - 'yaml': {'google-cloud-bigquery-storage-core', 'pyyaml'}}, - 'required no match': {}} - assert import_to_artifact == expected_result \ No newline at end of file + expected_result = { + "builtin": {}, + "questionable": { + "stdlib_list": {"stdlib-list"}, + "IPython.core.inputsplitter": {"ipython", "autovizwidget"}, + }, + "questionable no match": {}, + "required": { + "pydantic": { + "compass-interface-http", + "compass-interface-pandas", + "pydantic", + }, + "requests": { + "apache-libcloud", + "arm_pyart", + "autovizwidget", + "dbxfs", + "google-api-core", + "google-cloud-bigquery-storage-core", + "requests", + }, + "yaml": {"google-cloud-bigquery-storage-core", "pyyaml"}, + }, + "required no match": {}, + } + assert import_to_artifact == expected_result