Dev/errors

dashscope/api_entities/aiohttp_request.py

@@ -3,7 +3,7 @@
 
 import json
 from http import HTTPStatus
-from typing import Optional
+from typing import Optional, Dict
 
 import aiohttp
 
@@ -17,7 +17,11 @@
 )
 from dashscope.common.error import UnsupportedHTTPMethod
 from dashscope.common.logging import logger
-from dashscope.common.utils import async_to_sync
+from dashscope.common.error_registry import INTERNAL_ERROR
+from dashscope.common.utils import (
+    async_to_sync,
+    _handle_aiohttp_failed_response,
+)
 
 
 class AioHttpRequest(AioBaseRequest):
@@ -85,10 +89,10 @@ def __init__(
         else:
             self.timeout = timeout  # type: ignore[has-type]
 
-    def add_header(self, key, value):
+    def add_header(self, key: str, value: str) -> None:
         self.headers[key] = value
 
-    def add_headers(self, headers):
+    def add_headers(self, headers: Dict[str, str]) -> None:
         self.headers = {**self.headers, **headers}
 
     def call(self):
@@ -97,9 +101,8 @@ def call(self):
             return (item for item in response)
         else:
             output = next(response)
-            try:
-                next(response)
-            except StopIteration:
+            # Consume remaining items to ensure generator completes
+            for _ in response:
                 pass
             return output
 
@@ -109,9 +112,8 @@ async def aio_call(self):
             return (item async for item in response)
         else:
             result = await response.__anext__()
-            try:
-                await response.__anext__()
-            except StopAsyncIteration:
+            # Consume remaining items to ensure generator completes
+            async for _ in response:
                 pass
             return result
 
@@ -142,6 +144,7 @@ async def _handle_response(  # pylint: disable=too-many-branches
         response: aiohttp.ClientResponse,
     ):
         request_id = ""
+        headers = dict(response.headers)
         if (
             response.status == HTTPStatus.OK
             and self.stream
@@ -162,26 +165,37 @@ async def _handle_response(  # pylint: disable=too-many-branches
                     if "request_id" in msg:
                         request_id = msg["request_id"]
                 except json.JSONDecodeError:
+                    logger.error(
+                        "Failed to parse SSE stream data: %s",
+                        data[:200] if len(data) > 200 else data,
+                    )
                     yield DashScopeAPIResponse(
                         request_id=request_id,
-                        status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
-                        code="Unknown",
-                        message=data,
+                        status_code=INTERNAL_ERROR.status_code,
+                        code=INTERNAL_ERROR.error_code,
+                        message=INTERNAL_ERROR.error_msg,
+                        headers=headers,
                     )
                     continue
-                if is_error:
+                if is_error and msg is not None:
                     yield DashScopeAPIResponse(
                         request_id=request_id,
                         status_code=status_code,
-                        code=msg["code"],
-                        message=msg["message"],
+                        code=msg.get("code")
+                        or msg.get("error_code")
+                        or f"http_{status_code}",
+                        message=msg.get("message")
+                        or msg.get("error_message")
+                        or f"HTTP {status_code} error",
+                        headers=headers,
                     )
                 else:
                     yield DashScopeAPIResponse(
                         request_id=request_id,
                         status_code=HTTPStatus.OK,
                         output=output,
                         usage=usage,
+                        headers=headers,
                     )
         elif (
             response.status == HTTPStatus.OK
@@ -201,6 +215,7 @@ async def _handle_response(  # pylint: disable=too-many-branches
                 request_id=request_id,
                 status_code=HTTPStatus.OK,
                 output=output,
+                headers=headers,
             )
         elif response.status == HTTPStatus.OK:
             json_content = await response.json()
@@ -217,51 +232,22 @@ async def _handle_response(  # pylint: disable=too-many-branches
                 status_code=HTTPStatus.OK,
                 output=output,
                 usage=usage,
+                headers=headers,
             )
         else:
-            if "application/json" in response.content_type:
-                error = await response.json()
-                if "request_id" in error:
-                    request_id = error["request_id"]
-                if "message" not in error:
-                    message = ""
-                    logger.error(
-                        "Request: %s failed, status: %s",
-                        self.url,
-                        response.status,
-                    )
-                else:
-                    message = error["message"]
-                    logger.error(
-                        "Request: %s failed, status: %s, message: %s",
-                        self.url,
-                        response.status,
-                        error["message"],
-                    )
-                yield DashScopeAPIResponse(
-                    request_id=request_id,
-                    status_code=response.status,
-                    code=error["code"],
-                    message=message,
-                )
-            else:
-                msg = await response.read()
-                yield DashScopeAPIResponse(
-                    request_id=request_id,
-                    status_code=response.status,
-                    code="Unknown",
-                    message=msg.decode("utf-8"),
-                )
+            yield _handle_aiohttp_failed_response(response)
 
     # pylint: disable=too-many-branches
     async def _handle_request(self):
         try:
+            # Session management:
+            # - External session: managed by caller, we never close it
+            # - Shared session: managed by get_shared_aio_session(),
+            #   uses connection pooling and is closed when no longer needed
             if self._external_aio_session is not None:
                 session = self._external_aio_session
-                should_close = False
             else:
                 session = await get_shared_aio_session()
-                should_close = False
 
             if self.stream:
                 request_timeout = aiohttp.ClientTimeout(
@@ -315,8 +301,21 @@ async def _handle_request(self):
                     async for rsp in self._handle_response(response):
                         yield rsp
             finally:
-                if should_close:
-                    await session.close()
+                # Note: We don't close the session here because:
+                # - External sessions are managed by the caller
+                # - Shared sessions use connection pooling and are
+                #   managed centrally by get_shared_aio_session()
+                pass
         except Exception as e:
-            logger.debug(e)
-            raise e
+            logger.error(
+                "Request failed: url=%s, method=%s, error=%s",
+                self.url,
+                self.method,
+                str(e),
+                exc_info=True,
+            )
+            from dashscope.common.error import DashScopeException
+
+            raise DashScopeException(
+                f"Request failed: {str(e)}",
+            ) from e

dashscope/api_entities/api_request_factory.py

@@ -69,6 +69,68 @@ def _get_protocol_params(kwargs):
     )
 
 
+def _build_http_url(
+    base_address: str,
+    is_service: bool,
+    task_group: str,
+    task: str,
+    function: str,
+    extra_url_parameters: dict,
+) -> str:
+    """Build HTTP/HTTPS URL from components with validation."""
+    if base_address is None:
+        base_address = dashscope.base_http_api_url
+
+    # Validate base_address has proper scheme (http:// or https://)
+    if base_address and not base_address.startswith(
+        ("http://", "https://"),
+    ):
+        from dashscope.common.error import InvalidBaseURL
+
+        raise InvalidBaseURL(
+            f"Invalid URL '{base_address}': No scheme supplied. "
+            f"Perhaps you meant https://{base_address}?",
+        )
+
+    if not base_address.endswith("/"):
+        http_url = base_address + "/"
+    else:
+        http_url = base_address
+
+    if is_service:
+        http_url = http_url + SERVICE_API_PATH + "/"
+
+    if task_group:
+        http_url += f"{task_group}/"
+    if task:
+        http_url += f"{task}/"
+    if function:
+        http_url += function
+    if extra_url_parameters is not None and extra_url_parameters:
+        http_url += "?" + urlencode(extra_url_parameters)
+
+    return http_url
+
+
+def _build_websocket_url(base_address: str) -> str:
+    """Build and validate WebSocket URL."""
+    if base_address is not None:
+        websocket_url = base_address
+    else:
+        websocket_url = dashscope.base_websocket_api_url
+
+    # Validate websocket_url has proper scheme (ws:// or wss://)
+    if websocket_url and not websocket_url.startswith(("ws://", "wss://")):
+        from dashscope.common.error import InvalidBaseURL
+
+        raise InvalidBaseURL(
+            f"Invalid URL '{websocket_url}': No scheme supplied. "
+            f"Perhaps you meant wss://{websocket_url}?",
+        )
+
+    return websocket_url
+
+
 def _build_api_request(  # pylint: disable=too-many-branches
     model: str,
     input: object,  # pylint: disable=redefined-builtin
@@ -102,24 +164,14 @@ def _build_api_request(  # pylint: disable=too-many-branches
     encryption = None
 
     if api_protocol in [ApiProtocol.HTTP, ApiProtocol.HTTPS]:
-        if base_address is None:
-            base_address = dashscope.base_http_api_url
-        if not base_address.endswith("/"):
-            http_url = base_address + "/"
-        else:
-            http_url = base_address
-
-        if is_service:
-            http_url = http_url + SERVICE_API_PATH + "/"
-
-        if task_group:
-            http_url += f"{task_group}/"
-        if task:
-            http_url += f"{task}/"
-        if function:
-            http_url += function
-        if extra_url_parameters is not None and extra_url_parameters:
-            http_url += "?" + urlencode(extra_url_parameters)
+        http_url = _build_http_url(
+            base_address,
+            is_service,
+            task_group,
+            task,
+            function,
+            extra_url_parameters,
+        )
 
         if enable_encryption is True:
             encryption = Encryption()
@@ -142,10 +194,7 @@ def _build_api_request(  # pylint: disable=too-many-branches
             session=session,
         )
     elif api_protocol == ApiProtocol.WEBSOCKET:
-        if base_address is not None:
-            websocket_url = base_address
-        else:
-            websocket_url = dashscope.base_websocket_api_url
+        websocket_url = _build_websocket_url(base_address)
         pre_task_id = kwargs.pop("pre_task_id", None)
         request = WebSocketRequest(
             url=websocket_url,