Structured errors part 1: common HTTP wrapper (#22)

* fix unit tests after optionset / logicalname changes

* undo code from another branch

* Add HTTP wrapper and extract DV response info

* duplicate line

* comment fixes

* PR fixes

---------

Co-authored-by: Tim Pellissier <tpellissier@microsoft.com>

Author: tpellissier-msft

examples/quickstart_file_upload.py

@@ -192,7 +192,6 @@ def ensure_file_attribute_generic(schema_name: str, label: str, key_prefix: str)
             f"SchemaName eq '{schema_name}'"
         )
         r = odata._request("get", url, headers=odata._headers())
-        r.raise_for_status()
         val = []
         try:
             val = r.json().get("value", [])
@@ -217,7 +216,6 @@ def ensure_file_attribute_generic(schema_name: str, label: str, key_prefix: str)
     try:
         url = f"{odata.api}/EntityDefinitions({meta_id})/Attributes"
         r = odata._request("post", url, headers=odata._headers(), json=payload)
-        r.raise_for_status()
         print({f"{key_prefix}_file_attribute_created": True})
         time.sleep(2)
         return True
@@ -291,7 +289,6 @@ def get_dataset_info(file_path: Path):
         odata = client._get_odata()
         dl_url_single = f"{odata.api}/{entity_set}({record_id})/{small_file_attr_logical}/$value"  # raw entity_set URL OK
         resp_single = odata._request("get", dl_url_single, headers=odata._headers())
-        resp_single.raise_for_status()
         content_single = resp_single.content or b""
         import hashlib  # noqa: WPS433
         downloaded_hash = hashlib.sha256(content_single).hexdigest() if content_single else None
@@ -317,7 +314,6 @@ def get_dataset_info(file_path: Path):
         ))
         print({"small_replace_upload_completed": True, "small_replace_source_size": replace_size_small})
         resp_single_replace = odata._request("get", dl_url_single, headers=odata._headers())
-        resp_single_replace.raise_for_status()
         content_single_replace = resp_single_replace.content or b""
         downloaded_hash_replace = hashlib.sha256(content_single_replace).hexdigest() if content_single_replace else None
         hash_match_replace = (downloaded_hash_replace == replace_hash_small) if (downloaded_hash_replace and replace_hash_small) else None
@@ -348,7 +344,6 @@ def get_dataset_info(file_path: Path):
         odata = client._get_odata()
         dl_url_chunk = f"{odata.api}/{entity_set}({record_id})/{chunk_file_attr_logical}/$value"  # raw entity_set for download
         resp_chunk = odata._request("get", dl_url_chunk, headers=odata._headers())
-        resp_chunk.raise_for_status()
         content_chunk = resp_chunk.content or b""
         import hashlib  # noqa: WPS433
         dst_hash_chunk = hashlib.sha256(content_chunk).hexdigest() if content_chunk else None
@@ -373,7 +368,6 @@ def get_dataset_info(file_path: Path):
         ))
         print({"chunk_replace_upload_completed": True})
         resp_chunk_replace = odata._request("get", dl_url_chunk, headers=odata._headers())
-        resp_chunk_replace.raise_for_status()
         content_chunk_replace = resp_chunk_replace.content or b""
         dst_hash_chunk_replace = hashlib.sha256(content_chunk_replace).hexdigest() if content_chunk_replace else None
         hash_match_chunk_replace = (dst_hash_chunk_replace == replace_hash_chunk) if (dst_hash_chunk_replace and replace_hash_chunk) else None

src/dataverse_sdk/error_codes.py

@@ -0,0 +1,28 @@
+# HTTP subcode constants
+HTTP_400 = "http_400"
+HTTP_401 = "http_401"
+HTTP_403 = "http_403"
+HTTP_404 = "http_404"
+HTTP_409 = "http_409"
+HTTP_412 = "http_412"
+HTTP_415 = "http_415"
+HTTP_429 = "http_429"
+HTTP_500 = "http_500"
+HTTP_502 = "http_502"
+HTTP_503 = "http_503"
+HTTP_504 = "http_504"
+
+ALL_HTTP_SUBCODES = {
+    HTTP_400,
+    HTTP_401,
+    HTTP_403,
+    HTTP_404,
+    HTTP_409,
+    HTTP_412,
+    HTTP_415,
+    HTTP_429,
+    HTTP_500,
+    HTTP_502,
+    HTTP_503,
+    HTTP_504,
+}

