handle duplicate entry points

orsinium-labs · Feb 9, 2021 · a6d3d57 · a6d3d57
1 parent f813bcb
commit a6d3d57
Show file tree

Hide file tree

Showing 2 changed files with 27 additions and 20 deletions.
diff --git a/Taskfile.yml b/Taskfile.yml
@@ -79,8 +79,6 @@ tasks:
       - "{{.MYPY_ENV}}/bin/mypy --ignore-missing-imports --allow-redefinition deal/"
 
   pytest:run:
-    sources:
-      - deal/
     deps:
       - task: flit:install
         vars:

diff --git a/flake8_codes/_plugins/_discover.py b/flake8_codes/_plugins/_discover.py
@@ -1,11 +1,19 @@
 # built-in
 import re
 from collections import defaultdict
-from typing import DefaultDict, Iterable, Iterator, List, NamedTuple, Tuple
+from typing import DefaultDict, Dict, Iterable, Iterator, List, NamedTuple, Tuple
+
+from flake8.main.application import Application
+from flake8.plugins.manager import Plugin as Flake8Plugin
 
 # app
 from ._plugin import get_plugin_name
 
+try:
+    import importlib.metadata as importlib_metadata
+except ImportError:
+    import importlib_metadata
+
 
 REX_CODE = re.compile(r'^[A-Z]{1,5}[0-9]{0,5}$')
 
@@ -21,6 +29,7 @@
     'pycodestyle': ('W', 'E'),
     'pylint': ('C', 'E', 'F', 'I', 'R', 'W'),
 }
+NAMESPACE = 'flake8.extension'
 
 
 class Plugin(NamedTuple):
@@ -30,30 +39,30 @@ class Plugin(NamedTuple):
     version: str
 
 
-def get_installed(app) -> Iterator[Plugin]:
+def get_installed(app: Application) -> Iterator[Plugin]:
     plugins_codes: DefaultDict[Tuple[str, str], List[str]]
     plugins_codes = defaultdict(list)
     versions = dict()
-
-    app.initialize([])
     codes: Iterable[str]
 
-    for check_type in ('ast_plugins', 'logical_line_plugins', 'physical_line_plugins'):
-        for plugin in getattr(app.check_plugins, check_type):
-            key = (check_type, get_plugin_name(plugin.to_dictionary()))
-            versions[key[-1]] = plugin.version
+    entry_points = importlib_metadata.entry_points()[NAMESPACE]
+    for entry_point in entry_points:
+        plugin = Flake8Plugin(entry_point.name, entry_point, local=False)
+        check_type = plugin.parameter_names[0]
+        key = (check_type, get_plugin_name(plugin.to_dictionary()))
+        versions[key[-1]] = plugin.version
 
-            # if codes for plugin specified explicitly in ALIASES, use it
-            codes = ALIASES.get(plugin.plugin_name, [])
-            if codes:
-                plugins_codes[key] = list(codes)
-                continue
+        # if codes for plugin specified explicitly in ALIASES, use it
+        codes = ALIASES.get(plugin.plugin_name, [])
+        if codes:
+            plugins_codes[key] = list(codes)
+            continue
 
-            # otherwise get codes from plugin entrypoint
-            code = plugin.name
-            if not REX_CODE.match(code):
-                raise ValueError('Invalid code format: {}'.format(code))
-            plugins_codes[key].append(code)
+        # otherwise get codes from plugin entrypoint
+        code = plugin.name
+        if not REX_CODE.match(code):
+            raise ValueError('Invalid code format: {}'.format(code))
+        plugins_codes[key].append(code)
 
     if 'flake8-docstrings' in versions:
         versions['flake8-docstrings'] = versions['flake8-docstrings'].split(',')[0]