Send patches - preferably formatted by git format-patch - to patches at archlinux32 dot org.
summaryrefslogtreecommitdiff
path: root/archinstall/lib/plugins.py
diff options
context:
space:
mode:
Diffstat (limited to 'archinstall/lib/plugins.py')
-rw-r--r--archinstall/lib/plugins.py66
1 files changed, 38 insertions, 28 deletions
diff --git a/archinstall/lib/plugins.py b/archinstall/lib/plugins.py
index 0ff63610..b1ece04f 100644
--- a/archinstall/lib/plugins.py
+++ b/archinstall/lib/plugins.py
@@ -3,77 +3,86 @@ import importlib
import logging
import os
import sys
-import pathlib
import urllib.parse
import urllib.request
from importlib import metadata
+from pathlib import Path
from typing import Optional, List
-from types import ModuleType
from .output import log
from .storage import storage
plugins = {}
+
# 1: List archinstall.plugin definitions
# 2: Load the plugin entrypoint
# 3: Initiate the plugin and store it as .name in plugins
for plugin_definition in metadata.entry_points().select(group='archinstall.plugin'):
plugin_entrypoint = plugin_definition.load()
+
try:
plugins[plugin_definition.name] = plugin_entrypoint()
except Exception as err:
- log(err, level=logging.ERROR)
+ log(f'Error: {err}', level=logging.ERROR)
log(f"The above error was detected when loading the plugin: {plugin_definition}", fg="red", level=logging.ERROR)
-# The following functions and core are support structures for load_plugin()
-def localize_path(profile_path :str) -> str:
- if (url := urllib.parse.urlparse(profile_path)).scheme and url.scheme in ('https', 'http'):
- converted_path = f"/tmp/{os.path.basename(profile_path).replace('.py', '')}_{hashlib.md5(os.urandom(12)).hexdigest()}.py"
+def localize_path(path: Path) -> Path:
+ """
+ Support structures for load_plugin()
+ """
+ url = urllib.parse.urlparse(str(path))
+
+ if url.scheme and url.scheme in ('https', 'http'):
+ converted_path = Path(f'/tmp/{path.stem}_{hashlib.md5(os.urandom(12)).hexdigest()}.py')
with open(converted_path, "w") as temp_file:
temp_file.write(urllib.request.urlopen(url.geturl()).read().decode('utf-8'))
return converted_path
else:
- return profile_path
+ return path
-def import_via_path(path :str, namespace :Optional[str] = None) -> ModuleType:
+def import_via_path(path: Path, namespace: Optional[str] = None) -> Optional[str]:
if not namespace:
namespace = os.path.basename(path)
if namespace == '__init__.py':
- path = pathlib.PurePath(path)
namespace = path.parent.name
try:
spec = importlib.util.spec_from_file_location(namespace, path)
- imported = importlib.util.module_from_spec(spec)
- sys.modules[namespace] = imported
- spec.loader.exec_module(sys.modules[namespace])
+ if spec and spec.loader:
+ imported = importlib.util.module_from_spec(spec)
+ sys.modules[namespace] = imported
+ spec.loader.exec_module(sys.modules[namespace])
return namespace
except Exception as err:
- log(err, level=logging.ERROR)
+ log(f'Error: {err}', level=logging.ERROR)
log(f"The above error was detected when loading the plugin: {path}", fg="red", level=logging.ERROR)
try:
- del(sys.modules[namespace]) # noqa: E275
- except:
+ del sys.modules[namespace]
+ except Exception:
pass
-def find_nth(haystack :List[str], needle :str, n :int) -> int:
- start = haystack.find(needle)
- while start >= 0 and n > 1:
- start = haystack.find(needle, start + len(needle))
- n -= 1
- return start
+ return namespace
+
+
+def find_nth(haystack: List[str], needle: str, n: int) -> Optional[int]:
+ indices = [idx for idx, elem in enumerate(haystack) if elem == needle]
+ if n <= len(indices):
+ return indices[n - 1]
+ return None
+
-def load_plugin(path :str) -> ModuleType:
- parsed_url = urllib.parse.urlparse(path)
- log(f"Loading plugin {parsed_url}.", fg="gray", level=logging.INFO)
+def load_plugin(path: Path):
+ namespace: Optional[str] = None
+ parsed_url = urllib.parse.urlparse(str(path))
+ log(f"Loading plugin from url {parsed_url}.", level=logging.INFO)
# The Profile was not a direct match on a remote URL
if not parsed_url.scheme:
@@ -81,9 +90,10 @@ def load_plugin(path :str) -> ModuleType:
if os.path.isfile(path):
namespace = import_via_path(path)
elif parsed_url.scheme in ('https', 'http'):
- namespace = import_via_path(localize_path(path))
+ localized = localize_path(path)
+ namespace = import_via_path(localized)
- if namespace in sys.modules:
+ if namespace and namespace in sys.modules:
# Version dependency via __archinstall__version__ variable (if present) in the plugin
# Any errors in version inconsistency will be handled through normal error handling if not defined.
if hasattr(sys.modules[namespace], '__archinstall__version__'):
@@ -99,7 +109,7 @@ def load_plugin(path :str) -> ModuleType:
plugins[namespace] = sys.modules[namespace].Plugin()
log(f"Plugin {plugins[namespace]} has been loaded.", fg="gray", level=logging.INFO)
except Exception as err:
- log(err, level=logging.ERROR)
+ log(f'Error: {err}', level=logging.ERROR)
log(f"The above error was detected when initiating the plugin: {path}", fg="red", level=logging.ERROR)
else:
log(f"Plugin '{path}' is missing a valid entry-point or is corrupt.", fg="yellow", level=logging.WARNING)