src/dataverse_sdk/errors.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+from typing import Any, Dict, Optional
+import datetime as _dt
+
+class DataverseError(Exception):
+    """Base structured error for the Dataverse SDK."""
+    def __init__(
+        self,
+        message: str,
+        *,
+        code: str,
+        subcode: Optional[str] = None,
+        status_code: Optional[int] = None,
+        details: Optional[Dict[str, Any]] = None,
+        source: Optional[Dict[str, Any]] = None,
+        is_transient: Optional[bool] = None,
+    ) -> None:
+        super().__init__(message)
+        self.message = message
+        self.code = code
+        self.subcode = subcode
+        self.status_code = status_code
+        self.details = details or {}
+        self.source = source or {}
+        self.is_transient = is_transient
+        self.timestamp = _dt.datetime.utcnow().isoformat() + "Z"
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "message": self.message,
+            "code": self.code,
+            "subcode": self.subcode,
+            "status_code": self.status_code,
+            "details": self.details,
+            "source": self.source,
+            "is_transient": self.is_transient,
+            "timestamp": self.timestamp,
+        }
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return f"{self.__class__.__name__}(code={self.code!r}, subcode={self.subcode!r}, message={self.message!r})"
+
+class HttpError(DataverseError):
+    def __init__(
+        self,
+        message: str,
+        *,
+        subcode: Optional[str] = None,
+        status_code: Optional[int] = None,
+        details: Optional[Dict[str, Any]] = None,
+        source: Optional[Dict[str, Any]] = None,
+        is_transient: Optional[bool] = None,
+    ) -> None:
+        super().__init__(
+            message,
+            code="http",
+            subcode=subcode,
+            status_code=status_code,
+            details=details,
+            source=source,
+            is_transient=is_transient,
+        )
+
+__all__ = ["DataverseError", "HttpError"]

src/dataverse_sdk/odata.py

@@ -10,6 +10,8 @@
 
 from .http import HttpClient
 from .odata_upload_files import ODataFileUpload
+from .errors import HttpError
+from . import error_codes as ec
 
 
 _GUID_RE = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}")
@@ -60,9 +62,50 @@ def _headers(self) -> Dict[str, str]:
             "OData-Version": "4.0",
         }
 
-    def _request(self, method: str, url: str, **kwargs):
+    def _raw_request(self, method: str, url: str, **kwargs):
         return self._http.request(method, url, **kwargs)
 
