Refactor function

opentelemetry-api/src/opentelemetry/util/_importlib_metadata.py

@@ -13,15 +13,67 @@
 # limitations under the License.
 
 from sys import version_info
+from typing import Dict, List, Union
+
+# This is a cache to avoid going through creating a dictionary for all entry
+# points for 3.7 every time the entry points function is called.
+result_entry_points_37 = {}
 
 # FIXME remove this when support for 3.7 is dropped.
 if version_info.minor == 7:
     # pylint: disable=import-error
-    from importlib_metadata import (  # type: ignore
-        EntryPoint,
-        entry_points,
-        version,
+    from importlib_metadata import EntryPoint
+    from importlib_metadata import (
+        entry_points as importlib_metadata_entry_points,  # type: ignore
     )
+    from importlib_metadata import version
+
+    def entry_points(group: str = None, name: str = None) -> Union[List, Dict]:
+
+        if group is None and name is None:
+
+            if not result_entry_points_37:
+
+                for entry_point in importlib_metadata_entry_points():
+
+                    if entry_point.group not in result_entry_points_37.keys():
+                        result_entry_points_37[entry_point.group] = []
+
+                    (
+                        result_entry_points_37[entry_point.group].append(
+                            entry_point
+                        )
+                    )
+
+                for key, value in result_entry_points_37.items():
+                    result_entry_points_37[key] = tuple(value)
+
+            return result_entry_points_37
+
+        if group is not None and name is None:
+            return tuple(
+                entry_point
+                for entry_point in importlib_metadata_entry_points(group=group)
+            )
+
+        if group is None and name is not None:
+            return tuple(
+                entry_point
+                for entry_point in importlib_metadata_entry_points(name=name)
+            )
+
+        return tuple(
+            entry_point
+            for entry_point in importlib_metadata_entry_points(
+                group=group, name=name
+            )
+        )
+
+    __all__ = [
+        "entry_points",
+        "EntryPoint",
+        "version",
+    ]
 
 # FIXME remove this file when support for 3.9 is dropped.
 elif version_info.minor in (8, 9):
@@ -32,36 +84,83 @@
     )
     from importlib.metadata import version
 
-    def entry_points(group: str = None, name: str = None):  # type: ignore
+    def entry_points(group: str = None, name: str = None) -> Union[List, Dict]:
+
+        result_key_entry_points = importlib_metadata_entry_points()
 
-        if group is None:
-            return importlib_metadata_entry_points()
+        if group is None and name is None:
+            return result_key_entry_points
 
-        if name is None:
-            return importlib_metadata_entry_points()[group]
+        if group is not None and group not in result_key_entry_points.keys():
+            return tuple()
 
-        for entry_point in importlib_metadata_entry_points()[group]:
-            if entry_point.name == name:
-                return [entry_point]
+        if group is not None and name is None:
+            return result_key_entry_points[group]
+
+        if group is None and name is not None:
+
+            name_entry_points = []
+
+            for result_entry_points in result_key_entry_points.values():
+                for result_entry_point in result_entry_points:
+                    if result_entry_point.name == name:
+                        name_entry_points.append(result_entry_point)
+
+            return tuple(name_entry_points)
+
+        name_group_entry_points = []
+
+        for result_entry_point in result_key_entry_points[group]:
+            if result_entry_point.name == name:
+                name_group_entry_points.append(result_entry_point)
+                break
+
+        return tuple(name_group_entry_points)
 
     __all__ = [
         "entry_points",
+        "EntryPoint",
         "version",
     ]
 
 else:
+    from importlib.metadata import EntryPoint
     from importlib.metadata import (
-        EntryPoint,
-        EntryPoints,
-        SelectableGroups,
-        entry_points,
-        version,
+        entry_points as importlib_metadata_entry_points,
     )
+    from importlib.metadata import version
+
+    def entry_points(group: str = None, name: str = None) -> Union[List, Dict]:
+
+        if group is None and name is None:
+            result_entry_points = {}
+
+            for key, value in importlib_metadata_entry_points().items():
+                result_entry_points[key] = tuple(value)
+
+            return result_entry_points
+
+        if group is not None and name is None:
+            return tuple(
+                entry_point
+                for entry_point in importlib_metadata_entry_points(group=group)
+            )
+
+        if group is None and name is not None:
+            return tuple(
+                entry_point
+                for entry_point in importlib_metadata_entry_points(name=name)
+            )
+
+        return tuple(
+            entry_point
+            for entry_point in importlib_metadata_entry_points(
+                group=group, name=name
+            )
+        )
 
     __all__ = [
         "entry_points",
         "version",
         "EntryPoint",
-        "SelectableGroups",
-        "EntryPoints",
     ]

