distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""ClientAppLoader — dynamic client application loader.
2
3Ported from net.i2p.router.startup.LoadClientAppsJob.
4
5Loads client applications from configuration at router startup.
6Each app entry specifies a module path, class name, optional args,
7and a startup delay.
8"""
9
10from __future__ import annotations
11
12import importlib
13import logging
14import time
15
16logger = logging.getLogger(__name__)
17
18
19class ClientAppLoader:
20 """Loads client applications from config at router startup."""
21
22 def __init__(self, config: list[dict] | None = None) -> None:
23 self._config = config or []
24 self._loaded: list = []
25
26 def load_apps(self) -> list:
27 """Load and instantiate all configured client apps.
28
29 Config format:
30 [
31 {
32 "module": "mypackage.mymodule",
33 "class": "MyApp",
34 "args": ["arg1", "arg2"], # optional
35 "delay": 0.0, # optional, seconds
36 },
37 ...
38 ]
39
40 Returns list of instantiated app objects.
41 """
42 self._loaded = []
43 for entry in self._config:
44 module_path = entry.get("module", "")
45 class_name = entry.get("class", "")
46 args = entry.get("args", [])
47 delay = entry.get("delay", 0.0)
48
49 if not module_path or not class_name:
50 logger.warning("Skipping incomplete app config: %s", entry)
51 continue
52
53 app = self._load_app(module_path, class_name, args, delay)
54 if app is not None:
55 self._loaded.append(app)
56
57 logger.info("Loaded %d client apps", len(self._loaded))
58 return list(self._loaded)
59
60 def _load_app(self, module_path: str, class_name: str,
61 args: list, delay: float) -> object | None:
62 """Load a single client app."""
63 if delay > 0:
64 time.sleep(delay)
65
66 try:
67 module = importlib.import_module(module_path)
68 cls = getattr(module, class_name)
69 return cls(*args)
70 except (ImportError, AttributeError) as e:
71 logger.error("Failed to load %s.%s: %s", module_path, class_name, e)
72 return None
73 except Exception as e:
74 logger.error("Error instantiating %s.%s: %s", module_path, class_name, e)
75 return None
76
77 @property
78 def loaded_apps(self) -> list:
79 return list(self._loaded)