+    def _request(self, method: str, url: str, *, expected: tuple[int, ...] = (200, 201, 202, 204), **kwargs):
+        """Execute HTTP request; raise HttpError with structured details on failure.
+
+        Returns the raw response for success codes; raises HttpError with extracted
+        Dataverse error payload fields and correlation identifiers otherwise.
+        """
+        r = self._raw_request(method, url, **kwargs)
+        if r.status_code in expected:
+            return r
+        payload = {}
+        try:
+            payload = r.json() if getattr(r, 'text', None) else {}
+        except Exception:
+            payload = {}
+        svc_err = payload.get("error") if isinstance(payload, dict) else None
+        svc_code = svc_err.get("code") if isinstance(svc_err, dict) else None
+        svc_msg = svc_err.get("message") if isinstance(svc_err, dict) else None
+        message = svc_msg or f"HTTP {r.status_code}"
+        subcode = f"http_{r.status_code}"
+
+        headers = getattr(r, 'headers', {}) or {}
+        details = {
+            "service_error_code": svc_code,
+            "body_excerpt": (getattr(r, 'text', '') or '')[:200],
+            "correlation_id": headers.get("x-ms-correlation-request-id") or headers.get("x-ms-correlation-id"),
+            "request_id": headers.get("x-ms-client-request-id") or headers.get("request-id"),
+            "traceparent": headers.get("traceparent"),
+        }
+        ra = headers.get("Retry-After")
+        if ra:
+            details["retry_after"] = ra
+        is_transient = r.status_code in (429, 502, 503, 504)
+        raise HttpError(
+            message,
+            subcode=subcode,
+            status_code=r.status_code,
+            details=details,
+            source={"method": method, "url": url},
+            is_transient=is_transient,
+        )
+
     # ----------------------------- CRUD ---------------------------------
     def _create(self, logical_name: str, data: Union[Dict[str, Any], List[Dict[str, Any]]]) -> Union[str, List[str]]:
         """Create one or many records by logical (singular) name.
@@ -108,7 +151,6 @@ def _create_single(self, entity_set: str, logical_name: str, record: Dict[str, A
         url = f"{self.api}/{entity_set}"
         headers = self._headers().copy()
         r = self._request("post", url, headers=headers, json=record)
-        r.raise_for_status()
 
         ent_loc = r.headers.get("OData-EntityId") or r.headers.get("OData-EntityID")
         if ent_loc:
@@ -144,7 +186,6 @@ def _create_multiple(self, entity_set: str, logical_name: str, records: List[Dic
         # The action currently returns only Ids; no need to request representation.
         headers = self._headers().copy()
         r = self._request("post", url, headers=headers, json=payload)
-        r.raise_for_status()
         try:
             body = r.json() if r.text else {}
         except ValueError:
@@ -264,7 +305,6 @@ def _update(self, logical_name: str, key: str, data: Dict[str, Any]) -> None:
         headers = self._headers().copy()
         headers["If-Match"] = "*"
         r = self._request("patch", url, headers=headers, json=data)
-        r.raise_for_status()
 
     def _update_multiple(self, entity_set: str, logical_name: str, records: List[Dict[str, Any]]) -> None:
         """Bulk update existing records via the collection-bound UpdateMultiple action.
