changed globals to import_module

config/config.yml

@@ -21,14 +21,14 @@ PROFILES: # profiles determine which schema is used for the linked data
     prefix: "nisv"  # based on @prefix nisv: <http://data.rdlabs.beeldengeluid.nl/schema/> .
     schema: "resource/bengSchema.ttl"
     mapping: "resource/daan-mapping-storage.ttl"
-    storage_handler: DAANStorageLODHandler
+    storage_handler: apis.resource.DAANStorageLODHandler.DAANStorageLODHandler
   - 
     title: "NISV Catalogue using schema.org ontology"
     uri: "https://schema.org/"
     prefix: "sdo"  # based on @prefix sdo: <https://schema.org/> .
     schema: "resource/schema-dot-org.ttl"
     mapping: "resource/daan-mapping-schema-org.ttl"
-    storage_handler: SDOStorageLODHandler
+    storage_handler: apis.resource.SDOStorageLODHandler.SDOStorageLODHandler
     ob_links: "resource/ob_link_matches.json"
     roles: "resource/music_roles.csv"
     default: True  # this profile is loaded in memory by default

src/apis/resource/resource_api.py

@@ -14,11 +14,11 @@
     json_iri_lit_from_rdf_graph,
     json_iri_bnode_from_rdf_graph,
 )
-from apis.resource.DAANStorageLODHandler import DAANStorageLODHandler  # noqa: F401
-from apis.resource.SDOStorageLODHandler import SDOStorageLODHandler  # noqa: F401
+from util.base_util import import_class
 
 logger = logging.getLogger()
 
+
 api = Namespace(
     "resource",
     description="Resources in RDF for Netherlands Institute for Sound and Vision.",
@@ -155,9 +155,8 @@ def _get_lod_resource(
             f"Getting requested resource with level: '{level}' and '{identifier}' from the flex store using profile '{profile_prefix}'."
         )
 
-        # make sure the approved lodhandlers are imported (then available in globals)
-        # for us, its the classes: DAANStorageLODHandler and SDOStorageLODHandler
-        resp, status_code, headers = globals()[profile["storage_handler"]](
+        storage_handler_class = import_class(profile["storage_handler"])
+        resp, status_code, headers = storage_handler_class(
             app_config, profile
         ).get_storage_record(level, identifier, mt.to_ld_format())
 

src/tests/unit_tests/apis/resource/sdo_storage_lod_handler_test.py

@@ -6,6 +6,7 @@
 from rdflib.plugin import PluginException
 from models.SDORdfConcept import SDORdfConcept
 from apis.mime_type_util import MimeType
+from util.base_util import import_class
 from apis.resource.SDOStorageLODHandler import SDOStorageLODHandler
 
 
@@ -50,9 +51,8 @@ def test_get_storage_record__ob_scene_payload(application_settings, i_ob_scene_p
     try:
         profile = application_settings.get("ACTIVE_PROFILE")
         storage_base_url = application_settings.get("STORAGE_BASE_URL")
-        sdo_handler = globals()[profile["storage_handler"]](
-            application_settings, profile
-        )
+        sdo_handler_class = import_class(profile["storage_handler"])
+        sdo_handler = sdo_handler_class(application_settings, profile)
         when(sdo_handler)._prepare_storage_uri(
             storage_base_url, DUMMY_LEVEL, DUMMY_ID
         ).thenReturn(DUMMY_STORAGE_URL)
@@ -101,9 +101,8 @@ def test_get_storage_record__error_scene_payload(
     try:
         profile = application_settings.get("ACTIVE_PROFILE")
         storage_base_url = application_settings.get("STORAGE_BASE_URL")
-        sdo_handler = globals()[profile["storage_handler"]](
-            application_settings, profile
-        )
+        sdo_handler_class = import_class(profile["storage_handler"])
+        sdo_handler = sdo_handler_class(application_settings, profile)
         when(sdo_handler)._prepare_storage_uri(
             storage_base_url, "scene", "2101702260627885424"
         ).thenReturn(DUMMY_STORAGE_URL)
@@ -127,9 +126,8 @@ def test_get_storage_record__no_storage_data(application_settings):
     try:
         profile = application_settings.get("ACTIVE_PROFILE")
         storage_base_url = application_settings.get("STORAGE_BASE_URL")
-        sdo_handler = globals()[profile["storage_handler"]](
-            application_settings, profile
-        )
+        sdo_handler_class = import_class(profile["storage_handler"])
+        sdo_handler = sdo_handler_class(application_settings, profile)
         when(sdo_handler)._prepare_storage_uri(
             storage_base_url, "scene", "2101702260627885424"
         ).thenReturn(DUMMY_STORAGE_URL)
@@ -153,8 +151,11 @@ def test_get_storage_record__no_storage_data(application_settings):
 
 @pytest.fixture(scope="function")
 def sdo_storage_lod_handler(application_settings):
-    active_profile = application_settings.get("ACTIVE_PROFILE")
-    yield SDOStorageLODHandler(application_settings, active_profile)
+    for p in application_settings["PROFILES"]:
+        if "uri" in p and p["uri"] == "https://schema.org/":
+            sdo_handler_class = import_class(p["storage_handler"])
+            yield sdo_handler_class(application_settings, p)
+            break
 
 
 def test_init(sdo_storage_lod_handler):

src/util/base_util.py

@@ -1,7 +1,9 @@
+import sys
 import logging
 import os.path
 import validators
 from pathlib import Path
+from importlib import import_module
 
 logger = logging.getLogger(__name__)
 LOG_FORMAT = "%(asctime)s|%(levelname)s|%(process)d|%(module)s|%(funcName)s|%(lineno)d|%(message)s"
@@ -121,7 +123,10 @@ def __check_setting(config, key, t, optional=False):
 
 
 def __check_storage_handler(handler: str) -> bool:
-    return handler in ["DAANStorageLODHandler", "SDOStorageLODHandler"]
+    return handler in [
+        "apis.resource.DAANStorageLODHandler.DAANStorageLODHandler",
+        "apis.resource.SDOStorageLODHandler.SDOStorageLODHandler",
+    ]
 
 
 def __check_log_level(level: str) -> bool:
@@ -138,3 +143,24 @@ def __validate_file_paths(paths: list) -> bool:
 
 def get_parent_dir(path: str) -> Path:
     return Path(path).parent
+
+
+def import_class(module_class_string: str):
+    """Returns an imported class, depending on the definition in the
+    module_class_string parameter, coming from the config file.
+    Calling method needs to create the class instance dynamically.
+    :param module_class_path: string containing "<module_path>.<class_name>".
+    """
+    try:
+        module_name, class_name = module_class_string.rsplit(".", 1)
+        module = import_module(module_name)
+        logger.debug(f"reading class {class_name} from module {module_name}")
+        return getattr(module, class_name)
+    except ValueError:
+        logger.exception("Module class string incorrect.")
+        sys.exit()
+    except ModuleNotFoundError:
+        logger.exception("Module path incorrectly configured")
+    except AttributeError:
+        logger.exception("Module class incorrectly configured")
+    return None