Crude implementation of tool which helps to analyze Python imports: find unused, find used, etc...

imports.py

import ast
import logging
import operator as op
import signal
import sys

import astor


logging.basicConfig(level=logging.INFO, stream=sys.stdout, format='%(message)s')
_log = logging.getLogger(__name__)

signal.signal(signal.SIGPIPE, signal.SIG_DFL)


def _to_source(node):
    return astor.to_source(node).strip()


class Import(object):
    def __init__(self, node):
        self.node = node

    def lineno(self):
        return self.node.lineno

    def module(self):
        return self.node.names[0].name

    def package(self):
        try:
            top, _ = self.module().split('.', 1)
        except ValueError:
            return self.module()
        else:
            return top

    def is_valid(self):
        return len(self.node.names) == 1

    def __str__(self):
        return _to_source(self.node)


class ImportFrom(object):
    def __init__(self, node):
        self.node = node

    def lineno(self):
        return self.node.lineno

    def module(self):
        return self.node.module

    def last_module(self):
        try:
            _, last = self.module().split('.', 1)
        except ValueError:
            return self.module()
        else:
            return last

    def package(self):
        try:
            top, _ = self.module().split('.', 1)
        except ValueError:
            return self.module()
        else:
            return top

    def is_relative(self):
        return self.node.level > 0

    def is_relative_from_current(self):
        return self.is_relative() and self.module() is None

    def names(self):
        return sorted(map(op.attrgetter('name'), self.node.names))

    def __str__(self):
        return _to_source(self.node)


def main():
    for fn in produce_filenames_from_stdin():
        for imp in extract_imports(produce_ast(fn), [
                ActionableFilter(is_import, Import),
                ActionableFilter(is_import_from, ImportFrom),
            ]):

            if isinstance(imp, ImportFrom) and imp.is_relative():
                continue

            # _log.info('%s:%s:%s', fn, imp.lineno(), imp.package())
            _log.info('%s:%s:%s', fn, imp.lineno(), imp)


def produce_filenames_from_stdin():
    for line in sys.stdin:
        yield line.strip()


def produce_ast(source):
    with open(source) as f:
        return ast.parse(f.read())


class ActionableFilter(object):
    def __init__(self, predicate, operation):
        self._predicate = predicate
        self._operation = operation

    def applicable(self, node):
        return self._predicate(node)

    def do(self, node):
        return self._operation(node)


def is_import(node):
    return isinstance(node, ast.Import)


def is_import_from(node):
    return isinstance(node, ast.ImportFrom)


def extract_imports(tree, filters):
    for node in ast.walk(tree):
        for f in filters:
            if f.applicable(node):
                yield f.do(node)
                
                break


if __name__ == '__main__':
    main()