Dev/errors

dashscope/agentstudio/exceptions.py

@@ -14,13 +14,31 @@
 instead of nested ``error.{code,message}``. We accept both shapes and
 normalize to the documented form. The compatibility branch is marked
 with ``# TODO(bma-fix)`` so we can remove it once the backend aligns.
+
+Codes and default messages come from :mod:`dashscope.common.error_registry`
+(the single source of truth). :func:`from_response` normalizes to that
+taxonomy — rewriting legacy aliases and falling back to the per-status code
+— so ``.code`` is always unified; the server's raw payload stays on ``.raw``.
 """
 
 from __future__ import annotations
 
+import re
 from typing import Any, Dict, Mapping, Optional
 
+from dashscope.common import error_registry as _error_registry
 from dashscope.common.error import DashScopeException
+from dashscope.common.error_registry import (
+    AUTH_FAILED,
+    INTERNAL_ERROR,
+    INVALID_REQUEST,
+    PERMISSION_DENIED,
+    PublicError,
+    RATE_LIMIT_EXCEEDED,
+    REQUEST_TIMEOUT,
+    RESOURCE_NOT_FOUND,
+    SERVICE_UNAVAILABLE,
+)
 
 
 class AgentStudioError(DashScopeException):
@@ -104,7 +122,9 @@ class AuthenticationError(APIStatusError):
 
 
 class PermissionDeniedError(APIStatusError):
-    code = "permission_denied_error"
+    # Unified external standard (registry anthropic_error_code / the
+    # canonical Anthropic API error type).
+    code = "permission_error"
 
 
 class NotFoundError(APIStatusError):
@@ -162,17 +182,81 @@ class StreamClosedError(StreamError):
     504: InternalServerError,
 }
 
+# Registry row per status, for the default code/message when the server
+# omits one. 409 has no row and falls back to a bare HTTP status.
+_STATUS_TO_PUBLIC: Dict[int, PublicError] = {
+    400: INVALID_REQUEST,
+    401: AUTH_FAILED,
+    403: PERMISSION_DENIED,
+    404: RESOURCE_NOT_FOUND,
+    429: RATE_LIMIT_EXCEEDED,
+    500: INTERNAL_ERROR,
+    502: INTERNAL_ERROR,
+    503: SERVICE_UNAVAILABLE,
+    504: REQUEST_TIMEOUT,
+}
+
+# Class routing: normalized code -> exception type. Legacy aliases are
+# rewritten to their canonical form by ``_normalize_code`` before routing.
 _CODE_TO_CLASS: Dict[str, type] = {
     "invalid_request_error": InvalidRequestError,
     "authentication_error": AuthenticationError,
-    "permission_denied_error": PermissionDeniedError,
+    "permission_error": PermissionDeniedError,
     "not_found_error": NotFoundError,
     "conflict_error": ConflictError,
     "rate_limit_error": RateLimitError,
     "overloaded_error": OverloadedError,
     "api_error": InternalServerError,
 }
 
+# Codes the registry defines; a server code already in this set is kept
+# as-is (e.g. ``billing_error`` stays distinct from ``rate_limit_error``).
+_REGISTRY_CODES = frozenset(
+    pe.anthropic_error_code
+    for pe in vars(_error_registry).values()
+    if isinstance(pe, PublicError)
+)
+
+# Pre-standard codes the backend has emitted historically, mapped to their
+# normalized form so old exceptions still resolve to a unified code.
+_LEGACY_CODE_ALIASES: Dict[str, str] = {
+    "permission_denied_error": "permission_error",  # TODO(bma-fix)
+}
+
+_PLACEHOLDER_RE = re.compile(r"\s*:?\s*\{[^}]+\}")
+
+
+def _normalize_code(code: Optional[str]) -> Optional[str]:
+    """Map a server-supplied code to the unified taxonomy.
+
+    Returns the canonical code for a known alias, the code itself when it is
+    already a registry code, or ``None`` when the SDK does not recognize it
+    (the caller then falls back to the registry's per-status code).
+    """
+
+    if code is None:
+        return None
+    if code in _LEGACY_CODE_ALIASES:
+        return _LEGACY_CODE_ALIASES[code]
+    if code in _REGISTRY_CODES:
+        return code
+    return None
+
+
+def _default_message(public: PublicError) -> str:
+    """Placeholder-free default message from a registry entry.
+
+    Registry ``error_msg`` values may embed ``{var}`` templates the server
+    fills in (e.g. ``"...not found: {resource}."``). When falling back to a
+    default we have no values, so strip the unresolved placeholders for a
+    clean sentence.
+    """
+
+    msg = _PLACEHOLDER_RE.sub("", public.error_msg).strip()
+    if msg and not msg.endswith("."):
+        msg += "."
+    return msg
+
 
 def from_response(
     *,
@@ -222,14 +306,24 @@ def from_response(
         if message is None:
             message = body.get("message")
 
-    if message is None:
-        message = f"HTTP {status_code}"
-    if code is None:
+    public = _STATUS_TO_PUBLIC.get(status_code)
+
+    # Normalized SDK code wins; unknown/omitted falls back to the status row.
+    normalized = _normalize_code(code)
+    if normalized is not None:
+        code = normalized
+    elif public is not None:
+        code = public.anthropic_error_code
+    else:
         code = _STATUS_TO_DEFAULT.get(  # type: ignore[attr-defined]
             status_code,
             APIStatusError,
         ).code
 
+    if message is None:
+        # Registry default (placeholder-free); else a bare HTTP status.
+        message = _default_message(public) if public else f"HTTP {status_code}"
+
     cls = (
         _CODE_TO_CLASS.get(code)
         or _STATUS_TO_DEFAULT.get(status_code)

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