Merge pull request #413 from binance/release_common_v2.0.0

Release common v2.0.0

Author: alplabin

common/CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## 2.0.0 - 2025-08-22
+
+### Changed (3)
+
+- Added custom REST headers
+- Added `subscribe_user_data`, `on` and `unsubscribe` method to `WebSocketAPIBase`
+- Updated `RequestStream` response type to `RequestStreamHandle`
+
 ## 1.2.0 - 2025-08-07
 
 ### Changed (1)

common/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "binance-common"
-version = "1.2.0"
+version = "2.0.0"
 description = "Binance Common Types and Utilities for Binance Connectors"
 authors = ["Binance"]
 license = "MIT"

common/src/binance_common/configuration.py

@@ -1,9 +1,10 @@
 import ssl
 
-from typing import Optional, Dict, Union
+from typing import Optional, Dict, Union, List
 from http.client import HTTPSConnection
 
 from binance_common.constants import TimeUnit, WebsocketMode
+from binance_common.headers import parse_custom_headers
 
 
 class ConfigurationRestAPI:
@@ -19,6 +20,7 @@ class ConfigurationRestAPI:
     - Time Unit
     - Proxy support
     - Retries & Backoff
+    - Custom REST headers
     """
 
     def __init__(
@@ -36,6 +38,7 @@ def __init__(
         time_unit: Optional[str] = None,
         private_key: Optional[Union[bytes, str]] = None,
         private_key_passphrase: Optional[str] = None,
+        custom_headers: Optional[dict[str, Union[str, List[str]]]] = {},
     ):
         """
         Initialize the API configuration.
