modernize type hints and remove deprecated python support

Author: thrau

plux/build/discovery.py

@@ -32,7 +32,7 @@ def __init__(self, modules: t.Iterable[ModuleType], resolver: PluginSpecResolver
         self.modules = modules
         self.resolver = resolver or PluginSpecResolver()
 
-    def find_plugins(self) -> t.List[PluginSpec]:
+    def find_plugins(self) -> list[PluginSpec]:
         plugins = list()
 
         for module in self.modules:

plux/build/setuptools.py

@@ -64,7 +64,7 @@ class plugins(InfoCommon, setuptools.Command):
 
     description = "Discover plux plugins and store them in .egg_info"
 
-    user_options: t.ClassVar[t.List[t.Tuple[str, str, str]]] = [
+    user_options: t.ClassVar[list[tuple[str, str, str]]] = [
         (
             "exclude=",
             "e",
@@ -368,7 +368,7 @@ def entry_points_from_egg_info(egg_info_dir: str) -> EntryPointDict:
     return entry_points_from_metadata_path(egg_info_dir)
 
 
-def _should_read_existing_egg_info() -> t.Tuple[bool, t.Optional[str]]:
+def _should_read_existing_egg_info() -> tuple[bool, str | None]:
     # we want to read the .egg-info dir only if it exists, and if we are creating the egg_info or
     # installing it with pip install -e (which calls 'setup.py develop')
 
@@ -419,7 +419,7 @@ def _is_local_build_context():
     return False
 
 
-def find_egg_info_dir() -> t.Optional[str]:
+def find_egg_info_dir() -> str | None:
     """
     Heuristic to find the .egg-info dir of the current build context.
     """
@@ -546,8 +546,8 @@ class DistributionPackageFinder(_PackageFinder):
     def __init__(
         self,
         distribution: setuptools.Distribution,
-        exclude: t.Optional[t.Iterable[str]] = None,
-        include: t.Optional[t.Iterable[str]] = None,
+        exclude: t.Iterable[str] | None = None,
+        include: t.Iterable[str] | None = None,
     ):
         self.distribution = distribution
         self.exclude = _Filter(exclude or [])
@@ -596,7 +596,7 @@ class PluginFromPackageFinder(PluginFinder):
     def __init__(self, finder: _PackageFinder):
         self.finder = finder
 
-    def find_plugins(self) -> t.List[PluginSpec]:
+    def find_plugins(self) -> list[PluginSpec]:
         collector = ModuleScanningPluginFinder(self.load_modules())
         return collector.find_plugins()
 

plux/core/entrypoint.py

@@ -11,7 +11,7 @@ class EntryPoint(t.NamedTuple):
     group: str
 
 
-EntryPointDict = t.Dict[str, t.List[str]]
+EntryPointDict = dict[str, list[str]]
 
 
 def discover_entry_points(finder: PluginFinder) -> EntryPointDict:
@@ -24,7 +24,7 @@ def discover_entry_points(finder: PluginFinder) -> EntryPointDict:
     return to_entry_point_dict([spec_to_entry_point(spec) for spec in finder.find_plugins()])
 
 
-def to_entry_point_dict(eps: t.List[EntryPoint]) -> EntryPointDict:
+def to_entry_point_dict(eps: list[EntryPoint]) -> EntryPointDict:
     """
     Convert the list of EntryPoint objects to a dictionary that maps entry point groups to their respective list of
     ``name=value`` entry points. Each pair is represented as a string.

plux/core/plugin.py

@@ -35,7 +35,7 @@ class Plugin(abc.ABC):
 
     namespace: str
     name: str
-    requirements: t.List[str]
+    requirements: list[str]
 
     def should_load(self) -> bool:
         return True
@@ -94,7 +94,7 @@ class PluginFinder(abc.ABC):
     (e.g., using StevedorePluginFinder that finds plugins from entrypoints)
     """
 
-    def find_plugins(self) -> t.List[PluginSpec]:
+    def find_plugins(self) -> list[PluginSpec]:
         raise NotImplementedError  # pragma: no cover
 
 
@@ -148,8 +148,8 @@ def on_load_before(
         self,
         plugin_spec: PluginSpec,
         plugin: Plugin,
-        load_args: t.Union[t.List, t.Tuple],
-        load_kwargs: t.Dict,
+        load_args: list | tuple,
+        load_kwargs: dict,
     ):
         pass
 
@@ -163,7 +163,7 @@ def on_load_exception(self, plugin_spec: PluginSpec, plugin: Plugin, exception:
 class CompositePluginLifecycleListener(PluginLifecycleListener):
     """A PluginLifecycleListener decorator that dispatches to multiple delegates."""
 
-    listeners: t.List[PluginLifecycleListener]
+    listeners: list[PluginLifecycleListener]
 
     def __init__(self, initial: t.Iterable[PluginLifecycleListener] = None):
         self.listeners = list(initial) if initial else []
@@ -210,7 +210,7 @@ class FunctionPlugin(Plugin):
     def __init__(
         self,
         fn: t.Callable,
-        should_load: t.Union[bool, t.Callable[[], bool]] = None,
+        should_load: bool | t.Callable[[], bool] = None,
         load: t.Callable = None,
     ) -> None:
         super().__init__()
@@ -238,7 +238,7 @@ def should_load(self) -> bool:
 def plugin(
     namespace,
     name=None,
-    should_load: t.Union[bool, t.Callable[[], bool]] = None,
+    should_load: bool | t.Callable[[], bool] = None,
     load: t.Callable = None,
 ):
     """

plux/runtime/cache.py

@@ -6,7 +6,6 @@
 import struct
 import sys
 import threading
-import typing as t
 from importlib import metadata
 from pathlib import Path
 
@@ -67,7 +66,7 @@ class EntryPointsCache(EntryPointsResolver):
     _instance: "EntryPointsCache" = None
     _instance_lock: threading.RLock = threading.RLock()
 
-    _cache: t.Dict[t.Tuple[str, ...], t.Dict[str, t.List[metadata.EntryPoint]]]
+    _cache: dict[tuple[str, ...], dict[str, list[metadata.EntryPoint]]]
     """For each specific path (like sys.path), we store a dict of group -> [entry point]."""
 
     def __init__(self):
@@ -76,7 +75,7 @@ def __init__(self):
         self._resolver = MetadataEntryPointsResolver()
         self._cache_dir = get_user_cache_dir() / "plux"
 
-    def get_entry_points(self) -> t.Dict[str, t.List[metadata.EntryPoint]]:
+    def get_entry_points(self) -> dict[str, list[metadata.EntryPoint]]:
         """
         Returns a dictionary of entry points for the current ``sys.path``.
         """
@@ -98,7 +97,7 @@ def get_entry_points(self) -> t.Dict[str, t.List[metadata.EntryPoint]]:
 
             return self._cache[key]
 
-    def _build_and_store_index(self, cache_dir: Path) -> t.Dict[str, t.List[metadata.EntryPoint]]:
+    def _build_and_store_index(self, cache_dir: Path) -> dict[str, list[metadata.EntryPoint]]:
         # first, try and load the index from the file system
         path_key = self._calculate_hash_key(sys.path)
 
@@ -117,7 +116,7 @@ def _build_and_store_index(self, cache_dir: Path) -> t.Dict[str, t.List[metadata
 
         return index
 
-    def _calculate_hash_key(self, path: t.List[str]):
+    def _calculate_hash_key(self, path: list[str]):
         """
         Calculates a hash of all modified times of all ``entry_point.txt`` files in the path. Basic idea
         taken from ``stevedore._cache._hash_settings_for_path``. The main difference to it is that it also

plux/runtime/filter.py

@@ -54,7 +54,7 @@ class MatchingPluginFilter:
     A MatchingPluginFilter can be used to exclude specific plugins from loading.
     """
 
-    exclusions: t.List[PluginSpecMatcher]
+    exclusions: list[PluginSpecMatcher]
 
     def __init__(self):
         self.exclusions = []

plux/runtime/manager.py

@@ -20,7 +20,7 @@
 P = t.TypeVar("P", bound=Plugin)
 
 
-def _call_safe(func: t.Callable, args: t.Tuple, exception_message: str):
+def _call_safe(func: t.Callable, args: tuple, exception_message: str):
     """
     Call the given function with the given arguments, and if it fails, log the given exception_message. If
     logging.DEBUG is set for the logger, then we also log the traceback. An exception is made for any
@@ -47,7 +47,7 @@ class PluginLifecycleNotifierMixin:
     Mixin that provides functions to dispatch calls to a PluginLifecycleListener in a safe way.
     """
 
-    listeners: t.List[PluginLifecycleListener]
+    listeners: list[PluginLifecycleListener]
 
     def _fire_on_resolve_after(self, plugin_spec):
         for listener in self.listeners:
@@ -155,19 +155,19 @@ class PluginManager(PluginLifecycleNotifierMixin, t.Generic[P]):
 
     namespace: str
 
-    load_args: t.Union[t.List, t.Tuple]
-    load_kwargs: t.Dict[str, t.Any]
-    listeners: t.List[PluginLifecycleListener]
-    filters: t.List[PluginFilter]
+    load_args: list | tuple
+    load_kwargs: dict[str, t.Any]
+    listeners: list[PluginLifecycleListener]
+    filters: list[PluginFilter]
 
     def __init__(
         self,
         namespace: str,
-        load_args: t.Union[t.List, t.Tuple] = None,
-        load_kwargs: t.Dict = None,
-        listener: t.Union[PluginLifecycleListener, t.Iterable[PluginLifecycleListener]] = None,
+        load_args: list | tuple = None,
+        load_kwargs: dict = None,
+        listener: PluginLifecycleListener | t.Iterable[PluginLifecycleListener] = None,
         finder: PluginFinder = None,
-        filters: t.List[PluginFilter] = None,
+        filters: list[PluginFilter] = None,
     ):
         """
         Create a new PluginManager.
@@ -236,7 +236,7 @@ def load(self, name: str) -> P:
 
         return container.plugin
 
-    def load_all(self, propagate_exceptions=False) -> t.List[P]:
+    def load_all(self, propagate_exceptions=False) -> list[P]:
         """
         Attempts to load all plugins found in the namespace, and returns those that were loaded successfully. If
         propagate_exception is set to True, then the method will re-raise any errors as soon as it encouters them.
@@ -261,13 +261,13 @@ def load_all(self, propagate_exceptions=False) -> t.List[P]:
 
         return plugins
 
-    def list_plugin_specs(self) -> t.List[PluginSpec]:
+    def list_plugin_specs(self) -> list[PluginSpec]:
         return [container.plugin_spec for container in self._plugins.values()]
 
-    def list_names(self) -> t.List[str]:
+    def list_names(self) -> list[str]:
         return [spec.name for spec in self.list_plugin_specs()]
 
-    def list_containers(self) -> t.List[PluginContainer[P]]:
+    def list_containers(self) -> list[PluginContainer[P]]:
         return list(self._plugins.values())
 
     def get_container(self, name: str) -> PluginContainer[P]:
@@ -280,7 +280,7 @@ def is_loaded(self, name: str) -> bool:
         return self._require_plugin(name).is_loaded
 
     @property
-    def _plugins(self) -> t.Dict[str, PluginContainer[P]]:
+    def _plugins(self) -> dict[str, PluginContainer[P]]:
         if self._plugin_index is None:
             with self._init_mutex:
                 if self._plugin_index is None:
@@ -363,7 +363,7 @@ def _plugin_from_spec(self, plugin_spec: PluginSpec) -> P:
 
         return factory()
 
-    def _init_plugin_index(self) -> t.Dict[str, PluginContainer]:
+    def _init_plugin_index(self) -> dict[str, PluginContainer]:
         return {plugin.name: plugin for plugin in self._import_plugins() if plugin}
 
     def _import_plugins(self) -> t.Iterable[PluginContainer]:

plux/runtime/metadata.py

@@ -6,32 +6,23 @@
 import inspect
 import io
 import os
-import sys
 import typing as t
 from functools import lru_cache
 from importlib import metadata
 
-if sys.version_info >= (3, 10):
-    from importlib.metadata import packages_distributions as metadata_packages_distributions
-else:
-
-    def metadata_packages_distributions() -> t.Mapping[str, t.List[str]]:
-        """Copied from Pyton 3.10 stdlib."""
-        pkg_to_dist = collections.defaultdict(list)
-        for dist in metadata.distributions():
-            for pkg in (dist.read_text("top_level.txt") or "").split():
-                pkg_to_dist[pkg].append(dist.metadata["Name"])
-        return dict(pkg_to_dist)
-
-
 from plux.core.entrypoint import EntryPointDict, to_entry_point_dict
 from plux.core.plugin import PluginSpec
 
 Distribution = metadata.Distribution
 
 
+def metadata_packages_distributions():
+    """Wrapper around ``importlib.metadata.packages_distributions``, which returns a mapping of top-level packages."""
+    return metadata.packages_distributions()
+
+
 @lru_cache()
-def packages_distributions() -> t.Mapping[str, t.List[str]]:
+def packages_distributions() -> t.Mapping[str, list[str]]:
     """
     Cache wrapper around ``metadata.packages_distributions``, which returns a mapping of top-level packages to
     their distributions. This index is created by scanning all ``top_level.txt`` metadata files in the path.
@@ -53,7 +44,7 @@ def packages_distributions() -> t.Mapping[str, t.List[str]]:
     return distributions
 
 
-def resolve_distribution_information(plugin_spec: PluginSpec) -> t.Optional[Distribution]:
+def resolve_distribution_information(plugin_spec: PluginSpec) -> Distribution | None:
     """
     Resolves for a PluginSpec the python distribution package it comes from. Currently, this raises an
     error for plugins that come from a namespace package (i.e., when a package is part of multiple
@@ -83,7 +74,7 @@ def entry_points_from_metadata_path(metadata_path: str) -> EntryPointDict:
 
 def resolve_entry_points(
     distributions: t.Iterable[Distribution] = None,
-) -> t.List[metadata.EntryPoint]:
+) -> list[metadata.EntryPoint]:
     """
     Resolves all entry points using a combination of ``importlib.metadata``, and also follows entry points
     links in ``entry_points_editable.txt`` created by plux while building editable wheels.
@@ -120,7 +111,7 @@ def resolve_entry_points(
 
 def build_entry_point_index(
     entry_points: t.Iterable[metadata.EntryPoint],
-) -> t.Dict[str, t.List[metadata.EntryPoint]]:
+) -> dict[str, list[metadata.EntryPoint]]:
     """
     Organizes the given list of entry points into a dictionary that maps entry point groups to their
     respective entry points, which resembles the data structure of an ``entry_points.txt``.
@@ -138,24 +129,17 @@ def build_entry_point_index(
     return dict(result)
 
 
-if sys.version_info >= (3, 10):
-
-    def parse_entry_points_text(text: str) -> t.List[metadata.EntryPoint]:
-        """
-        Parses the content of an ``entry_points.txt`` into a list of entry point objects.
-
-        :param text: the string to parse
-        :return: a list of metadata EntryPoint objects
-        """
-        return metadata.EntryPoints._from_text(text)
-
-else:
+def parse_entry_points_text(text: str) -> list[metadata.EntryPoint]:
+    """
+    Parses the content of an ``entry_points.txt`` into a list of entry point objects.
 
-    def parse_entry_points_text(text: str) -> t.List[metadata.EntryPoint]:
-        return metadata.EntryPoint._from_text(text)
+    :param text: the string to parse
+    :return: a list of metadata EntryPoint objects
+    """
+    return metadata.EntryPoints._from_text(text)
 
 
-def serialize_entry_points_text(index: t.Dict[str, t.List[metadata.EntryPoint]]) -> str:
+def serialize_entry_points_text(index: dict[str, list[metadata.EntryPoint]]) -> str:
     """
     Serializes an entry point index generated via ``build_entry_point_index`` into a string that can be
     written into an ``entry_point.txt``. Example::
@@ -185,7 +169,7 @@ class EntryPointsResolver:
     Interface for something that builds an entry point index.
     """
 
-    def get_entry_points(self) -> t.Dict[str, t.List[metadata.EntryPoint]]:
+    def get_entry_points(self) -> dict[str, list[metadata.EntryPoint]]:
         raise NotImplementedError
 
 
@@ -194,5 +178,5 @@ class MetadataEntryPointsResolver(EntryPointsResolver):
     Implementation that uses regular ``importlib.metadata`` methods to resolve entry points.
     """
 
-    def get_entry_points(self) -> t.Dict[str, t.List[metadata.EntryPoint]]:
+    def get_entry_points(self) -> dict[str, list[metadata.EntryPoint]]:
         return build_entry_point_index(resolve_entry_points(metadata.distributions()))

plux/runtime/resolve.py

@@ -31,7 +31,7 @@ def __init__(
         self.spec_resolver = spec_resolver or PluginSpecResolver()
         self.entry_points_resolver = entry_points_resolver or EntryPointsCache.instance()
 
-    def find_plugins(self) -> t.List[PluginSpec]:
+    def find_plugins(self) -> list[PluginSpec]:
         specs = []
         finds = self.entry_points_resolver.get_entry_points().get(self.namespace, [])
         for ep in finds: