Load plugin modules from a directory.
(plugin_dirs: Union[str, List[str]])
| 24 | |
| 25 | |
| 26 | def load_plugin_from_dirs(plugin_dirs: Union[str, List[str]]) -> None: |
| 27 | """ |
| 28 | Load plugin modules from a directory. |
| 29 | """ |
| 30 | logger = get_logger(__name__) |
| 31 | if not isinstance(plugin_dirs, list): |
| 32 | plugin_dirs = [plugin_dirs] |
| 33 | plugin_dirs = set(plugin_dirs) |
| 34 | for plugin_dir in plugin_dirs: |
| 35 | if not os.path.exists(plugin_dir): |
| 36 | logger.error(f"plugin-dir [{plugin_dir}] does not exist.") |
| 37 | continue |
| 38 | if not os.path.isdir(plugin_dir): |
| 39 | logger.error(f"plugin-dir [{plugin_dir}] is not a directory.") |
| 40 | continue |
| 41 | |
| 42 | for file in Path(plugin_dir).glob("*.py"): |
| 43 | if file.name.startswith("__"): |
| 44 | continue |
| 45 | logger.info(f"Loading plugin modules from [{file}]...") |
| 46 | # load modules from file |
| 47 | try: |
| 48 | load_from_file(str(file)) |
| 49 | except Exception as e: |
| 50 | logger.warning(f"Failed to load plugin module from [{file}]: {e}") |
| 51 | |
| 52 | |
| 53 | def load_from_file(file_path: str): |
no test coverage detected