@@ -315,7 +355,6 @@ def _update_multiple(self, entity_set: str, logical_name: str, records: List[Dic
         url = f"{self.api}/{entity_set}/Microsoft.Dynamics.CRM.UpdateMultiple"
         headers = self._headers().copy()
         r = self._request("post", url, headers=headers, json=payload)
-        r.raise_for_status()
         # Intentionally ignore response content: no stable contract for IDs across environments.
         return None
 
@@ -325,8 +364,7 @@ def _delete(self, logical_name: str, key: str) -> None:
         url = f"{self.api}/{entity_set}{self._format_key(key)}"
         headers = self._headers().copy()
         headers["If-Match"] = "*"
-        r = self._request("delete", url, headers=headers)
-        r.raise_for_status()
+        self._request("delete", url, headers=headers)
 
     def _get(self, logical_name: str, key: str, select: Optional[str] = None) -> Dict[str, Any]:
         """Retrieve a single record.
@@ -346,7 +384,6 @@ def _get(self, logical_name: str, key: str, select: Optional[str] = None) -> Dic
         entity_set = self._entity_set_from_logical(logical_name)
         url = f"{self.api}/{entity_set}{self._format_key(key)}"
         r = self._request("get", url, headers=self._headers(), params=params)
-        r.raise_for_status()
         return r.json()
 
     def _get_multiple(
@@ -392,7 +429,6 @@ def _get_multiple(
 
         def _do_request(url: str, *, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
             r = self._request("get", url, headers=headers, params=params)
-            r.raise_for_status()
             try:
                 return r.json()
             except ValueError:
@@ -468,17 +504,6 @@ def _query_sql(self, sql: str) -> list[dict[str, Any]]:
         url = f"{self.api}/{entity_set}"
         params = {"sql": sql}
         r = self._request("get", url, headers=headers, params=params)
-        try:
-            r.raise_for_status()
-        except Exception as e:
-            # Attach response snippet to aid debugging unsupported SQL patterns
-            resp_text = None
-            try:
-                resp_text = r.text[:500] if getattr(r, 'text', None) else None
-            except Exception:
-                pass
-            detail = f" SQL query failed (status={getattr(r, 'status_code', '?')}): {resp_text}" if resp_text else ""
-            raise RuntimeError(str(e) + detail) from e
         try:
             body = r.json()
         except ValueError:
@@ -530,7 +555,6 @@ def _entity_set_from_logical(self, logical: str) -> str:
             "$filter": f"LogicalName eq '{logical_escaped}'",
         }
         r = self._request("get", url, headers=self._headers(), params=params)
-        r.raise_for_status()
         try:
             body = r.json()
             items = body.get("value", []) if isinstance(body, dict) else []
@@ -576,7 +600,6 @@ def _get_entity_by_schema(self, schema_name: str) -> Optional[Dict[str, Any]]:
             "$filter": f"SchemaName eq '{schema_escaped}'",
         }
         r = self._request("get", url, headers=self._headers(), params=params)
-        r.raise_for_status()
         items = r.json().get("value", [])
         return items[0] if items else None
 
@@ -596,7 +619,6 @@ def _create_entity(self, schema_name: str, display_name: str, attributes: List[D
         }
         headers = self._headers()
         r = self._request("post", url, headers=headers, json=payload)
-        r.raise_for_status()
         ent = self._wait_for_entity_ready(schema_name)
         if not ent or not ent.get("EntitySetName"):
             raise RuntimeError(
@@ -769,7 +791,7 @@ def _optionset_map(self, logical_name: str, attr_logical: str) -> Optional[Dict[
         # Retry up to 3 times on 404 (new or not-yet-published attribute metadata). If still 404, raise.
         r_type = None
         for attempt in range(3):
-            r_type = self._request("get", url_type, headers=self._headers())
+            r_type = self._raw_request("get", url_type, headers=self._headers())
             if r_type.status_code != 404:
                 break
             if attempt < 2:
@@ -780,7 +802,9 @@ def _optionset_map(self, logical_name: str, attr_logical: str) -> Optional[Dict[
             raise RuntimeError(
                 f"Picklist attribute metadata not found after retries: entity='{logical_name}' attribute='{attr_logical}' (404)"
             )
-        r_type.raise_for_status()
+        if not (200 <= r_type.status_code < 300):
+            # Re-issue via _send to raise structured HttpError (rare path)
+            self._request("get", url_type, headers=self._headers())
 
         body_type = r_type.json()
         items = body_type.get("value", []) if isinstance(body_type, dict) else []
@@ -800,14 +824,15 @@ def _optionset_map(self, logical_name: str, attr_logical: str) -> Optional[Dict[
         # Step 2 fetch with retries: expanded OptionSet (cast form first)
         r_opts = None
         for attempt in range(3):
-            r_opts = self._request("get", cast_url, headers=self._headers())
+            r_opts = self._raw_request("get", cast_url, headers=self._headers())
             if r_opts.status_code != 404:
                 break
             if attempt < 2:
                 time.sleep(0.4 * (2 ** attempt))  # 0.4s, 0.8s
         if r_opts.status_code == 404:
             raise RuntimeError(f"Picklist OptionSet metadata not found after retries: entity='{logical_name}' attribute='{attr_logical}' (404)")
-        r_opts.raise_for_status()
+        if not (200 <= r_opts.status_code < 300):
+            self._request("get", cast_url, headers=self._headers())
 
         attr_full = {}
         try:
@@ -969,7 +994,6 @@ def _list_tables(self) -> List[Dict[str, Any]]:
             "$filter": "IsPrivate eq false"
         }
         r = self._request("get", url, headers=self._headers(), params=params)
-        r.raise_for_status()
         return r.json().get("value", [])
 
     def _delete_table(self, tablename: str) -> None:
@@ -982,7 +1006,6 @@ def _delete_table(self, tablename: str) -> None:
         url = f"{self.api}/EntityDefinitions({metadata_id})"
         headers = self._headers()
         r = self._request("delete", url, headers=headers)
-        r.raise_for_status()
 
     def _create_table(self, tablename: str, schema: Dict[str, Any]) -> Dict[str, Any]:
         # Accept a friendly name and construct a default schema under 'new_'.