refactor: extract functions

lnbits/core/models/extensions.py

@@ -8,7 +8,7 @@ import shutil
 import zipfile
 from asyncio.tasks import create_task
 from collections.abc import Mapping
-from pathlib import Path
+from pathlib import Path, PurePosixPath
 from typing import Any
 
 import httpx
@@ -473,6 +473,30 @@ class InstallableExtension(BaseModel):
                 os.remove(ext_zip_file)
             raise AssertionError("File hash missmatch. Will not install.")
 
+    def load_archive_config(self) -> dict[str, Any]:
+        if not self.zip_path.is_file():
+            return {}
+
+        try:
+            with zipfile.ZipFile(self.zip_path, "r") as archive:
+                config_name = self._archive_config_name(archive.namelist())
+                if not config_name:
+                    return {}
+                with archive.open(config_name) as config_file:
+                    config = json.load(config_file)
+        except Exception as exc:
+            raise ValueError(f"Cannot read extension config for '{self.id}'.") from exc
+
+        return config if isinstance(config, dict) else {}
+
+    @staticmethod
+    def _archive_config_name(names: list[str]) -> str | None:
+        for name in names:
+            path = PurePosixPath(name)
+            if len(path.parts) == 2 and path.name == "config.json":
+                return name
+        return None
+
     def extract_archive(self):
         logger.info(f"Extracting extension {self.name} ({self.installed_version}).")
         Path(settings.lnbits_extensions_upgrade_path).mkdir(parents=True, exist_ok=True)

lnbits/core/services/extensions.py

@@ -1,9 +1,5 @@
 import asyncio
 import importlib
-import json
-import zipfile
-from pathlib import PurePosixPath
-from typing import Any
 
 from loguru import logger
 
@@ -56,7 +52,7 @@ async def install_extension(
     if not skip_download:
         await ext_info.download_archive()
 
-    extension_config = _load_extension_archive_config(ext_info)
+    extension_config = ext_info.load_archive_config()
     ext_info.permissions = validate_wasm_extension_permissions(
         ext_info, granted_permissions, extension_config
     )
@@ -84,31 +80,6 @@ async def install_extension(
     return extension
 
 
-def _load_extension_archive_config(ext_info: InstallableExtension) -> dict[str, Any]:
-    if not ext_info.zip_path.is_file():
-        return {}
-
-    try:
-        with zipfile.ZipFile(ext_info.zip_path, "r") as archive:
-            config_name = _archive_config_name(archive.namelist())
-            if not config_name:
-                return {}
-            with archive.open(config_name) as config_file:
-                config = json.load(config_file)
-    except Exception as exc:
-        raise ValueError(f"Cannot read extension config for '{ext_info.id}'.") from exc
-
-    return config if isinstance(config, dict) else {}
-
-
-def _archive_config_name(names: list[str]) -> str | None:
-    for name in names:
-        path = PurePosixPath(name)
-        if len(path.parts) == 2 and path.name == "config.json":
-            return name
-    return None
-
-
 async def check_extensions_limit(installed_ext: InstallableExtension | None = None):
     if settings.lnbits_max_extensions == 0 or installed_ext:
         return