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

- 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`.

Author: ttypic

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.
 

ably/realtime/channel.py

@@ -911,6 +911,10 @@ def presence(self):
         """Get the RealtimePresence object for this channel"""
         return self.__presence
 
+    @property
+    def annotations(self) -> RealtimeAnnotations:
+        return self._Channel__annotations
+
     @property
     def modes(self):
         """Get the list of channel modes"""

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
@@ -74,7 +75,7 @@ def construct_validate_annotation(msg_or_serial, annotation: dict | Annotation):
     elif isinstance(annotation, Annotation):
         annotation_values = annotation.as_dict()
     else:
-        annotation_values = annotation
+        annotation_values = annotation.copy()
 
     annotation_values['message_serial'] = message_serial
 
@@ -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.
 

ably/rest/auth.py

@@ -90,7 +90,7 @@ def __init__(self, ably: AblyRest | AblyRealtime, options: Options):
     async def get_auth_transport_param(self):
         auth_credentials = {}
         if self.auth_options.client_id:
-            auth_credentials["client_id"] = self.auth_options.client_id
+            auth_credentials["clientId"] = self.auth_options.client_id
         if self.__auth_mechanism == Auth.Method.BASIC:
             key_name = self.__auth_options.key_name
             key_secret = self.__auth_options.key_secret

ably/rest/channel.py

@@ -26,11 +26,14 @@
     IncompatibleClientIdException,
     catch_all,
 )
+from ably.rest.annotations import RestAnnotations
 
 log = logging.getLogger(__name__)
 
 
 class Channel:
+    __annotations: RestAnnotations
+
     def __init__(self, ably, name, options):
         self.__ably = ably
         self.__name = name
@@ -366,7 +369,7 @@ def presence(self):
         return self.__presence
 
     @property
-    def annotations(self):
+    def annotations(self) -> RestAnnotations:
         return self.__annotations
 
     @options.setter

ably/transport/websockettransport.py

@@ -189,6 +189,7 @@ async def on_protocol_message(self, msg):
             ProtocolMessageAction.DETACHED,
             ProtocolMessageAction.MESSAGE,
             ProtocolMessageAction.PRESENCE,
+            ProtocolMessageAction.ANNOTATION,
             ProtocolMessageAction.SYNC
         ):
             self.connection_manager.on_channel_message(msg)

ably/types/annotation.py

@@ -122,22 +122,17 @@ def as_dict(self, binary=False):
 
         Note: Annotations are not encrypted as they need to be parsed by the server.
         """
-        # Encode data
-        encoded = encode_data(self.data, self._encoding_array, binary)
-
         request_body = {
             'action': int(self.action) if self.action is not None else None,
             'serial': self.serial,
             'messageSerial': self.message_serial,
             'type': self.type,  # Annotation type (not data type)
             'name': self.name,
             'count': self.count,
-            'data': encoded.get('data'),
-            'encoding': encoded.get('encoding', ''),
-            'dataType': encoded.get('type'),  # Data type (not annotation type)
             'clientId': self.client_id or None,
             'timestamp': self.timestamp or None,
             'extras': self.extras,
+            **encode_data(self.data, self._encoding_array, binary)
         }
 
         # None values aren't included