fix: Set spec to b"{}" on empty input

cloudquery/sdk/internal/memdb/memdb.py

@@ -1,14 +1,22 @@
+import json
+
 from cloudquery.sdk import plugin
 from cloudquery.sdk import message
 from cloudquery.sdk import schema
 from typing import List, Generator, Dict
 import pyarrow as pa
 from cloudquery.sdk.types import JSONType
+from dataclasses import dataclass, field
 
 NAME = "memdb"
 VERSION = "development"
 
 
+@dataclass
+class Spec:
+    abc: str = field(default="abc")
+
+
 class MemDB(plugin.Plugin):
     def __init__(self) -> None:
         super().__init__(
@@ -73,6 +81,12 @@ def __init__(self) -> None:
             ),
         }
 
+    def init(self, spec, no_connection: bool = False):
+        if no_connection:
+            return
+        self._spec_json = json.loads(spec)
+        self._spec = Spec(**self._spec_json)
+
     def get_tables(self, options: plugin.TableOptions = None) -> List[plugin.Table]:
         tables = list(self._tables.values())
         return schema.filter_dfs(tables, options.tables, options.skip_tables)

cloudquery/sdk/internal/servers/plugin_v3/plugin.py

@@ -32,7 +32,9 @@ def GetSpecSchema(self, request, context):
         return plugin_pb2.GetSpecSchema.Response(json_schema=self._plugin.json_schema())
 
     def Init(self, request: plugin_pb2.Init.Request, context):
-        self._plugin.init(request.spec, no_connection=request.no_connection)
+        self._plugin.init(
+            sanitize_spec(request.spec), no_connection=request.no_connection
+        )
         return plugin_pb2.Init.Response()
 
     def GetTables(self, request: plugin_pb2.GetTables.Request, context):
@@ -113,3 +115,7 @@ def msg_iterator() -> Generator[WriteMessage, None, None]:
     def Close(self, request, context):
         self._plugin.close()
         return plugin_pb2.Close.Response()
+
+
+def sanitize_spec(spec=None):
+    return b"{}" if spec is None or spec == b"" or spec == b"null" else spec

tests/internal/memdb/memdb.py

@@ -1,10 +1,11 @@
 from cloudquery.sdk.internal import memdb
+from cloudquery.sdk.internal.servers.plugin_v3 import plugin
 from cloudquery.sdk.plugin import SyncOptions
 
 
 def test_memdb():
     p = memdb.MemDB()
-    p.init(None)
+    p.init(plugin.sanitize_spec(b"null"))
     msgs = []
     for msg in p.sync(SyncOptions(tables=["*"])):
         msgs.append(msg)