feat(plugin): basic communication

pkg/plugin/handler.py

@@ -1,7 +1,64 @@
 from __future__ import annotations
 
+from typing import Any
+
+import sqlalchemy
+
 from langbot_plugin.runtime.io import handler
+from langbot_plugin.runtime.io.connection import Connection
+from langbot_plugin.entities.io.actions.enums import (
+    CommonAction,
+    RuntimeToLangBotAction,
+)
+
+from ..entity.persistence import plugin as persistence_plugin
+
+from ..core import app
 
 
 class RuntimeConnectionHandler(handler.Handler):
     """Runtime connection handler"""
+
+    ap: app.Application
+
+    def __init__(self, connection: Connection, ap: app.Application):
+        super().__init__(connection)
+        self.ap = ap
+
+        @self.action(RuntimeToLangBotAction.GET_PLUGIN_SETTINGS)
+        async def get_plugin_settings(data: dict[str, Any]) -> handler.ActionResponse:
+            """Get plugin settings"""
+
+            plugin_author = data['plugin_author']
+            plugin_name = data['plugin_name']
+
+            result = await self.ap.persistence_mgr.execute_async(
+                sqlalchemy.select(persistence_plugin.PluginSetting)
+                .where(persistence_plugin.PluginSetting.plugin_author == plugin_author)
+                .where(persistence_plugin.PluginSetting.plugin_name == plugin_name)
+            )
+
+            data = {
+                'enabled': False,
+                'priority': 0,
+                'plugin_config': {},
+            }
+
+            setting = result.first()
+
+            if setting is not None:
+                data['enabled'] = setting.enabled
+                data['priority'] = setting.priority
+                data['plugin_config'] = setting.config
+
+            return handler.ActionResponse.success(
+                data=data,
+            )
+
+    async def ping(self) -> dict[str, Any]:
+        """Ping the runtime"""
+        return await self.call_action(
+            CommonAction.PING,
+            {},
+            timeout=10,
+        )