✏️ Editing: feature.py Read Only

from defence360agent.rpc_tools import ValidationError
from defence360agent.rpc_tools.lookup import RootEndpoints, bind
from defence360agent.utils import Scope
from im360.subsys.features import get_applicable_features
from defence360agent.subsys.features.abstract_feature import (
    FeatureError,
    FeatureStatus,
)

class FeaturesEndpoints(RootEndpoints):
    SCOPE = Scope.IM360

@bind("features", "list")
    async def feature_list(self):
        return {"items": list(get_applicable_features().keys())}

@bind("features", "install")
    async def feature_install(self, name):
        return await self.call("install", name)

@bind("features", "remove")
    async def feature_remove(self, name):
        return await self.call("remove", name)

@bind("features", "status")
    async def feature_status(self, name):
        return await self.call("status", name)

async def call(self, action, name):
        try:
            _cls = get_applicable_features()[name]
        except KeyError:
            raise ValidationError("Feature is not available: {}".format(name))
        else:
            if _cls is None:
                raise ValidationError(f"No implementation for feature: {name}")

try:
            feature = await _cls(sink=self._sink).init()
            return await getattr(feature, action)()
        except FeatureError as e:
            # implemented new format for status, but not for install/remove yet
            # also will be nice to move log file path to status response
            # because now, if status returned "installing", UI have to call
            # "install" (!) to get log file path
            if action == "status":
                return {
                    "items": {
                        "status": FeatureStatus.ERROR,
                        "message": str(e),
                    }
                }
            else:
                raise ValidationError(str(e))

🔐 Sid Gifari File Manager Pro