[AIT-316] feat: introduce support for message annotations

ttypic · ttypic · commit 70701af3f97d · 2026-01-29T16:23:41.000Z
- Added `RealtimeAnnotations` class to manage annotation creation, deletion, and subscription on realtime channels.
- Introduced `Annotation` and `AnnotationAction` types to encapsulate annotation details and actions.
- Extended flags to include `ANNOTATION_PUBLISH` and `ANNOTATION_SUBSCRIBE`.
- Refactored data encoding logic into `ably.util.encoding`.
- Integrated annotation handling into `RealtimeChannel` and `RestChannel`.
diff --git a/ably/realtime/annotations.py b/ably/realtime/annotations.py
@@ -40,7 +40,7 @@ def __init__(self, channel: RealtimeChannel, connection_manager: ConnectionManag
         self.__subscriptions = EventEmitter()
         self.__rest_annotations = RestAnnotations(channel)
 
-    async def publish(self, msg_or_serial, annotation: dict | Annotation, params: dict=None):
+    async def publish(self, msg_or_serial, annotation: dict | Annotation, params: dict | None = None):
         """
         Publish an annotation on a message via the realtime connection.
 
@@ -84,7 +84,12 @@ async def publish(self, msg_or_serial, annotation: dict | Annotation, params: di
         # Send via WebSocket
         await self.__connection_manager.send_protocol_message(protocol_message)
 
-    async def delete(self, msg_or_serial, annotation: dict | Annotation, params=None, timeout=None):
+    async def delete(
+        self,
+        msg_or_serial,
+        annotation: dict | Annotation,
+        params: dict | None = None,
+    ):
         """
         Delete an annotation on a message.
 
@@ -95,7 +100,6 @@ async def delete(self, msg_or_serial, annotation: dict | Annotation, params=None
             msg_or_serial: Either a message serial (string) or a Message object
             annotation: Dict containing annotation properties or Annotation object
             params: Optional dict of query parameters
-            timeout: Optional timeout (not used for realtime, kept for compatibility)
 
         Returns:
             None
@@ -161,13 +165,13 @@ async def subscribe(self, *args):
 
         # Check if ANNOTATION_SUBSCRIBE mode is enabled
         if self.__channel.state == ChannelState.ATTACHED:
-            if not Flag.ANNOTATION_SUBSCRIBE in self.__channel.modes:
+            if Flag.ANNOTATION_SUBSCRIBE not in self.__channel.modes:
                 raise AblyException(
-                    "You are trying to add an annotation listener, but you haven't requested the "
+                    message="You are trying to add an annotation listener, but you haven't requested the "
                     "annotation_subscribe channel mode in ChannelOptions, so this won't do anything "
                     "(we only deliver annotations to clients who have explicitly requested them)",
-                    93001,
-                    400
+                    code=93001,
+                    status_code=400,
                 )
 
     def unsubscribe(self, *args):
@@ -219,7 +223,7 @@ def _process_incoming(self, incoming_annotations):
             annotation_type = annotation.type or ''
             self.__subscriptions._emit(annotation_type, annotation)
 
-    async def get(self, msg_or_serial, params=None):
+    async def get(self, msg_or_serial, params: dict | None = None):
         """
         Retrieve annotations for a message with pagination support.
 
diff --git a/ably/rest/annotations.py b/ably/rest/annotations.py
@@ -9,6 +9,7 @@
 from ably.http.paginatedresult import PaginatedResult, format_params
 from ably.types.annotation import (
     Annotation,
+    AnnotationAction,
     make_annotation_response_handler,
 )
 from ably.types.message import Message
@@ -108,23 +109,27 @@ def __base_path_for_serial(self, serial):
         channel_path = '/channels/{}/'.format(parse.quote_plus(self.__channel.name, safe=':'))
         return channel_path + 'messages/' + parse.quote_plus(serial, safe=':') + '/annotations'
 
-    async def publish(self, msg_or_serial, annotation_values, params=None, timeout=None):
+    async def publish(
+        self,
+        msg_or_serial,
+        annotation: dict | Annotation,
+        params: dict | None = None,
+    ):
         """
         Publish an annotation on a message.
 
         Args:
             msg_or_serial: Either a message serial (string) or a Message object
-            annotation_values: Dict containing annotation properties (type, name, data, etc.)
+            annotation: Dict containing annotation properties (type, name, data, etc.) or Annotation object
             params: Optional dict of query parameters
-            timeout: Optional timeout for the HTTP request
 
         Returns:
             None
 
         Raises:
             AblyException: If the request fails or inputs are invalid
         """
-        annotation = construct_validate_annotation(msg_or_serial, annotation_values)
+        annotation = construct_validate_annotation(msg_or_serial, annotation)
 
         # Convert to wire format
         request_body = annotation.as_dict(binary=self.__channel.ably.options.use_binary_protocol)
@@ -145,9 +150,14 @@ async def publish(self, msg_or_serial, annotation_values, params=None, timeout=N
             path += '?' + parse.urlencode(params)
 
         # Send request
-        await self.__channel.ably.http.post(path, body=request_body, timeout=timeout)
-
-    async def delete(self, msg_or_serial, annotation_values, params=None, timeout=None):
+        await self.__channel.ably.http.post(path, body=request_body)
+
+    async def delete(
+        self,
+        msg_or_serial,
+        annotation: dict | Annotation,
+        params: dict | None = None,
+    ):
         """
         Delete an annotation on a message.
 
@@ -156,9 +166,8 @@ async def delete(self, msg_or_serial, annotation_values, params=None, timeout=No
 
         Args:
             msg_or_serial: Either a message serial (string) or a Message object
-            annotation_values: Dict containing annotation properties
+            annotation: Dict containing annotation properties or Annotation object
             params: Optional dict of query parameters
-            timeout: Optional timeout for the HTTP request
 
         Returns:
             None
@@ -167,11 +176,14 @@ async def delete(self, msg_or_serial, annotation_values, params=None, timeout=No
             AblyException: If the request fails or inputs are invalid
         """
         # Set action to delete
-        annotation_values = annotation_values.copy()
-        annotation_values['action'] = 'annotation.delete'
-        return await self.publish(msg_or_serial, annotation_values, params, timeout)
+        if isinstance(annotation, Annotation):
+            annotation_values = annotation.as_dict()
+        else:
+            annotation_values = annotation.copy()
+        annotation_values['action'] = AnnotationAction.ANNOTATION_DELETE
+        return await self.publish(msg_or_serial, annotation_values, params)
 
-    async def get(self, msg_or_serial, params=None):
+    async def get(self, msg_or_serial, params: dict | None = None):
         """
         Retrieve annotations for a message with pagination support.
 
diff --git a/test/ably/realtime/realtimeannotations_test.py b/test/ably/realtime/realtimeannotations_test.py
@@ -5,11 +5,10 @@
 
 from ably import AblyException
 from ably.types.annotation import AnnotationAction
+from ably.types.channelmode import ChannelMode
 from ably.types.channeloptions import ChannelOptions
-from ably.types.message import MessageAction
 from test.ably.testapp import TestApp
 from test.ably.utils import BaseAsyncTestCase, assert_waiter
-from ably.types.channelmode import ChannelMode
 
 log = logging.getLogger(__name__)
 
@@ -20,11 +19,17 @@ class TestRealtimeAnnotations(BaseAsyncTestCase):
     @pytest.fixture(autouse=True)
     async def setup(self, transport):
         self.test_vars = await TestApp.get_test_vars()
+        # Use clientId to match JS tests (required for reaction:distinct.v1 annotation type)
+        import random
+        import string
+        client_id = ''.join(random.choices(string.ascii_letters + string.digits, k=10))
         self.ably = await TestApp.get_ably_realtime(
             use_binary_protocol=True if transport == 'msgpack' else False,
+            client_id=client_id,
         )
         self.rest = await TestApp.get_ably_rest(
             use_binary_protocol=True if transport == 'msgpack' else False,
+            client_id=client_id,
         )
 
     async def test_publish_and_subscribe_annotations(self):
@@ -39,7 +44,6 @@ async def test_publish_and_subscribe_annotations(self):
             self.get_channel_name('mutable:publish_subscribe_annotation'),
             channel_options
         )
-        rest_channel = self.rest.channels[channel.name]
         await channel.attach()
 
         # Setup annotation listener
@@ -65,15 +69,15 @@ def on_message(msg):
 
         # Publish annotation using realtime
         await channel.annotations.publish(publish_result.serials[0], {
-            'type': 'reaction:multiple.v1',
+            'type': 'reaction:distinct.v1',
             'name': '👍'
         })
 
         # Wait for annotation
         annotation = await annotation_future
         assert annotation.action == AnnotationAction.ANNOTATION_CREATE
         assert annotation.message_serial == publish_result.serials[0]
-        assert annotation.type == 'reaction:multiple.v1'
+        assert annotation.type == 'reaction:distinct.v1'
         assert annotation.name == '👍'
         assert annotation.serial > annotation.message_serial
 
@@ -92,22 +96,25 @@ def on_message(msg):
         # await channel.annotations.subscribe(on_annotation2)
         #
         # await rest_channel.annotations.publish(publish_result.serials[0], {
-        #     'type': 'reaction:multiple.v1',
+        #     'type': 'reaction:distinct.v1',
         #     'name': '😕'
         # })
         #
         # annotation = await annotation_future2
         # assert annotation.action == AnnotationAction.ANNOTATION_CREATE
         # assert annotation.message_serial == publish_result.serials[0]
-        # assert annotation.type == 'reaction:multiple.v1'
+        # assert annotation.type == 'reaction:distinct.v1'
         # assert annotation.name == '😕'
         # assert annotation.serial > annotation.message_serial
 
     async def test_get_all_annotations_for_a_message(self):
         """Test retrieving all annotations with pagination (matches JS test)"""
-        channel_options = ChannelOptions(params={
-            'modes': 'publish,subscribe,annotation_publish,annotation_subscribe'
-        })
+        channel_options = ChannelOptions(modes=[
+            ChannelMode.PUBLISH,
+            ChannelMode.SUBSCRIBE,
+            ChannelMode.ANNOTATION_PUBLISH,
+            ChannelMode.ANNOTATION_SUBSCRIBE
+        ])
         channel = self.ably.channels.get(
             self.get_channel_name('mutable:get_all_annotations_for_a_message'),
             channel_options
@@ -131,7 +138,7 @@ def on_message(msg):
         emojis = ['👍', '😕', '👎', '👍👍', '😕😕', '👎👎']
         for emoji in emojis:
             await channel.annotations.publish(message.serial, {
-                'type': 'reaction:multiple.v1',
+                'type': 'reaction:distinct.v1',
                 'name': emoji
             })
 
@@ -149,7 +156,7 @@ async def check_annotations():
         # Verify annotations
         assert annotations[0].action == AnnotationAction.ANNOTATION_CREATE
         assert annotations[0].message_serial == message.serial
-        assert annotations[0].type == 'reaction:multiple.v1'
+        assert annotations[0].type == 'reaction:distinct.v1'
         assert annotations[0].name == '👍'
         assert annotations[1].name == '😕'
         assert annotations[2].name == '👎'
@@ -176,9 +183,12 @@ async def check_annotations():
 
     async def test_subscribe_by_annotation_type(self):
         """Test subscribing to specific annotation types"""
-        channel_options = ChannelOptions(params={
-            'modes': 'publish,subscribe,annotation_publish,annotation_subscribe'
-        })
+        channel_options = ChannelOptions(modes=[
+            ChannelMode.PUBLISH,
+            ChannelMode.SUBSCRIBE,
+            ChannelMode.ANNOTATION_PUBLISH,
+            ChannelMode.ANNOTATION_SUBSCRIBE
+        ])
         channel = self.ably.channels.get(
             self.get_channel_name('mutable:subscribe_by_type'),
             channel_options
@@ -201,30 +211,31 @@ async def on_reaction(annotation):
             if not reaction_future.done():
                 reaction_future.set_result(annotation)
 
-        await channel.annotations.subscribe('reaction:multiple.v1', on_reaction)
+        await channel.annotations.subscribe('reaction:distinct.v1', on_reaction)
 
         # Publish message and annotation
         await channel.publish('message', 'test')
         message = await message_future
 
-        # Temporary anti-flake measure (matches JS test)
-        await asyncio.sleep(1)
 
         await channel.annotations.publish(message.serial, {
-            'type': 'reaction:multiple.v1',
+            'type': 'reaction:distinct.v1',
             'name': '👍'
         })
 
         # Should receive the annotation
         annotation = await reaction_future
-        assert annotation.type == 'reaction:multiple.v1'
+        assert annotation.type == 'reaction:distinct.v1'
         assert annotation.name == '👍'
 
     async def test_unsubscribe_annotations(self):
         """Test unsubscribing from annotations"""
-        channel_options = ChannelOptions(params={
-            'modes': 'publish,subscribe,annotation_publish,annotation_subscribe'
-        })
+        channel_options = ChannelOptions(modes=[
+            ChannelMode.PUBLISH,
+            ChannelMode.SUBSCRIBE,
+            ChannelMode.ANNOTATION_PUBLISH,
+            ChannelMode.ANNOTATION_SUBSCRIBE
+        ])
         channel = self.ably.channels.get(
             self.get_channel_name('mutable:unsubscribe_annotations'),
             channel_options
@@ -251,11 +262,8 @@ async def on_annotation(annotation):
         await channel.publish('message', 'test')
         message = await message_future
 
-        # Temporary anti-flake measure (matches JS test)
-        await asyncio.sleep(1)
-
         await channel.annotations.publish(message.serial, {
-            'type': 'reaction:multiple.v1',
+            'type': 'reaction:distinct.v1',
             'name': '👍'
         })
 
@@ -267,7 +275,7 @@ async def on_annotation(annotation):
 
         # Publish another annotation
         await channel.annotations.publish(message.serial, {
-            'type': 'reaction:multiple.v1',
+            'type': 'reaction:distinct.v1',
             'name': '😕'
         })
 
@@ -276,9 +284,12 @@ async def on_annotation(annotation):
 
     async def test_delete_annotation(self):
         """Test deleting annotations"""
-        channel_options = ChannelOptions(params={
-            'modes': 'publish,subscribe,annotation_publish,annotation_subscribe'
-        })
+        channel_options = ChannelOptions(modes=[
+            ChannelMode.PUBLISH,
+            ChannelMode.SUBSCRIBE,
+            ChannelMode.ANNOTATION_PUBLISH,
+            ChannelMode.ANNOTATION_SUBSCRIBE
+        ])
         channel = self.ably.channels.get(
             self.get_channel_name('mutable:delete_annotation'),
             channel_options
@@ -305,11 +316,8 @@ async def on_annotation(annotation):
         await channel.publish('message', 'test')
         message = await message_future
 
-        # Temporary anti-flake measure (matches JS test)
-        await asyncio.sleep(1)
-
         await channel.annotations.publish(message.serial, {
-            'type': 'reaction:multiple.v1',
+            'type': 'reaction:distinct.v1',
             'name': '👍'
         })
 
@@ -319,7 +327,7 @@ async def on_annotation(annotation):
 
         # Delete the annotation
         await channel.annotations.delete(message.serial, {
-            'type': 'reaction:multiple.v1',
+            'type': 'reaction:distinct.v1',
             'name': '👍'
         })
 
@@ -330,9 +338,10 @@ async def on_annotation(annotation):
     async def test_subscribe_without_annotation_mode_fails(self):
         """Test that subscribing without annotation_subscribe mode raises an error"""
         # Create channel without annotation_subscribe mode
-        channel_options = ChannelOptions(params={
-            'modes': 'publish,subscribe'
-        })
+        channel_options = ChannelOptions(modes=[
+            ChannelMode.PUBLISH,
+            ChannelMode.SUBSCRIBE
+        ])
         channel = self.ably.channels.get(
             self.get_channel_name('mutable:no_annotation_mode'),
             channel_options
diff --git a/test/ably/rest/restannotations_test.py b/test/ably/rest/restannotations_test.py