@@ -54,6 +57,7 @@ def __init__(
             time_unit (Optional[str]): Time unit for time-based responses (default: None).
             private_key (Optional[Union[bytes, str]]): Private key for authentication (default: None).
             private_key_passphrase (Optional[str]): Passphrase for private key (default: None).
+            custom_headers (Optional[dict[str, Union[str, List[str]]]]): Custom REST headers (default: {}).
         """
 
         self.api_key = api_key
@@ -73,6 +77,7 @@ def __init__(
         self.base_headers = {
             "Accept": "application/json",
             "X-MBX-APIKEY": str(self.api_key) if self.api_key else "",
+            **parse_custom_headers(custom_headers)
         }
 
 

common/src/binance_common/headers.py

@@ -0,0 +1,53 @@
+import logging
+
+from typing import List, Optional, Union
+
+def sanitize_header_value(value: Union[str, List[str]]) -> Union[str, List[str]]:
+    """Sanitizes a header value by checking for and preventing carriage return and line feed characters.
+
+    Args:
+        value (Union[str, List[str]]): The header value or array of header values to sanitize.
+
+    Raises:
+        ValueError: If the header value contains CR or LF characters.
+
+    Returns:
+        Union[str, List[str]]: The sanitized header value(s).
+    """
+
+    if "\r" in value or "\n" in value:
+        raise ValueError(f'Invalid header value (contains CR/LF): "{value}"')
+    return value
+
+def parse_custom_headers(custom_headers: Optional[dict[str, Union[str, List[str]]]]) -> dict[str, Union[str, List[str]]]:
+    """Parses custom headers for the API client.
+
+    Args:
+        custom_headers (Optional[dict[str, Union[str, List[str]]]]): A dictionary of custom headers to parse.
+
+    Returns:
+        dict[str, Union[str, List[str]]]: A dictionary of parsed custom headers.
+    """
+
+    forbidden_headers = {"host", "authorization", "cookie", ":method", ":path"}
+    parsed_headers = {}
+
+    if not custom_headers:
+        return {}
+
+    for key, value in custom_headers.items():
+        header_name = key.strip()
+        if header_name.lower() in forbidden_headers:
+            logging.warning(f"Header '{header_name}' is not allowed to be set.")
+            continue
+
+        try:
+            if isinstance(value, list):
+                parsed_headers[header_name] = [sanitize_header_value(v) for v in value]
+            else:
+                parsed_headers[header_name] = sanitize_header_value(value)
+        except ValueError as e:
+            logging.warning(f"Dropping header '{header_name}' due to invalid value.")
+            continue
+
+    return parsed_headers

common/src/binance_common/models.py

@@ -2,7 +2,8 @@
 from pydantic import BaseModel
 
 T = TypeVar("T")
-
+T_Response = TypeVar("T_Response")
+T_Stream = TypeVar("T_Stream")
 
 class RateLimit(BaseModel):
     """Represents a single rate limit entry.
@@ -90,3 +91,19 @@ def data(self) -> T:
         :return: The parsed data of type T.
         """
         return self._data_function()
+
+
+class WebsocketApiUserDataStreamResponse(Generic[T_Response, T_Stream]):
+    """A wrapper for WebSocket API user data stream responses.
+
+    :param response: A callable that lazily returns a WebSocket API responses of type T_Response.
+    :param stream: A callable that lazily returns a RequestStreamHandle responses of type T_Stream.
+    """
+
+    def __init__(
+        self,
+        response: WebsocketApiResponse[T_Response],
+        stream: T_Stream,
+    ):
+        self.response = response
+        self.stream = stream

common/src/binance_common/utils.py

@@ -103,8 +103,8 @@ def snake_to_camel(snake_str: str) -> str:
         str: The converted camelCase string.
     """
 
-    parts = snake_str.split('_')
-    return parts[0] + ''.join(word.capitalize() for word in parts[1:])
+    parts = snake_str.split("_")
+    return parts[0] + "".join(word.capitalize() for word in parts[1:])
 
 
 def make_serializable(val) -> Union[dict, list, str, int, float, bool]:
@@ -117,7 +117,7 @@ def make_serializable(val) -> Union[dict, list, str, int, float, bool]:
     """
 
     if isinstance(val, list):
-        return [v.__dict__ if hasattr(v, '__dict__') else v for v in val]
+        return [v.__dict__ if hasattr(v, "__dict__") else v for v in val]
     if isinstance(val, bool):
         return str(val).lower()
     if isinstance(val , Enum):
@@ -544,9 +544,9 @@ def convert(val, expected_type=None):
             elif expected_type == str:
                 return val_stripped
         val_lower = val_stripped.lower()
-        if val_lower == 'true':
+        if val_lower == "true":
             return True
-        elif val_lower == 'false':
+        elif val_lower == "false":
             return False
         try:
             return int(val_stripped)

common/src/binance_common/websocket.py

@@ -6,7 +6,7 @@
 
 from collections import OrderedDict
 from pydantic import BaseModel
-from typing import Callable, Optional, Dict, Union, TypeVar, Type
+from typing import Callable, Optional, Dict, Generic, Union, TypeVar, Type
 from types import SimpleNamespace
 from urllib.parse import urlencode
 
@@ -193,29 +193,35 @@ async def receive_loop(self, connection: WebSocketConnection):
                         raise ValueError(f"Error received from server: {data['error']}")
 
                     stream = data.get("stream")
-                    callbacks = (
-                        connection.stream_callback_map.get(stream) if stream else None
-                    )
-                    response_model = (
-                        connection.response_types.get(stream) if stream else None
-                    )
+                    subscription_id = data.get("subscriptionId")
+
+                    key = stream or subscription_id
+                    callbacks = connection.stream_callback_map.get(key) if key is not None else None
+
                     if callbacks:
                         try:
-                            for callback in callbacks:
-                                if response_model is None:
-                                    callback(data)
-                                else:
-                                    data = data["data"]
-                                    if isinstance(data, list):
-                                        callback(
-                                            [response_model.model_validate_json(json.dumps(item)) for item in data]
-                                        )
+                            if stream:
+                                response_model = connection.response_types.get(stream)
+                                payload = data["data"] if response_model else data
+
+                                for callback in callbacks:
+                                    if response_model:
+                                        if isinstance(payload, list):
+                                            parsed = [
+                                                response_model.model_validate_json(json.dumps(item))
+                                                for item in payload
+                                            ]
+                                            callback(parsed)
+                                        else:
+                                            callback(response_model.model_validate_json(json.dumps(payload)))
                                     else:
-                                        callback(response_model.model_validate_json(json.dumps(data)))
+                                        callback(payload)
+                            else:
+                                payload = data["event"]
+                                for callback in callbacks:
+                                    callback(payload)
                         except Exception as e:
-                            raise ValueError(
-                                f"Error in callback for stream {stream}: {e}"
-                            )
+                            raise ValueError(f"Error in callback for key {key}: {e}")
                     else:
                         logging.info(f"Received message: {data}")
             elif msg.type == aiohttp.WSMsgType.PING:
@@ -723,22 +729,111 @@ async def ping_ws_api(self, connection: WebSocketConnection):
 
         await super().ping(connection)
 
+    async def subscribe_user_data(self, id: str):
+        if self.configuration.mode == WebsocketMode.SINGLE:
+            connection = self.connections[0]
+        else:
+            connection = self.connections[
+                self.round_robin_index % len(self.connections)
+            ]
+            self.round_robin_index = (self.round_robin_index + 1) % len(
+                self.connections
+            )
+        global_stream_connections.stream_connections_map[id] = connection
+        connection.stream_callback_map.update({id: []})
+
+    def on(self, event: str, callback: Callable[[T], None], id: str) -> None:
+        """Set the callback function for incoming messages on a specific ID.
+
+        Args:
+            event (str): Event type.
+            callback (Callable): Callback function.
+            id (str): User Data ID.
+        """
+
+        if event != "message":
+            raise ValueError(f"Unsupported event: {event}")
+
+        connection = (
+            global_stream_connections.stream_connections_map[id]
+            if id in global_stream_connections.stream_connections_map
+            else None
+        )
+
+        if connection:
+            connection.stream_callback_map[id].append(callback)
+        else:
+            logging.warning(f"Stream {id} not connected.")
+
+    async def unsubscribe(self, id: str):
+        """Unsubscribe from a user data ID.
+
+        Args:
+            id (str): user data ID to unsubscribe from.
+        """
+
+        if self.connections is None or len(self.connections) == 0:
+            logging.warning("No user data connections available for unsubscription.")
+            return
+
+        if id not in global_stream_connections.stream_connections_map:
+            logging.warning(f"Stream {id} is not subscribed.")
+            return
+
+        connection = (
+            global_stream_connections.stream_connections_map[id]
+            if id in global_stream_connections.stream_connections_map
+            else None
+        )
+        if connection:
+            global_stream_connections.stream_connections_map.pop(id, None)
+            logging.info(f"Unsubscribed from stream: {id}")
+        else:
+            raise ValueError(f"Subscription id {id} not connected.")
+
+
+class RequestStreamHandle(Generic[T]):
+    """A wrapper for Request Stream Method.
+
+    :param websocket_base: WebSocket base.
+    :param stream: Stream name.
+    :param response_model: The Pydantic model to validate the response data.
+    """
+
+    def __init__(
+        self,
+        websocket_base: WebSocketStreamBase | WebSocketAPIBase,
+        stream: str,
+        response_model: Type[T] = None,
+    ):
+        self._websocket_base = websocket_base
+        self._stream = stream
+        self._response_model = response_model
+
+    async def unsubscribe(self) -> None:
+        if isinstance(self._websocket_base, WebSocketStreamBase):
+            await self._websocket_base.unsubscribe(streams=self._stream)
+        else:
+            await self._websocket_base.unsubscribe(id=self._stream)
+
+    def on(self, event: str, callback: Callable[[T], None]) -> None:
+        self._websocket_base.on(event, callback, self._stream)
+
 
 async def RequestStream(
-    websocket_base: WebSocketStreamBase, stream: str, response_model: Type[T] = None
-) -> SimpleNamespace:
+    websocket_base: WebSocketStreamBase | WebSocketAPIBase, stream: str, response_model: Type[T] = None
+) -> RequestStreamHandle[T]:
     """Decorator to create a request stream for a specific stream.
 
     Args:
-        websocket_base (WebSocketStreamBase): WebSocket stream base.
+        websocket_base (WebSocketStreamBase | WebSocketAPIBase): WebSocket base.
         stream (str): Stream name.
+        response_model (Type[T], optional): Response model for the stream.
     """
-    await websocket_base.subscribe(streams=[stream], response_model=response_model)
-
-    def on(event: str, callback: Callable[[T], None]):
-        websocket_base.on(event, callback, stream)
 
-    async def unsubscribe():
-        await websocket_base.unsubscribe(streams=stream)
+    if isinstance(websocket_base, WebSocketStreamBase):
+        await websocket_base.subscribe(streams=[stream], response_model=response_model)
+    else:
+        await websocket_base.subscribe_user_data(id=stream)
 
-    return SimpleNamespace(on=on, unsubscribe=unsubscribe)
+    return RequestStreamHandle(websocket_base, stream, response_model)