Load conda-lock plugin

conda-store-server/conda_store_server/app.py

@@ -7,6 +7,7 @@
 import sys
 from contextlib import contextmanager
 from typing import Any, Dict
+import pluggy
 
 import pydantic
 from celery import Celery, group
@@ -27,6 +28,8 @@
 from traitlets.config import LoggingConfigurable
 
 from conda_store_server import CONDA_STORE_DIR, BuildKey, api, registry, storage
+from conda_store_server.plugins import hookspec, registry
+from conda_store_server.exception import CondaStorePluginNotFoundError
 from conda_store_server._internal import conda_utils, environment, orm, schema, utils
 
 
@@ -471,6 +474,35 @@ def celery_app(self):
         self._celery_app.config_from_object(self.celery_config)
         return self._celery_app
 
+    @property
+    def plugin_registry(self):
+        if hasattr(self, "_plugin_registry"):
+            return self._plugin_registry
+
+        self._plugin_registry = registry.PluginRegistry()
+        self._plugin_registry.collect_plugins()
+        return self._plugin_registry
+
+    @property
+    def plugin_manager(self):
+        if hasattr(self, "_plugin_manager"):
+            return self._plugin_manager
+
+        self._plugin_manager = pluggy.PluginManager(hookspec.spec_name)
+        self._plugin_manager.add_hookspecs(hookspec.CondaStoreSpecs)
+
+        # Load lock plugin
+        self.load_plugin_by_name("lock-conda_lock")
+
+        return self._plugin_manager
+
+    def load_plugin_by_name(self, name, *args, **kwargs):
+        """Loads a plugin from the plugin registry into the plugin manager"""
+        target_plugin = self.plugin_registry.get_plugin(name)
+        if target_plugin is None:
+            raise CondaStorePluginNotFoundError(name, self.plugin_registry.list_plugin_names())
+        self.plugin_manager.register(target_plugin(*args, **kwargs))
+
     def ensure_settings(self, db: Session):
         """Ensure that conda-store traitlets settings are applied"""
         settings = schema.Settings(

conda-store-server/conda_store_server/exception.py

@@ -0,0 +1,19 @@
+# Copyright (c) conda-store development team. All rights reserved.
+# Use of this source code is governed by a BSD-style
+# license that can be found in the LICENSE file.
+
+
+class CondaStoreError(Exception):
+    pass
+
+
+class CondaStorePluginNotFoundError(CondaStoreError):
+    """Exception raised by conda store when a specified plugin is not found
+    Attributes:
+        plugin -- plugin that was not found
+        available_plugins -- list of registered plugins
+    """
+
+    def __init__(self, plugin, available_plugins):
+        self.message = f"Plugin {plugin} was requested but not found! The following plugins are available {','.join(available_plugins)}"
+        super().__init__(self.message)