Fix reading rows when using interactive (streaming) queries (#15)

Author: Kayrnt

.changes/unreleased/Fixes-20251030-001010.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix reading rows when using interactive (streaming) queries
+time: 2025-10-30T00:10:10.751126+01:00
+custom:
+    Author: Kayrnt
+    Issue: ""

src/deltastream/api/rows.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import List, Optional, Any
+from typing import List, Optional, Any, Union
 import re
 import json
 from decimal import Decimal
@@ -22,13 +22,13 @@ class Column:
 
 
 def castRowData(
-    row_data_strings: List[ResultSetDataInnerInner], columns: List[Column]
+    row_data_strings: List[Union[ResultSetDataInnerInner, Any]], columns: List[Column]
 ) -> List[Any]:
     """
     Cast row data strings to their appropriate Python types based on column definitions.
 
     Args:
-        row_data_strings: List of string values or None from the database
+        row_data_strings: List of ResultSetDataInnerInner objects or plain values from WebSocket
         columns: List of Column objects defining the data types
 
     Returns:
@@ -38,7 +38,13 @@ def castRowData(
 
     for i, column in enumerate(columns):
         row_data_string = row_data_strings[i]
-        value = row_data_string.actual_instance
+        # Handle both ResultSetDataInnerInner objects and plain values (from WebSocket)
+        value: Any
+        if hasattr(row_data_string, "actual_instance"):
+            value = row_data_string.actual_instance
+        else:
+            value = row_data_string
+
         if value is None:
             row_data.append(None)
             continue

src/deltastream/api/streaming_rows.py

@@ -68,7 +68,9 @@ async def open(self) -> None:
 
         async def handle_messages():
             try:
-                ws = await ws_connect(self.req.uri)
+                # Convert HTTPS URL to WSS for WebSocket connection
+                ws_uri = self.req.uri.replace("https://", "wss://", 1)
+                ws = await ws_connect(ws_uri)
                 self.ws = ws
 
                 # Send authentication
@@ -86,7 +88,31 @@ async def handle_messages():
                     data = json.loads(message)
 
                     if data["type"] == "metadata":
-                        self.metadata = PrintTopicMetadata(**data)
+                        # Convert column dictionaries to Column objects
+                        columns = []
+                        for col in data.get("columns", []):
+                            if isinstance(col, dict):
+                                # Provide default value for nullable if missing
+                                columns.append(
+                                    Column(
+                                        name=col.get("name", ""),
+                                        type=col.get("type", ""),
+                                        nullable=col.get(
+                                            "nullable", True
+                                        ),  # Default to True if not specified
+                                        length=col.get("length"),
+                                        precision=col.get("precision"),
+                                        scale=col.get("scale"),
+                                    )
+                                )
+                            else:
+                                columns.append(col)
+
+                        self.metadata = PrintTopicMetadata(
+                            type=data["type"],
+                            headers=data.get("headers", {}),
+                            columns=columns,
+                        )
                         deferred_ready.resolve()
 
                     elif data["type"] == "data":
@@ -124,17 +150,7 @@ def columns(self) -> List[Column]:
         if self.metadata is None:
             return []
 
-        return [
-            Column(
-                name=column.name,
-                type=column.type,
-                nullable=column.nullable,
-                length=column.length,
-                precision=column.precision,
-                scale=column.scale,
-            )
-            for column in self.metadata.columns
-        ]
+        return self.metadata.columns
 
     async def close(self) -> None:
         """Closes the WebSocket connection."""