Update all pkg_resources references to use importlib

Automated, slop transformation.

Author: ndevenish

src/dlstbx/cli/run_health_checks.py

@@ -2,9 +2,9 @@
 
 import argparse
 import concurrent.futures
+import importlib.metadata
 import logging
 
-import pkg_resources
 import zocalo.configuration
 
 import dlstbx.health_checks as hc
@@ -15,7 +15,8 @@
 
 def run():
     check_functions = {
-        e.name: e.load for e in pkg_resources.iter_entry_points("zocalo.health_checks")
+        e.name: e.load
+        for e in importlib.metadata.entry_points(group="zocalo.health_checks")
     }
 
     parser = argparse.ArgumentParser(

src/dlstbx/cli/wrap.py

@@ -7,12 +7,12 @@
 
 import argparse
 import faulthandler
+import importlib.metadata
 import json
 import logging
 import signal
 import sys
 
-import pkg_resources
 import workflows.recipe.wrapper
 import workflows.services.common_service
 import workflows.transport
@@ -63,7 +63,7 @@ def run():
     zc.activate()
 
     known_wrappers = {
-        e.name: e.load for e in pkg_resources.iter_entry_points("zocalo.wrappers")
+        e.name: e.load for e in importlib.metadata.entry_points(group="zocalo.wrappers")
     }
 
     # Set up parser

src/dlstbx/mimas/__init__.py

@@ -3,11 +3,11 @@
 import dataclasses
 import enum
 import functools
+import importlib.metadata
 import numbers
 from typing import Any, Callable, List, Optional, Tuple, Union
 
 import gemmi
-import pkg_resources
 import zocalo.configuration
 
 from dlstbx.mimas.specification import BaseSpecification
@@ -411,7 +411,7 @@ def inner_wrapper(
 def _get_handlers() -> dict[str, Callable]:
     return {
         e.name: e.load()
-        for e in pkg_resources.iter_entry_points("zocalo.mimas.handlers")
+        for e in importlib.metadata.entry_points(group="zocalo.mimas.handlers")
     }
 
 

src/dlstbx/requirements.py

@@ -4,14 +4,15 @@
 # command line options are passed through to the conda install process
 from __future__ import annotations
 
+import importlib.metadata
 import json
 import os
 import pathlib
 import subprocess
 import sys
 from typing import Optional
 
-import pkg_resources
+from packaging.requirements import Requirement
 
 import dlstbx
 
@@ -81,17 +82,15 @@ def check():
     conda_environment = {
         package["name"]: package["version"] for package in json.loads(conda_list)
     }
-    requirements = [
-        (spec, pkg_resources.Requirement.parse(spec)) for spec in sorted(conda_required)
-    ]
+    requirements = [(spec, Requirement(spec)) for spec in sorted(conda_required)]
 
     # Now we should have an unduplicated set of requirements
     action_list = []
     for original_spec, requirement in requirements:
         # Check if package is installed in development mode
         try:
-            currentversion = pkg_resources.require(requirement.name)[0].version
-        except Exception:
+            currentversion = importlib.metadata.version(requirement.name)
+        except importlib.metadata.PackageNotFoundError:
             pass
         else:
             location = None
@@ -101,13 +100,13 @@ def check():
                     with open(egg_link) as fh:
                         location = fh.readline().strip()
                         break
-            if location and currentversion in requirement:
+            if location and currentversion in requirement.specifier:
                 print(
                     "requires conda package %s, has %s as developer installation"
                     % (requirement, currentversion)
                 )
                 continue
-            elif location and currentversion not in requirement:
+            elif location and currentversion not in requirement.specifier:
                 _notice(
                     "    WARNING: Can not update package {package} automatically.",
                     "",
@@ -125,7 +124,7 @@ def check():
 
         # Check if package is installed with conda
         if requirement.name in conda_environment:
-            if conda_environment[requirement.name] in requirement:
+            if conda_environment[requirement.name] in requirement.specifier:
                 print(
                     "requires conda package %s, has %s"
                     % (requirement, conda_environment[requirement.name])

src/dlstbx/services/cluster.py

@@ -3,14 +3,14 @@
 import datetime
 import errno
 import getpass
+import importlib.metadata
 import json
 import logging
 import math
 import os
 import pathlib
 from typing import Optional
 
-import pkg_resources
 import pydantic
 import requests
 import workflows.recipe
@@ -159,8 +159,8 @@ def initializing(self):
 
         self.schedulers = {
             f.name: f.load()
-            for f in pkg_resources.iter_entry_points(
-                "zocalo.services.cluster.schedulers"
+            for f in importlib.metadata.entry_points(
+                group="zocalo.services.cluster.schedulers"
             )
         }
         self.log.debug(f"Supported schedulers: {', '.join(self.schedulers.keys())}")

src/dlstbx/services/images.py

@@ -1,14 +1,14 @@
 from __future__ import annotations
 
 import errno
+import importlib.metadata
 import logging
 import os
 import re
 import time
 from typing import Any, Callable, Dict, NamedTuple, Protocol
 
 import PIL.Image
-import pkg_resources
 import procrunner
 import workflows.recipe
 from workflows.services.common_service import CommonService
@@ -53,7 +53,9 @@ def initializing(self):
         self.log.info("Image service starting")
         self.image_functions: dict[str, Callable] = {
             e.name: e.load()
-            for e in pkg_resources.iter_entry_points("zocalo.services.images.plugins")
+            for e in importlib.metadata.entry_points(
+                group="zocalo.services.images.plugins"
+            )
         }
         workflows.recipe.wrap_subscribe(
             self._transport,

tests/conftest.py

@@ -1,11 +1,12 @@
 # pytest configuration file
 from __future__ import annotations
 
+import importlib.metadata
 import os
 from typing import List
+from unittest.mock import patch
 
 import ispyb.sqlalchemy
-import pkg_resources
 import pytest
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
@@ -85,20 +86,23 @@ def handle_i99(scenario: mimas.MimasScenario, **kwargs) -> List[mimas.Invocation
 
 @pytest.fixture
 def with_dummy_plugins():
-    # Get the current distribution and entry map
-    dist = pkg_resources.get_distribution("dlstbx")
-    entry_map = pkg_resources.get_entry_map("dlstbx", group="zocalo.mimas.handlers")
-
-    # Create the fake entry point definitions and add the mapping
-    entry_map["i99"] = pkg_resources.EntryPoint.parse(
-        f"i99 = {__name__}:handle_i99", dist=dist
-    )
-    entry_map["i99_rotation"] = pkg_resources.EntryPoint.parse(
-        f"i99_rotation = {__name__}:handle_i99_rotation", dist=dist
-    )
-    mimas._get_handlers.cache_clear()
-    yield
-    # cleanup
-    del entry_map["i99"]
-    del entry_map["i99_rotation"]
+    real_entry_points = importlib.metadata.entry_points(group="zocalo.mimas.handlers")
+    fake_entry_points = [
+        importlib.metadata.EntryPoint(
+            name="i99", value=f"{__name__}:handle_i99", group="zocalo.mimas.handlers"
+        ),
+        importlib.metadata.EntryPoint(
+            name="i99_rotation",
+            value=f"{__name__}:handle_i99_rotation",
+            group="zocalo.mimas.handlers",
+        ),
+    ]
+    combined = list(real_entry_points) + fake_entry_points
+
+    with patch(
+        "dlstbx.mimas.importlib.metadata.entry_points",
+        return_value=combined,
+    ):
+        mimas._get_handlers.cache_clear()
+        yield
     mimas._get_handlers.cache_clear()