Move multipart helpers to after _BatchBase class in _batch_base.py

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Abel Milash

src/PowerPlatform/Dataverse/data/_batch_base.py

@@ -252,140 +252,6 @@ class _BatchContext(Protocol):
     records: Any  # used by BatchDataFrameOperations to delegate create/update/delete
 
 
-# ---------------------------------------------------------------------------
-# Multipart parsing helpers
-# ---------------------------------------------------------------------------
-
-
-def _raise_top_level_batch_error(response: Any) -> None:
-    """Parse a non-multipart batch response and raise HttpError with the service message.
-
-    Dataverse returns ``application/json`` with an ``{"error": {...}}`` payload when
-    it rejects the batch request at the HTTP level (e.g. malformed multipart body,
-    missing OData headers). This helper surfaces that detail instead of silently
-    returning an empty ``BatchResult``.
-    """
-    status_code: int = getattr(response, "status_code", 0)
-    service_error_code: Optional[str] = None
-    try:
-        payload = response.json()
-        error = payload.get("error", {})
-        service_error_code = error.get("code") or None
-        message: str = error.get("message") or response.text or "Unexpected non-multipart response from $batch"
-    except Exception:
-        message = (getattr(response, "text", None) or "") or "Unexpected non-multipart response from $batch"
-    raise HttpError(
-        message=f"Batch request rejected by Dataverse: {message}",
-        status_code=status_code,
-        subcode=_http_subcode(status_code) if status_code else None,
-        service_error_code=service_error_code,
-    )
-
-
-_BOUNDARY_RE = re.compile(r'boundary="?([^";,\s]+)"?', re.IGNORECASE)
-
-
-def _extract_boundary(content_type: str) -> Optional[str]:
-    m = _BOUNDARY_RE.search(content_type)
-    return m.group(1) if m else None
-
-
-def _split_multipart(body: str, boundary: str) -> List[Tuple[Dict[str, str], str]]:
-    delimiter = f"--{boundary}"
-    parts: List[Tuple[Dict[str, str], str]] = []
-    lines = body.replace("\r\n", "\n").split("\n")
-    current: List[str] = []
-    in_part = False
-    for line in lines:
-        stripped = line.rstrip("\r")
-        if stripped == delimiter:
-            if in_part and current:
-                parts.append(_parse_mime_part("\n".join(current)))
-                current = []
-            in_part = True
-        elif stripped == f"{delimiter}--":
-            if in_part and current:
-                parts.append(_parse_mime_part("\n".join(current)))
-            break
-        elif in_part:
-            current.append(line)
-    return parts
-
-
-def _parse_mime_part(raw: str) -> Tuple[Dict[str, str], str]:
-    if "\n\n" in raw:
-        header_block, body = raw.split("\n\n", 1)
-    else:
-        header_block, body = raw, ""
-    headers: Dict[str, str] = {}
-    for line in header_block.splitlines():
-        if ":" in line:
-            name, _, value = line.partition(":")
-            headers[name.strip().lower()] = value.strip()
-    return headers, body.strip()
-
-
-def _parse_http_response_part(text: str, content_id: Optional[str]) -> Optional[BatchItemResponse]:
-    lines = text.replace("\r\n", "\n").splitlines()
-    if not lines:
-        return None
-    status_line = ""
-    idx = 0
-    for i, line in enumerate(lines):
-        if line.startswith("HTTP/"):
-            status_line = line
-            idx = i + 1
-            break
-    if not status_line:
-        return None
-    parts = status_line.split(" ", 2)
-    if len(parts) < 2:
-        return None
-    try:
-        status_code = int(parts[1])
-    except ValueError:
-        return None
-    resp_headers: Dict[str, str] = {}
-    body_start = idx
-    for i in range(idx, len(lines)):
-        if lines[i] == "":
-            body_start = i + 1
-            break
-        if ":" in lines[i]:
-            name, _, value = lines[i].partition(":")
-            resp_headers[name.strip().lower()] = value.strip()
-    entity_id: Optional[str] = None
-    odata_id = resp_headers.get("odata-entityid", "")
-    if odata_id:
-        m = _GUID_RE.search(odata_id)
-        if m:
-            entity_id = m.group(0)
-    body_text = "\n".join(lines[body_start:]).strip()
-    data: Optional[Dict[str, Any]] = None
-    error_message: Optional[str] = None
-    error_code: Optional[str] = None
-    if body_text:
-        try:
-            parsed = json.loads(body_text)
-            if isinstance(parsed, dict):
-                err = parsed.get("error")
-                if isinstance(err, dict):
-                    error_message = err.get("message")
-                    error_code = err.get("code")
-                else:
-                    data = parsed
-        except (json.JSONDecodeError, ValueError):
-            pass
-    return BatchItemResponse(
-        status_code=status_code,
-        content_id=content_id,
-        entity_id=entity_id,
-        data=data,
-        error_message=error_message,
-        error_code=error_code,
-    )
-
-
 # ---------------------------------------------------------------------------
 # Batch base: pure serialisation and pure table resolvers
 # ---------------------------------------------------------------------------