opentelemetry-api/tests/util/test__importlib_metadata.py

@@ -12,13 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Dict, Tuple
 from unittest import TestCase
-from warnings import catch_warnings, filterwarnings
 
 from opentelemetry.metrics import MeterProvider
 from opentelemetry.util._importlib_metadata import EntryPoint
 from opentelemetry.util._importlib_metadata import (
-    entry_points as entry_points_function,  # SelectableGroups,; EntryPoints
+    entry_points as importlib_metadata_entry_points,
 )
 
 
@@ -28,7 +28,7 @@ def test_entry_points(self):
         self.assertIsInstance(
             next(
                 iter(
-                    entry_points_function(
+                    importlib_metadata_entry_points(
                         group="opentelemetry_meter_provider",
                         name="default_meter_provider",
                     )
@@ -43,52 +43,36 @@ def test_uniform_behavior(self):
         version.
         """
 
-        selectable_groups = entry_points_function()
+        entry_points = importlib_metadata_entry_points()
+        self.assertIsInstance(entry_points, Dict)
 
-        # self.assertIsInstance(selectable_groups, SelectableGroups)
+        entry_points = entry_points["opentelemetry_propagator"]
+        self.assertIsInstance(entry_points, Tuple)
+        self.assertIsInstance(entry_points[0], EntryPoint)
 
-        # Supressing the following warning here:
-        # DeprecationWarning: SelectableGroups dict interface is deprecated. Use select.
-        # The behavior of the importlib metadata library is hard to understand,
-        # this is True: selectable_groups is selectable_groups.select(). So,
-        # using select, as the warning says yields the same problem. Also
-        # select does not accept any parameters.
-
-        with catch_warnings():
-            filterwarnings("ignore", category=DeprecationWarning)
-            entry_points = selectable_groups.select()[
-                "opentelemetry_propagator"
-            ]
-
-        # Supressing the following warning here:
-        # DeprecationWarning: DeprecationWarning: Accessing entry points by index is deprecated. Cast to tuple if needed.
-        # The behavior of the importlib metadata library is hard to understand,
-        # this is True: entry_points == .select(). So, using select, as the
-        # warning says yields the same problem. Also select does not accept any
-        # parameters.
-        with catch_warnings():
-            filterwarnings("ignore", category=DeprecationWarning)
-
-            self.assertIsInstance(entry_points.select()[0], EntryPoint)
-
-            entry_points = entry_points_function(
-                group="opentelemetry_propagator"
-            )
-
-            self.assertIsInstance(entry_points.select()[0], EntryPoint)
-
-            entry_points = entry_points_function(
-                group="opentelemetry_propagator", name="baggage"
-            )
+        entry_points = importlib_metadata_entry_points(
+            group="opentelemetry_propagator"
+        )
+        self.assertIsInstance(entry_points, Tuple)
+        self.assertIsInstance(entry_points[0], EntryPoint)
 
-            self.assertIsInstance(entry_points.select()[0], EntryPoint)
+        entry_points = importlib_metadata_entry_points(
+            group="opentelemetry_propagator", name="baggage"
+        )
+        self.assertIsInstance(entry_points, Tuple)
+        self.assertIsInstance(entry_points[0], EntryPoint)
 
-        entry_points = entry_points_function(group="abc")
+        entry_points = importlib_metadata_entry_points(name="baggage")
+        self.assertIsInstance(entry_points, Tuple)
+        self.assertIsInstance(entry_points[0], EntryPoint)
 
-        self.assertEqual(entry_points, [])
+        entry_points = importlib_metadata_entry_points(group="abc")
+        self.assertEqual(entry_points, tuple())
 
-        entry_points = entry_points_function(
+        entry_points = importlib_metadata_entry_points(
             group="opentelemetry_propagator", name="abc"
         )
+        self.assertEqual(entry_points, tuple())
 
-        self.assertEqual(entry_points, [])
+        entry_points = importlib_metadata_entry_points(group="abc", name="abc")
+        self.assertEqual(entry_points, tuple())