Initialize remote loaders

Author: DUYN

bot/remote_loaders/__init__.py

@@ -0,0 +1,2 @@
+from bot.remote_loaders.strategy_remote_loader import StrategyRemoteLoader
+from bot.remote_loaders.data_provider_remote_loader import DataProviderRemoteLoader

bot/remote_loaders/data_provider_remote_loader.py

@@ -0,0 +1,33 @@
+import logging
+from pathlib import Path
+
+from bot.data import DataProvider
+from bot import OperationalException
+from bot.constants import PLUGIN_STRATEGIES_DIR, PLUGIN_DATA_PROVIDERS_DIR
+from bot.remote_loaders.remote_loader import RemoteLoader
+
+logger = logging.getLogger(__name__)
+
+
+class DataProviderRemoteLoader(RemoteLoader):
+
+    def load_data_provider(self, data_provider_class_name: str) -> DataProvider:
+
+        logger.info("Loading remote data providers ...")
+
+        if not data_provider_class_name:
+            raise OperationalException("Provided data provider has it search class name not set")
+
+        data_providers_dir = Path(PLUGIN_DATA_PROVIDERS_DIR)
+        modules = self.locate_python_modules(data_providers_dir)
+        location = self.locate_class(modules, data_provider_class_name)
+        generator = self.create_class_generators(location, data_provider_class_name, DataProvider)
+        data_provider: DataProvider = next(generator, None)
+
+        if data_provider and issubclass(data_provider, DataProvider):
+            return data_provider
+
+        raise OperationalException(
+            f"Impossible to load Strategy '{data_provider_class_name}'. This strategy does not exist "
+            "or contains Python code errors."
+        )

bot/remote_loaders/remote_loader.py

@@ -0,0 +1,89 @@
+import logging
+from abc import ABC
+from pathlib import Path
+from inspect import getmembers, isclass
+from typing import List, Type, Generator, Any
+from importlib.util import spec_from_file_location, module_from_spec
+
+from bot import OperationalException, DependencyException
+
+
+logger = logging.getLogger(__name__)
+
+
+class RemoteLoader(ABC):
+    """
+    RemoteLoader class: abstract class with some util functions to handle searching for modules,
+    classes and instantiating classes
+    """
+
+    @staticmethod
+    def locate_python_modules(dir_path: Path) -> List[Path]:
+        """
+        Functions that will search through all the files in a directory to locate the class matching the
+        given class name.
+        """
+
+        if not dir_path.is_dir():
+            raise OperationalException("Given directory path is not a directory")
+
+        modules: List[Path] = []
+
+        logger.info("Searching in directory {} ...".format(dir_path))
+
+        for entry in dir_path.iterdir():
+
+            if not str(entry).endswith('.py'):
+                continue
+
+            logger.info("Found module: {}, appending it to search paths".format(str(entry)))
+            modules.append(entry)
+
+        return modules
+
+    @staticmethod
+    def locate_class(modules: List[Path], class_name: str) -> Path:
+        """
+        Function that will search all the given modules and will return the corresponding path where
+        the class is located.
+        """
+
+        logger.info("Searching for class: {}".format(class_name))
+
+        for module_path in modules:
+            spec = spec_from_file_location(module_path.stem, str(module_path))
+            module = module_from_spec(spec)
+
+            try:
+                spec.loader.exec_module(module)
+            except (ModuleNotFoundError, SyntaxError) as err:
+                # Catch errors in case a specific module is not installed
+                logger.warning(f"Could not import {module_path} due to '{err}'")
+
+            if hasattr(module, class_name):
+                logger.info("Found class {} in module {}".format(class_name, str(module_path)))
+                return module_path
+
+        raise DependencyException("Could not find given class in selection of modules")
+
+    @staticmethod
+    def create_class_generators(module_path: Path, class_name: str,
+                                class_type: Type[Any]) -> Generator[Any, None, None]:
+        """
+        Function that creates a generator for a given module path and class name
+        """
+        spec = spec_from_file_location(module_path.stem, str(module_path))
+        module = module_from_spec(spec)
+
+        try:
+            spec.loader.exec_module(module)
+        except (ModuleNotFoundError, SyntaxError) as err:
+            # Catch errors in case a specific module is not installed
+            logger.warning(f"Could not import {module_path} due to '{err}'")
+
+        object_generators = (
+            obj for name, obj in getmembers(module, isclass) if (class_name is None or class_name == name)
+            and class_type in obj.__bases__
+        )
+       
+        return object_generators

bot/remote_loaders/strategy_remote_loader.py

@@ -1,37 +1,34 @@
-import os
-from typing import Dict, Any, List
+import logging
+from pathlib import Path
 
+from bot.strategies import Strategy
 from bot import OperationalException
-from bot.strategy.strategy import Strategy
+from bot.constants import PLUGIN_STRATEGIES_DIR
+from bot.remote_loaders.remote_loader import RemoteLoader
 
+logger = logging.getLogger(__name__)
 
-class StrategyRemoteLoader:
 
-    def __init__(self, config: Dict[str, Any]):
+class StrategyRemoteLoader(RemoteLoader):
 
-        self._strategies: List[Strategy] = None
+    def load_strategy(self, strategy_class_name: str) -> Strategy:
 
-        if not config.get('strategies'):
-            raise OperationalException(
-                "Could not resolve strategies, please provide the strategy "
-                "paths in your config file. You could also use de default strategies, that can be found in the "
-                "strategies directory. If you have difficulties creating your own strategies, please see the "
-                "documentation"
-            )
+        logger.info("Loading remote strategies ...")
 
-        strategies = config.get('strategies')
+        if not strategy_class_name:
+            raise OperationalException("Provided strategies has it search class name not set")
 
-        for strategy in strategies:
-            path = strategy.get('path')
-            self._strategies.append(self.load_strategy(path))
-
-    @staticmethod
-    def load_strategy(path: str) -> Strategy:
-
-        if not path:
-            raise OperationalException("Provided strategies has it file path not set")
-
-        current_path = os.path.dirname(os.path.abspath(__file__))
+        strategies_dir = Path(PLUGIN_STRATEGIES_DIR)
+        modules = self.locate_python_modules(strategies_dir)
+        location = self.locate_class(modules, strategy_class_name)
+        generator = self.create_class_generators(location, strategy_class_name, Strategy)
+        strategy: Strategy = next(generator, None)
 
+        if strategy and issubclass(strategy, Strategy):
+            return strategy
 
+        raise OperationalException(
+            f"Impossible to load Strategy '{strategy_class_name}'. This strategy does not exist "
+            "or contains Python code errors."
+        )