'backports.entry_points_selectable' no longer needed as a package dependency

Author: tieneupin

pyproject.toml

@@ -29,7 +29,6 @@ classifiers = [
     "Programming Language :: Python :: 3.12",
 ]
 dependencies = [
-    "backports.entry_points_selectable",
     "defusedxml", # For safely parsing XML files
     "pydantic>=2",
     "pydantic-settings",

src/murfey/client/context.py

@@ -1,11 +1,10 @@
 from __future__ import annotations
 
 import logging
+from importlib.metadata import entry_points
 from pathlib import Path
 from typing import Any, Dict, List, NamedTuple
 
-from backports.entry_points_selectable import entry_points
-
 from murfey.client.instance_environment import MurfeyInstanceEnvironment
 
 logger = logging.getLogger("murfey.client.context")

src/murfey/server/api/clem.py

@@ -3,12 +3,14 @@
 import re
 import traceback
 from ast import literal_eval
-from importlib.metadata import EntryPoint  # type hinting only
+from importlib.metadata import (
+    EntryPoint,  # type hinting only
+    entry_points,
+)
 from logging import getLogger
 from pathlib import Path
 from typing import Literal, Optional, Type, Union
 
-from backports.entry_points_selectable import entry_points
 from fastapi import APIRouter
 from pydantic import BaseModel, field_validator
 from sqlalchemy.exc import NoResultFound
@@ -752,9 +754,7 @@ def process_raw_lifs(
     try:
         # Try and load relevant Murfey workflow
         workflow: EntryPoint = list(
-            entry_points().select(
-                group="murfey.workflows", name="clem.process_raw_lifs"
-            )
+            entry_points(group="murfey.workflows", name="clem.process_raw_lifs")
         )[0]
     except IndexError:
         raise RuntimeError("The relevant Murfey workflow was not found")
@@ -792,9 +792,7 @@ def process_raw_tiffs(
     try:
         # Try and load relevant Murfey workflow
         workflow: EntryPoint = list(
-            entry_points().select(
-                group="murfey.workflows", name="clem.process_raw_tiffs"
-            )
+            entry_points(group="murfey.workflows", name="clem.process_raw_tiffs")
         )[0]
     except IndexError:
         raise RuntimeError("The relevant Murfey workflow was not found")
@@ -853,7 +851,7 @@ def align_and_merge_stacks(
     try:
         # Try and load relevant Murfey workflow
         workflow: EntryPoint = list(
-            entry_points().select(group="murfey.workflows", name="clem.align_and_merge")
+            entry_points(group="murfey.workflows", name="clem.align_and_merge")
         )[0]
     except IndexError:
         raise RuntimeError("The relevant Murfey workflow was not found")

src/murfey/server/feedback.py

@@ -13,13 +13,15 @@
 import time
 from datetime import datetime
 from functools import partial
-from importlib.metadata import EntryPoint  # For type hinting only
+from importlib.metadata import (
+    EntryPoint,  # For type hinting only
+    entry_points,
+)
 from pathlib import Path
 from typing import Dict, List, NamedTuple, Tuple
 
 import mrcfile
 import numpy as np
-from backports.entry_points_selectable import entry_points
 from sqlalchemy import func
 from sqlalchemy.exc import (
     InvalidRequestError,
@@ -2193,14 +2195,10 @@ def feedback_callback(header: dict, message: dict, _db=murfey_db) -> None:
             if murfey.server._transport_object:
                 murfey.server._transport_object.transport.ack(header)
             return None
-        elif (
-            message["register"] in entry_points().select(group="murfey.workflows").names
-        ):
+        elif message["register"] in entry_points(group="murfey.workflows").names:
             # Search for corresponding workflow
             workflows: list[EntryPoint] = list(
-                entry_points().select(
-                    group="murfey.workflows", name=message["register"]
-                )
+                entry_points(group="murfey.workflows", name=message["register"])
             )  # Returns either 1 item or empty list
             if not workflows:
                 logger.error(f"No workflow found for {sanitise(message['register'])}")

src/murfey/server/main.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
 import logging
+from importlib.metadata import entry_points
 
-from backports.entry_points_selectable import entry_points
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.staticfiles import StaticFiles

src/murfey/util/config.py

@@ -3,11 +3,11 @@
 import os
 import socket
 from functools import lru_cache
+from importlib.metadata import entry_points
 from pathlib import Path
 from typing import Any, Literal, Optional
 
 import yaml
-from backports.entry_points_selectable import entry_points
 from pydantic import BaseModel, ConfigDict, RootModel, ValidationInfo, field_validator
 from pydantic_settings import BaseSettings
 
@@ -270,6 +270,6 @@ def get_extended_machine_config(
     )
     if not machine_config:
         return None
-    model = entry_points().select(group="murfey.config", name=extension_name)[0].load()
+    model = entry_points(group="murfey.config", name=extension_name)[0].load()
     data = getattr(machine_config, extension_name, {})
     return model(**data)

src/murfey/workflows/register_atlas_update.py

@@ -1,6 +1,6 @@
 import logging
+from importlib.metadata import entry_points
 
-from backports.entry_points_selectable import entry_points
 from sqlmodel.orm.session import Session as SQLModelSession
 
 from murfey.server import _transport_object
@@ -25,9 +25,7 @@ def run(
         message["atlas_pixel_size"],
         message["sample"],
     )
-    if dcg_hooks := entry_points().select(
-        group="murfey.hooks", name="data_collection_group"
-    ):
+    if dcg_hooks := entry_points(group="murfey.hooks", name="data_collection_group"):
         try:
             for hook in dcg_hooks:
                 hook.load()(message["dcgid"], session_id=message["session_id"])

src/murfey/workflows/register_data_collection_group.py

@@ -1,8 +1,8 @@
 import logging
 import time
+from importlib.metadata import entry_points
 
 import ispyb.sqlalchemy._auto_db_schema as ISPyBDB
-from backports.entry_points_selectable import entry_points
 from sqlmodel import select
 from sqlmodel.orm.session import Session as SQLModelSession
 
@@ -87,9 +87,7 @@ def run(
         )
         return {"success": False, "requeue": True}
 
-    if dcg_hooks := entry_points().select(
-        group="murfey.hooks", name="data_collection_group"
-    ):
+    if dcg_hooks := entry_points(group="murfey.hooks", name="data_collection_group"):
         try:
             for hook in dcg_hooks:
                 hook.load()(dcgid, session_id=message["session_id"])

tests/server/test_feedback.py

@@ -1,7 +1,7 @@
+from importlib.metadata import entry_points
 from unittest.mock import MagicMock
 
 import pytest
-from backports.entry_points_selectable import entry_points
 from pytest_mock import MockerFixture
 
 feedback_callback_params_matrix = (
@@ -46,7 +46,7 @@ def test_feedback_callback(
     mock_sql_session.return_value = mock_murfey_db
 
     # Load the entry point and patch the executable it calls
-    eps = list(entry_points().select(group="murfey.workflows", name=entry_point_name))
+    eps = list(entry_points(group="murfey.workflows", name=entry_point_name))
     assert len(eps) == 1  # Entry point should be present and unique
     mock_function = mocker.patch(eps[0].value.replace(":", "."))