stable/freqtrade/resolvers/iresolver.py

105 lines
3.9 KiB
Python
Raw Normal View History

# pragma pylint: disable=attribute-defined-outside-init
"""
This module load custom objects
"""
import importlib.util
import inspect
import logging
2018-11-24 19:39:16 +00:00
from pathlib import Path
from typing import Any, List, Optional, Tuple, Union, Generator
logger = logging.getLogger(__name__)
2019-09-12 01:39:52 +00:00
class IResolver:
"""
This class contains all the logic to load custom classes
"""
2019-10-30 14:55:35 +00:00
def build_search_paths(self, config, current_path: Path, user_subdir: Optional[str] = None,
extra_dir: Optional[str] = None) -> List[Path]:
2019-10-30 14:55:35 +00:00
abs_paths: List[Path] = [current_path]
if user_subdir:
abs_paths.insert(0, config['user_data_dir'].joinpath(user_subdir))
if extra_dir:
# Add extra directory to the top of the search paths
abs_paths.insert(0, Path(extra_dir).resolve())
return abs_paths
@staticmethod
def _get_valid_object(object_type, module_path: Path,
object_name: str) -> Generator[Any, None, None]:
"""
Generator returning objects with matching object_type and object_name in the path given.
:param object_type: object_type (class)
:param module_path: absolute path to the module
:param object_name: Class name of the object
:return: generator containing matching objects
"""
# Generate spec based on absolute path
# Pass object_name as first argument to have logging print a reasonable name.
spec = importlib.util.spec_from_file_location(object_name, str(module_path))
module = importlib.util.module_from_spec(spec)
try:
spec.loader.exec_module(module) # type: ignore # importlib does not use typehints
2019-03-15 18:50:38 +00:00
except (ModuleNotFoundError, SyntaxError) as err:
# Catch errors in case a specific module is not installed
2019-03-15 18:50:38 +00:00
logger.warning(f"Could not import {module_path} due to '{err}'")
valid_objects_gen = (
obj for name, obj in inspect.getmembers(module, inspect.isclass)
if object_name == name and object_type in obj.__bases__
)
return valid_objects_gen
@staticmethod
2018-11-24 19:39:16 +00:00
def _search_object(directory: Path, object_type, object_name: str,
2019-07-12 20:45:49 +00:00
kwargs: dict = {}) -> Union[Tuple[Any, Path], Tuple[None, None]]:
"""
Search for the objectname in the given directory
:param directory: relative or absolute directory path
:return: object instance
"""
2019-02-19 21:27:20 +00:00
logger.debug("Searching for %s %s in '%s'", object_type.__name__, object_name, directory)
2018-11-24 19:39:16 +00:00
for entry in directory.iterdir():
# Only consider python files
2018-11-24 19:39:16 +00:00
if not str(entry).endswith('.py'):
logger.debug('Ignoring %s', entry)
continue
module_path = entry.resolve()
obj = next(IResolver._get_valid_object(object_type, module_path, object_name), None)
if obj:
2019-07-12 20:45:49 +00:00
return (obj(**kwargs), module_path)
return (None, None)
@staticmethod
def _load_object(paths: List[Path], object_type, object_name: str,
2019-07-21 17:21:50 +00:00
kwargs: dict = {}) -> Optional[Any]:
"""
Try to load object from path list.
"""
for _path in paths:
try:
(module, module_path) = IResolver._search_object(directory=_path,
object_type=object_type,
object_name=object_name,
kwargs=kwargs)
if module:
logger.info(
f"Using resolved {object_type.__name__.lower()[1:]} {object_name} "
f"from '{module_path}'...")
return module
except FileNotFoundError:
logger.warning('Path "%s" does not exist.', _path.resolve())
return None