@@ -528,3 +394,137 @@ def _parse_batch_response(self, response: Any) -> BatchResult:
                 if item is not None:
                     responses.append(item)
         return BatchResult(responses=responses)
+
+
+# ---------------------------------------------------------------------------
+# Multipart parsing helpers
+# ---------------------------------------------------------------------------
+
+
+def _raise_top_level_batch_error(response: Any) -> None:
+    """Parse a non-multipart batch response and raise HttpError with the service message.
+
+    Dataverse returns ``application/json`` with an ``{"error": {...}}`` payload when
+    it rejects the batch request at the HTTP level (e.g. malformed multipart body,
+    missing OData headers). This helper surfaces that detail instead of silently
+    returning an empty ``BatchResult``.
+    """
+    status_code: int = getattr(response, "status_code", 0)
+    service_error_code: Optional[str] = None
+    try:
+        payload = response.json()
+        error = payload.get("error", {})
+        service_error_code = error.get("code") or None
+        message: str = error.get("message") or response.text or "Unexpected non-multipart response from $batch"
+    except Exception:
+        message = (getattr(response, "text", None) or "") or "Unexpected non-multipart response from $batch"
+    raise HttpError(
+        message=f"Batch request rejected by Dataverse: {message}",
+        status_code=status_code,
+        subcode=_http_subcode(status_code) if status_code else None,
+        service_error_code=service_error_code,
+    )
+
+
+_BOUNDARY_RE = re.compile(r'boundary="?([^";,\s]+)"?', re.IGNORECASE)
+
+
+def _extract_boundary(content_type: str) -> Optional[str]:
+    m = _BOUNDARY_RE.search(content_type)
+    return m.group(1) if m else None
+
+
+def _split_multipart(body: str, boundary: str) -> List[Tuple[Dict[str, str], str]]:
+    delimiter = f"--{boundary}"
+    parts: List[Tuple[Dict[str, str], str]] = []
+    lines = body.replace("\r\n", "\n").split("\n")
+    current: List[str] = []
+    in_part = False
+    for line in lines:
+        stripped = line.rstrip("\r")
+        if stripped == delimiter:
+            if in_part and current:
+                parts.append(_parse_mime_part("\n".join(current)))
+                current = []
+            in_part = True
+        elif stripped == f"{delimiter}--":
+            if in_part and current:
+                parts.append(_parse_mime_part("\n".join(current)))
+            break
+        elif in_part:
+            current.append(line)
+    return parts
+
+
+def _parse_mime_part(raw: str) -> Tuple[Dict[str, str], str]:
+    if "\n\n" in raw:
+        header_block, body = raw.split("\n\n", 1)
+    else:
+        header_block, body = raw, ""
+    headers: Dict[str, str] = {}
+    for line in header_block.splitlines():
+        if ":" in line:
+            name, _, value = line.partition(":")
+            headers[name.strip().lower()] = value.strip()
+    return headers, body.strip()
+
+
+def _parse_http_response_part(text: str, content_id: Optional[str]) -> Optional[BatchItemResponse]:
+    lines = text.replace("\r\n", "\n").splitlines()
+    if not lines:
+        return None
+    status_line = ""
+    idx = 0
+    for i, line in enumerate(lines):
+        if line.startswith("HTTP/"):
+            status_line = line
+            idx = i + 1
+            break
+    if not status_line:
+        return None
+    parts = status_line.split(" ", 2)
+    if len(parts) < 2:
+        return None
+    try:
+        status_code = int(parts[1])
+    except ValueError:
+        return None
+    resp_headers: Dict[str, str] = {}
+    body_start = idx
+    for i in range(idx, len(lines)):
+        if lines[i] == "":
+            body_start = i + 1
+            break
+        if ":" in lines[i]:
+            name, _, value = lines[i].partition(":")
+            resp_headers[name.strip().lower()] = value.strip()
+    entity_id: Optional[str] = None
+    odata_id = resp_headers.get("odata-entityid", "")
+    if odata_id:
+        m = _GUID_RE.search(odata_id)
+        if m:
+            entity_id = m.group(0)
+    body_text = "\n".join(lines[body_start:]).strip()
+    data: Optional[Dict[str, Any]] = None
+    error_message: Optional[str] = None
+    error_code: Optional[str] = None
+    if body_text:
+        try:
+            parsed = json.loads(body_text)
+            if isinstance(parsed, dict):
+                err = parsed.get("error")
+                if isinstance(err, dict):
+                    error_message = err.get("message")
+                    error_code = err.get("code")
+                else:
+                    data = parsed
+        except (json.JSONDecodeError, ValueError):
+            pass
+    return BatchItemResponse(
+        status_code=status_code,
+        content_id=content_id,
+        entity_id=entity_id,
+        data=data,
+        error_message=error_message,
+        error_code=error_code,
+    )