fix: fix image synthesis parameter passing and resource leak issues#156

Author: kevin

dashscope/aigc/image_synthesis.py

@@ -464,7 +464,8 @@ def async_call(  # pylint: disable=arguments-renamed  # type: ignore[override] #
         if color_palette is not None:
             kwargs["color_palette"] = color_palette
         task_group, f = _get_task_group_and_task(__name__)
-        inputs, kwargs, task = cls._get_input(
+        # pylint: disable=protected-access
+        inputs, kwargs, task = ImageSynthesis._get_input(
             model,
             prompt,
             negative_prompt,
@@ -785,6 +786,30 @@ async def sync_call(
         Note: This method currently now only supports wan2.2-t2i-flash and wan2.2-t2i-plus.  # noqa: E501  # pylint: disable=line-too-long
             Using other models will result in an error，More raw image models may be added for use later  # pylint: disable=line-too-long
         """
+        if size is not None:
+            kwargs["size"] = size
+        if n is not None:
+            kwargs["n"] = n
+        if seed is not None:
+            kwargs["seed"] = seed
+        if style is not None:
+            kwargs["style"] = style
+        if ref_strength is not None:
+            kwargs["ref_strength"] = ref_strength
+        if ref_mode is not None:
+            kwargs["ref_mode"] = ref_mode
+        if prompt_extend is not None:
+            kwargs["prompt_extend"] = prompt_extend
+        if watermark is not None:
+            kwargs["watermark"] = watermark
+        if bbox_list is not None:
+            kwargs["bbox_list"] = bbox_list
+        if enable_sequential is not None:
+            kwargs["enable_sequential"] = enable_sequential
+        if thinking_mode is not None:
+            kwargs["thinking_mode"] = thinking_mode
+        if color_palette is not None:
+            kwargs["color_palette"] = color_palette
         task_group, f = _get_task_group_and_task(__name__)
         # pylint: disable=protected-access
         inputs, kwargs, task = ImageSynthesis._get_input(
@@ -886,6 +911,30 @@ async def async_call(  # type: ignore[override]  # pylint: disable=arguments-ren
             DashScopeAPIResponse: The image synthesis
                 task id in the response.
         """
+        if size is not None:
+            kwargs["size"] = size
+        if n is not None:
+            kwargs["n"] = n
+        if seed is not None:
+            kwargs["seed"] = seed
+        if style is not None:
+            kwargs["style"] = style
+        if ref_strength is not None:
+            kwargs["ref_strength"] = ref_strength
+        if ref_mode is not None:
+            kwargs["ref_mode"] = ref_mode
+        if prompt_extend is not None:
+            kwargs["prompt_extend"] = prompt_extend
+        if watermark is not None:
+            kwargs["watermark"] = watermark
+        if bbox_list is not None:
+            kwargs["bbox_list"] = bbox_list
+        if enable_sequential is not None:
+            kwargs["enable_sequential"] = enable_sequential
+        if thinking_mode is not None:
+            kwargs["thinking_mode"] = thinking_mode
+        if color_palette is not None:
+            kwargs["color_palette"] = color_palette
         task_group, f = _get_task_group_and_task(__name__)
         # pylint: disable=protected-access
         inputs, kwargs, task = ImageSynthesis._get_input(

dashscope/api_entities/aio_session.py

@@ -46,10 +46,46 @@ async def get_shared_aio_session() -> aiohttp.ClientSession:
 
         connector = aiohttp.TCPConnector(ssl=get_ssl_context())
         session = aiohttp.ClientSession(connector=connector, trust_env=True)
+
+        # Register finalizer to close session when it's garbage collected.
+        # This prevents resource leaks even if
+        # close_shared_aio_session() is not called.
+        weakref.finalize(session, _sync_close_session, id(session))
+
         _aio_sessions[loop] = session
     return session
 
 
+def _sync_close_session(session_id: int) -> None:
+    """Synchronous callback when session is garbage collected.
+
+    Note: This runs during GC, so we cannot await. We try to close
+    synchronously if possible, otherwise log a warning.
+    """
+    try:
+        loop = asyncio.get_event_loop()
+        if not loop.is_closed() and not loop.is_running():
+            # Loop exists and is not running, we can use it
+            # Find the session in our cache
+            with _lock:
+                for stored_session in _aio_sessions.values():
+                    if (
+                        id(stored_session) == session_id
+                        and not stored_session.closed
+                    ):
+                        try:
+                            loop.run_until_complete(stored_session.close())
+                        except Exception:
+                            pass
+                        break
+    except RuntimeError:
+        # No event loop available, can't close asynchronously
+        pass
+    except Exception:
+        # Ignore errors during cleanup
+        pass
+
+
 async def close_shared_aio_session() -> None:
     """Close the shared session for the current event loop."""
     loop = asyncio.get_running_loop()

dashscope/audio/qwen_tts_realtime/qwen_tts_realtime.py

@@ -377,9 +377,18 @@ def on_message(  # pylint: disable=unused-argument
                             self.last_first_audio_delay,
                         )
             except json.JSONDecodeError:
-                logger.error("Failed to parse message as JSON.")
-                # pylint: disable=broad-exception-raised,raise-missing-from
-                raise Exception("Failed to parse message as JSON.")
+                logger.error(
+                    "Failed to parse message as JSON: %s",
+                    message[:200],
+                )
+                # Do not raise exception here, let the connection stay alive.
+                # Raising exception in callback can cause unexpected thread
+                # termination.
+            except Exception as e:
+                logger.error("Error processing message: %s", str(e))
+                # Do not raise exception here, let the connection stay alive.
+                # Raising exception in callback can cause unexpected thread
+                # termination.
         elif isinstance(message, (bytes, bytearray)):
             # If parsing fails, treat as binary message
             logger.error(