Merge pull request #178 from GetStream/audio-more

Non blocking json enc/dec

Author: tbarbugli

getstream/base.py

@@ -1,6 +1,7 @@
 import json
 import time
 import uuid
+import asyncio
 from typing import Any, Dict, Optional, Type, get_origin
 
 from getstream.models import APIError
@@ -355,6 +356,14 @@ async def _request_async(
         ) as span:
             call_kwargs = dict(kwargs)
             call_kwargs.pop("path_params", None)
+
+            if call_kwargs.get("json") is not None:
+                json_body = call_kwargs.pop("json")
+                json_str = await asyncio.to_thread(json.dumps, json_body)
+                call_kwargs["content"] = json_str
+                call_kwargs["headers"] = call_kwargs.get("headers", {})
+                call_kwargs["headers"]["Content-Type"] = "application/json"
+
             response = await getattr(self.client, method.lower())(
                 url_path, params=query_params, *args, **call_kwargs
             )
@@ -377,7 +386,9 @@ async def _request_async(
                 status_code=getattr(response, "status_code", None),
             )
             record_metrics(duration_ms, attributes=metric_attrs)
-            return self._parse_response(response, data_type or Dict[str, Any])
+            return await asyncio.to_thread(
+                self._parse_response, response, data_type or Dict[str, Any]
+            )
 
     async def patch(
         self,

getstream/utils/__init__.py

@@ -126,6 +126,28 @@ def build_query_param(**kwargs):
     return params
 
 
+async def build_query_param_async(**kwargs):
+    """
+    Async version that offloads CPU-bound JSON serialization to thread pool.
+
+    Constructs a dictionary of query parameters from keyword arguments.
+
+    This function handles various data types:
+    - JSON-serializable objects with a `to_json` method will be serialized using that method.
+    - Booleans are converted to lowercase strings.
+    - Lists are converted to comma-separated strings with URL-encoded values.
+    - Other types (strings, integers, dictionaries) are handled appropriately.
+
+    Args:
+        **kwargs: Arbitrary keyword arguments representing potential query parameters.
+
+    Returns:
+        dict: A dictionary where keys are parameter names and values are URL-ready strings.
+    """
+    # Use sync version in thread pool to avoid blocking event loop
+    return await asyncio.to_thread(build_query_param, **kwargs)
+
+
 def build_body_dict(**kwargs):
     """
     Constructs a dictionary for the body of a request, handling nested structures.
@@ -153,6 +175,24 @@ def handle_value(value):
     return data
 
 
+async def build_body_dict_async(**kwargs):
+    """
+    Async version that offloads CPU-bound to_dict() calls to thread pool.
+
+    Constructs a dictionary for the body of a request, handling nested structures.
+    If an object has a `to_dict` method, it calls this method to serialize the object.
+    It handles nested dictionaries and lists recursively.
+
+    Args:
+        **kwargs: Keyword arguments representing keys and values to be included in the body dictionary.
+
+    Returns:
+        dict: A dictionary with keys corresponding to kwargs keys and values processed, potentially recursively.
+    """
+    # Use sync version in thread pool to avoid blocking event loop
+    return await asyncio.to_thread(build_body_dict, **kwargs)
+
+
 def configure_logging(level=None, handler=None, format=None):
     """
     Configure logging for the Stream library.
@@ -232,7 +272,9 @@ async def sync_to_async(func, *args, **kwargs):
     "encode_datetime",
     "datetime_from_unix_ns",
     "build_query_param",
+    "build_query_param_async",
     "build_body_dict",
+    "build_body_dict_async",
     "validate_and_clean_url",
     "configure_logging",
     "UTC",