Allow installing plugs from env

docker/prod.Dockerfile

@@ -5,9 +5,11 @@ ARG TYPST_VERSION=0.11.0
 
 ARG BUILD_ENVIRONMENT="production"
 ARG APP_VERSION="unknown"
+ARG ADDITIONAL_PLUGS=""
 
 ENV BUILD_ENVIRONMENT=$BUILD_ENVIRONMENT
 ENV APP_VERSION=$APP_VERSION
+ENV ADDITIONAL_PLUGS=$ADDITIONAL_PLUGS
 ENV PYTHONUNBUFFERED=1
 ENV PYTHONDONTWRITEBYTECODE=1
 ENV PATH=/venv/bin:$PATH

install_plugins.py

@@ -1,3 +1,18 @@
+import json
+import logging
+import os
+
 from plug_config import manager
+from plugs.plug import Plug
+
+logger = logging.getLogger(__name__)
+
+
+if ADDITIONAL_PLUGS := os.getenv("ADDITIONAL_PLUGS"):
+    try:
+        for plug in json.loads(ADDITIONAL_PLUGS):
+            manager.add_plug(Plug(**plug))
+    except json.JSONDecodeError:
+        logger.error("ADDITIONAL_PLUGS is not a valid JSON")
 
 manager.install()

plugs/manager.py

@@ -2,28 +2,36 @@
 import sys
 from collections import defaultdict
 
+from plugs.plug import Plug
+
 
 class PlugManager:
     """
     Manager to manage plugs in care
     """
 
-    def __init__(self, plugs):
-        self.plugs = plugs
+    def __init__(self, plugs: list[Plug]):
+        self.plugs: list[Plug] = plugs
 
-    def install(self):
-        packages = [x.package_name + x.version for x in self.plugs]
+    def install(self) -> None:
+        packages: list[str] = [f"{x.package_name}{x.version}" for x in self.plugs]
         if packages:
-            subprocess.check_call(
-                [sys.executable, "-m", "pip", "install", " ".join(packages)]
-            )
+            subprocess.check_call([sys.executable, "-m", "pip", "install", *packages])  # noqa: S603
+
+    def add_plug(self, plug: Plug) -> None:
+        if not isinstance(plug, Plug):
+            msg = "plug must be an instance of Plug"
+            raise ValueError(msg)
+        self.plugs.append(plug)
 
-    def get_apps(self):
+    def get_apps(self) -> list[str]:
         return [plug.name for plug in self.plugs]
 
-    def get_config(self):
-        configs = defaultdict(dict)
+    def get_config(self) -> defaultdict[str, dict]:
+        configs: defaultdict[str, dict] = defaultdict(dict)
         for plug in self.plugs:
+            if plug.configs is None:
+                continue
             for key, value in plug.configs.items():
                 configs[plug.name][key] = value
         return configs

plugs/plug.py

@@ -1,10 +1,30 @@
+from dataclasses import _MISSING_TYPE, dataclass, field, fields
+
+
+@dataclass(slots=True)
 class Plug:
-    """
-    Abstraction of a plugin
-    """
-
-    def __init__(self, name, package_name, version, configs):
-        self.name = name
-        self.package_name = package_name
-        self.version = version
-        self.configs = configs
+    name: str
+    package_name: str
+    version: str = field(default="@main")
+    configs: dict = field(default_factory=dict)
+
+    def __post_init__(self):
+        for _field in fields(self):
+            if (
+                not isinstance(_field.default, _MISSING_TYPE)
+                and getattr(self, _field.name) is None
+            ):
+                setattr(self, _field.name, _field.default)
+
+        if not isinstance(self.name, str):
+            msg = "name must be a string"
+            raise ValueError(msg)
+        if not isinstance(self.package_name, str):
+            msg = "package_name must be a string"
+            raise ValueError(msg)
+        if not isinstance(self.version, str):
+            msg = "version must be a string"
+            raise ValueError(msg)
+        if not isinstance(self.configs, dict):
+            msg = "configs must be a dictionary"
+            raise ValueError(msg)