Merge pull request #1203 from danielballan/patch-query-params

Move array patch specification from `PUT` body to query parameters

Author: genematx

CHANGELOG.md

@@ -9,20 +9,45 @@ Write the date in place of the "Unreleased" in the case a new version is release
 
 - Tests to ensure that CSVAdapter can be used with a subset of columns.
 - `locking` key-word argument in HDFAdapter and HDF5Adapter.
+- Support for streaming uploaded tabular data.
 
 ### Changed
 
 - Enable Tiled server to accept bearer access tokens for authentication.
 - Modernize implementation of Device Code Flow.
+- The websocket messages contain a `"type"` field and are not formally
+  specified and validated using pydantic models in `tiled.stream_messages`.
 - In previous releases, the method `Subscription.start` launched a background
   thread. This was renamed to `Subscription.start_in_thread`. Now
   `Subscription.start` blocks the current thread, leaving any thread management
   up to the caller.
 - Encapsulated redis code in adapter.py into a StreamingCache object
-- Renamed `Subscription.stop` to `Subscription.close` to match the property
-  `Subscription.closed`.
+- Renamed `Subscription.stop` to `Subscription.disconnect`.
 - In `Subscription`, use a configurable `concurrent.futures.Executor` to
   execute callbacks.
+- Removed `Subscription.add_callback`, replaced by separate callback registries
+  connected to specific types of updates:
+
+  ```python
+  Subscription.stream_closed  # writer-initiated
+  Subscription.disconnected  # subscriber-initiated
+  ContainerSubscription.child_created
+  ContainerSubscription.child_metadata_updated
+  ArraySubscription.new_data
+  TableSubscription.new_data
+  ```
+
+- The signature of subscription callbacks has been changed. The
+  connection-related callbacks `stream_closed` and `disconnected`
+  receive `f(subscription: Subscription)`. The other callbacks
+  receive an instance of `tiled.client.stream.LiveUpdate`,
+  which provide a reference to the subscription
+  `update.subscription`, all the data from the websocket message,
+  and update-specific convenience methods such as
+  `LiveContainerUpdate.child()` when a new child is created in a
+  container and `ArraySubscription.new_data()` or
+  `TableSubscription.new_data()` to conveniently access an array
+  or table respectively.
 
 ### Fixed
 

docs/source/tutorials/streaming.md

@@ -5,8 +5,10 @@
 Streaming is an experimental feature in Tiled, subject to backward-compatible
 changes without notice.
 
-Additionally, it is only currently supported on arrays and containers.
-Support for tables and other structures is planned but not yet implemented.
+Additionally, it is only currently supported on containers, arrays (both
+uploaded and external registered data), and tables (uploaded data only).
+Support for external registered tabular data and other structures (sparse,
+awkward) is planned but not yet implemented.
 
 ```
 
@@ -177,5 +179,6 @@ serve` CLI they are `--cache-data-ttl` and `--cache-seq-ttl`.)
 
 This feature is in a very early preview stage.
 
-- Other data structures (table, sparse, awkward) are not yet supported.
-- Updates to metadata and deletions of nodes are not yet visible to subscribers.
+- External registered tabular data is not yet supported.
+- Other data structures (sparse, awkward) are not yet supported.
+- Deletions of nodes are not yet visible to subscribers.

tiled/_tests/test_subscription.py

@@ -17,6 +17,7 @@
 from ..structures.core import StructureFamily
 from ..structures.data_source import Asset, DataSource, Management
 from ..utils import safe_json_dump
+from .utils import fail_with_status_code
 
 pytestmark = pytest.mark.skipif(
     sys.platform == "win32", reason="Requires Redis service"
@@ -377,14 +378,27 @@ def on_child_created(update):
                 num=3,
             ),
         )
+        params = {
+            "patch_shape": ",".join(map(str, [1, 7, 13])),
+            "patch_offset": ",".join(map(str, [2, 0, 0])),
+        }
+
+        # First test invalid requests that will be bounced by the server.
+        for key in params:
+            bad_params = params.copy()
+            bad_params.pop(key)  # missing one of the patch params!
+            with fail_with_status_code(400):
+                x.context.http_client.put(
+                    x.uri.replace("/metadata/", "/data_source/", 1),
+                    content=safe_json_dump({"data_source": updated_data_source}),
+                    params=bad_params,
+                ).raise_for_status()
+
+        # Now do a request that is valid.
         x.context.http_client.put(
             x.uri.replace("/metadata/", "/data_source/", 1),
-            content=safe_json_dump(
-                {
-                    "data_source": updated_data_source,
-                    "patch": {"shape": (1, 7, 13), "offset": (2, 0, 0), "extend": True},
-                }
-            ),
+            content=safe_json_dump({"data_source": updated_data_source}),
+            params=params,
         ).raise_for_status()
         assert event.wait(timeout=5.0), "Timeout waiting for messages"
         x.close_stream()
@@ -395,7 +409,6 @@ def on_child_created(update):
     (update,) = updates
     assert update.patch.shape == (1, 7, 13)
     assert update.patch.offset == (2, 0, 0)
-    assert update.patch.extend
     actual_streamed = update.data()
     np.testing.assert_array_equal(actual_streamed, arr[2:])
 
