kraken: resolve cyclic import

Author: nicoschmdt

core/services/kraken/extension_logs.py

@@ -1,14 +1,13 @@
 import asyncio
 import json
 import time
-from typing import Callable, Dict, List, Optional, Tuple, cast
+from typing import Callable, Dict, Optional, Tuple
 
 from commonwealth.utils.zenoh_helper import ZenohSession
 from config import SERVICE_NAME
-from extension.extension import Extension
 from harbor import ContainerManager
 from loguru import logger
-from settings import ExtensionSettings
+from settings import ExtensionSettings, get_extension_settings
 
 
 class ExtensionLogPublisher:
@@ -51,7 +50,7 @@ async def _collect_desired_streams(self) -> Optional[Dict[str, ExtensionSettings
             return None
 
         running_names = {container.name.lstrip("/") for container in running_containers}
-        extensions = cast(List[ExtensionSettings], Extension._fetch_settings())
+        extensions = get_extension_settings()
 
         desired: Dict[str, ExtensionSettings] = {}
         for extension in extensions:

core/services/kraken/settings.py

@@ -1,11 +1,17 @@
 import json
 import re
-from typing import Any, Dict
+from typing import Any, Dict, List
 
 from commonwealth.settings import settings
+from commonwealth.settings.manager import Manager
+from config import SERVICE_NAME
 from pykson import BooleanField, IntegerField, JsonObject, ObjectListField, StringField
 
 
+def get_extension_settings() -> List["ExtensionSettings"]:
+    return list(Manager(SERVICE_NAME, SettingsV2).settings.extensions)
+
+
 class ExtensionSettings(JsonObject):
     identifier = StringField()
     name = StringField()

core/services/kraken/zenoh_handlers/extension_handler.py

@@ -1,11 +1,10 @@
-from typing import Any, List, cast
+from typing import Any
 
 from commonwealth.utils.zenoh_helper import ZenohRouter
-from extension.extension import Extension
 from extension_logs import ExtensionLogPublisher
 from harbor import ContainerManager
 from loguru import logger
-from settings import ExtensionSettings
+from settings import get_extension_settings
 
 
 class ExtensionHandlers:
@@ -17,7 +16,7 @@ async def logs_request_handler(self, extension_name: str) -> dict[str, Any]:
             return {"error": "extension_name parameter is required"}
 
         try:
-            extensions = cast(List[ExtensionSettings], Extension._fetch_settings())
+            extensions = get_extension_settings()
             extension = next((ext for ext in extensions if extension_name in (ext.identifier, ext.name)), None)
 
             if not extension: