fix: SyncMigrateTableMessage should have the pa.Schema argument named as "table" (#13)

Co-authored-by: Kemal Hadimli <[email protected]>

Author: disq

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

@@ -61,7 +61,7 @@ def Sync(self, request, context):
                     insert=plugin_pb2.Sync.MessageInsert(record=buf)
                 )
             elif isinstance(msg, SyncMigrateTableMessage):
-                buf = arrow.schema_to_bytes(msg.schema)
+                buf = arrow.schema_to_bytes(msg.table)
                 yield plugin_pb2.Sync.Response(
                     migrate_table=plugin_pb2.Sync.MessageMigrateTable(table=buf)
                 )

cloudquery/sdk/message/sync.py

@@ -5,11 +5,11 @@ class SyncMessage:
     pass
 
 
-class SyncInsertMessage:
+class SyncInsertMessage(SyncMessage):
     def __init__(self, record: pa.RecordBatch):
         self.record = record
 
 
-class SyncMigrateTableMessage:
-    def __init__(self, schema: pa.Schema):
-        self.schema = schema
+class SyncMigrateTableMessage(SyncMessage):
+    def __init__(self, table: pa.Schema):
+        self.table = table

cloudquery/sdk/scheduler/scheduler.py

@@ -174,7 +174,7 @@ def sync(
     ) -> Generator[SyncMessage, None, None]:
         res = queue.Queue()
         for resolver in resolvers:
-            yield SyncMigrateTableMessage(schema=resolver.table.to_arrow_schema())
+            yield SyncMigrateTableMessage(table=resolver.table.to_arrow_schema())
         thread = futures.ThreadPoolExecutor()
         thread.submit(self._sync, client, resolvers, res, deterministic_cq_id)
         total_table_resolvers = 0