@@ -519,7 +532,7 @@ def collect(update):
         assert_frame_equal(actual_updated, df2)
 
 
-def test_streaming_table_appends(tiled_websocket_context):
+def test_streaming_table_append(tiled_websocket_context):
     context = tiled_websocket_context
     client = from_context(context)
     updates = []

tiled/client/stream.py

@@ -655,11 +655,11 @@ class LiveTableData(TableData):
     def data(self):
         "Get table"
         # Registration occurs on import. Ensure this is imported.
-        from ..serialization import array
+        from ..serialization import table
 
-        del array
+        del table
 
-        # Decode payload (bytes) into array.
+        # Decode payload (bytes) into table.
         deserializer = default_deserialization_registry.dispatch("table", self.mimetype)
         return deserializer(self.payload)
 

tiled/server/dependencies.py

@@ -160,3 +160,25 @@ def offset_param(
 ):
     "Specify and parse an offset parameter."
     return tuple(map(int, offset.split(",")))
+
+
+def patch_shape_param(
+    patch_shape: Optional[str] = Query(
+        None, min_length=1, pattern="^[0-9]+(,[0-9]+)*$|^scalar$"
+    ),
+):
+    "Specify and parse an array patch shape parameter."
+    if patch_shape is None:
+        return None
+    return tuple(map(int, patch_shape.split(",")))
+
+
+def patch_offset_param(
+    patch_offset: Optional[str] = Query(
+        None, min_length=1, pattern="^[0-9]+(,[0-9]+)*$"
+    ),
+):
+    "Specify and parse an array patch offset parameter."
+    if patch_offset is None:
+        return None
+    return tuple(map(int, patch_offset.split(",")))

tiled/server/router.py

@@ -52,6 +52,7 @@
 from .. import __version__
 from ..links import links_for_node
 from ..ndslice import NDSlice
+from ..stream_messages import ArrayPatch
 from ..structures.core import Spec, StructureFamily
 from ..type_aliases import AccessTags, Scopes
 from ..utils import BrokenLink, ensure_awaitable, patch_mimetypes, path_from_uri
@@ -89,6 +90,8 @@
     get_entry,
     get_root_tree,
     offset_param,
+    patch_offset_param,
+    patch_shape_param,
     shape_param,
 )
 from .file_response_with_range import FileResponseWithRange
@@ -1636,6 +1639,8 @@ async def put_data_source(
         session_state: dict = Depends(get_session_state),
         authn_access_tags: Optional[AccessTags] = Depends(get_current_access_tags),
         authn_scopes: Scopes = Depends(get_current_scopes),
+        patch_shape: Optional[tuple[int, ...]] = Depends(patch_shape_param),
+        patch_offset: Optional[tuple[int, ...]] = Depends(patch_offset_param),
         _=Security(check_scopes, scopes=["write:metadata", "register"]),
     ):
         entry = await get_entry(
@@ -1650,7 +1655,23 @@ async def put_data_source(
             None,
             getattr(request.app.state, "access_policy", None),
         )
-        await entry.put_data_source(data_source=body.data_source, patch=body.patch)
+        patch_params = {
+            "shape": patch_shape,
+            "offset": patch_offset,
+        }
+        if all(value is None for value in patch_params.values()):
+            patch = None
+        elif all(value is not None for value in patch_params.values()):
+            patch = ArrayPatch(**patch_params)
+        else:
+            raise HTTPException(
+                status_code=HTTP_400_BAD_REQUEST,
+                detail=(
+                    "The query parameters patch_shape and patch_offset"
+                    "go together; either all or none must be specified."
+                ),
+            )
+        await entry.put_data_source(data_source=body.data_source, patch=patch)
 
     @router.delete("/metadata/{path:path}")
     async def delete(

tiled/server/schemas.py

@@ -3,17 +3,7 @@
 import enum
 import uuid
 from datetime import datetime
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    Dict,
-    Generic,
-    List,
-    Optional,
-    Tuple,
-    TypeVar,
-    Union,
-)
+from typing import TYPE_CHECKING, Any, Dict, Generic, List, Optional, TypeVar, Union
 
 import pydantic.generics
 from pydantic import ConfigDict, Field, StringConstraints
@@ -153,12 +143,6 @@ def from_orm(cls, orm: tiled.catalog.orm.Revision) -> Revision:
         )
 
 
-class Patch(pydantic.BaseModel):
-    offset: Tuple[int, ...]
-    shape: Tuple[int, ...]
-    extend: bool
-
-
 class DataSource(pydantic.BaseModel, Generic[StructureT]):
     id: Optional[int] = None
     structure_family: StructureFamily
@@ -465,7 +449,6 @@ def narrow_structure_type(self):
 
 class PutDataSourceRequest(pydantic.BaseModel):
     data_source: DataSource
-    patch: Optional[Patch] = None
 
 
 class PostMetadataResponse(pydantic.BaseModel, Generic[ResourceLinksT]):

tiled/stream_messages.py

@@ -88,7 +88,6 @@ class ArrayData(Update):
 class ArrayPatch(BaseModel):
     offset: tuple[int, ...]
     shape: tuple[int, ...]
-    extend: bool
 
 
 class ArrayRef(Update):