_odata.py

# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

"""Dataverse Web API client with CRUD, SQL query, and table/column metadata management."""

from __future__ import annotations

from typing import Any, Dict, Optional, List, Union, Iterable, Callable
from enum import Enum
from dataclasses import dataclass, field
import unicodedata
import time
import re
import json
import uuid
import warnings
from datetime import datetime, timezone
import importlib.resources as ir
from contextlib import contextmanager
from contextvars import ContextVar

from urllib.parse import quote as _url_quote, parse_qs, urlparse

from ..core._http import _HttpClient
from ._upload import _FileUploadMixin
from ._relationships import _RelationshipOperationsMixin
from ..models.relationship import (
    LookupAttributeMetadata,
    OneToManyRelationshipMetadata,
    CascadeConfiguration,
)
from ..models.labels import Label, LocalizedLabel
from ..common.constants import CASCADE_BEHAVIOR_REMOVE_LINK
from ..core.errors import *
from ._raw_request import _RawRequest
from ..core._error_codes import (
    _http_subcode,
    _is_transient_status,
    VALIDATION_SQL_NOT_STRING,
    VALIDATION_SQL_EMPTY,
    VALIDATION_UNSUPPORTED_COLUMN_TYPE,
    METADATA_ENTITYSET_NOT_FOUND,
    METADATA_ENTITYSET_NAME_MISSING,
    METADATA_TABLE_NOT_FOUND,
    METADATA_TABLE_ALREADY_EXISTS,
    METADATA_COLUMN_NOT_FOUND,
    VALIDATION_UNSUPPORTED_CACHE_KIND,
)

from .. import __version__ as _SDK_VERSION

_USER_AGENT = f"DataverseSvcPythonClient:{_SDK_VERSION}"
_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}")
_CALL_SCOPE_CORRELATION_ID: ContextVar[Optional[str]] = ContextVar("_CALL_SCOPE_CORRELATION_ID", default=None)
_DEFAULT_EXPECTED_STATUSES: tuple[int, ...] = (200, 201, 202, 204)


def _extract_pagingcookie(next_link: str) -> Optional[str]:
    """Extract the raw pagingcookie value from a SQL ``@odata.nextLink`` URL.

    The Dataverse SQL endpoint has a server-side bug where the pagingcookie
    (containing first/last record GUIDs) does not advance between pages even
    though ``pagenumber`` increments. Detecting a repeated cookie lets the
    pagination loop break instead of looping indefinitely.

    Returns the pagingcookie string if present, or ``None`` if not found.
    """
    try:
        qs = parse_qs(urlparse(next_link).query)
        skiptoken = qs.get("$skiptoken", [None])[0]
        if not skiptoken:
            return None
        # parse_qs already URL-decodes the value once, giving the outer XML with
        # pagingcookie still percent-encoded (e.g. pagingcookie="%3ccookie...").
        # A second decode is intentionally omitted: decoding again would turn %22
        # into " inside the cookie XML, breaking the regex and causing every page
        # to extract the same truncated prefix regardless of the actual GUIDs.
        m = re.search(r'pagingcookie="([^"]+)"', skiptoken)
        if m:
            return m.group(1)
    except Exception:
        pass
    return None


@dataclass
class _RequestContext:
    """Structured request context used by ``_request`` to clarify payload and metadata."""

    method: str
    url: str
    expected: tuple[int, ...] = _DEFAULT_EXPECTED_STATUSES
    headers: Optional[Dict[str, str]] = None
    kwargs: Dict[str, Any] = field(default_factory=dict)

    @classmethod
    def build(
        cls,
        method: str,
        url: str,
        *,
        expected: tuple[int, ...] = _DEFAULT_EXPECTED_STATUSES,
        merge_headers: Optional[Callable[[Optional[Dict[str, str]]], Dict[str, str]]] = None,
        **kwargs: Any,
    ) -> "_RequestContext":
        headers = kwargs.get("headers")
        headers = merge_headers(headers) if merge_headers else (headers or {})
        headers.setdefault("x-ms-client-request-id", str(uuid.uuid4()))
        headers.setdefault("x-ms-correlation-id", _CALL_SCOPE_CORRELATION_ID.get())
        kwargs["headers"] = headers
        return cls(
            method=method,
            url=url,
            expected=expected,
            headers=headers,
            kwargs=kwargs or {},
        )


class _ODataClient(_FileUploadMixin, _RelationshipOperationsMixin):
    """Dataverse Web API client: CRUD, SQL-over-API, and table metadata helpers."""

    @staticmethod
    def _escape_odata_quotes(value: str) -> str:
        """Escape single quotes for OData queries (by doubling them)."""
        return value.replace("'", "''")

    @staticmethod
    def _normalize_cache_key(table_schema_name: str) -> str:
        """Normalize table_schema_name to lowercase for case-insensitive cache keys."""
        return table_schema_name.lower() if isinstance(table_schema_name, str) else ""

    @staticmethod
    def _lowercase_keys(record: Dict[str, Any]) -> Dict[str, Any]:
        """Convert all dictionary keys to lowercase for case-insensitive column names.

        Dataverse LogicalNames for attributes are stored lowercase, but users may
        provide PascalCase names (matching SchemaName). This normalizes the input.

        Keys containing ``@odata.`` (e.g. ``new_CustomerId@odata.bind``) are
        preserved as-is because the navigation property portion before ``@``
        must retain its original casing (case-sensitive navigation property name).  The OData
        parser validates ``@odata.bind`` property names **case-sensitively**
        against the entity's declared navigation properties, so lowercasing
        these keys causes ``400 - undeclared property`` errors.
        """
        if not isinstance(record, dict):
            return record
        return {k.lower() if isinstance(k, str) and "@odata." not in k else k: v for k, v in record.items()}

    @staticmethod
    def _lowercase_list(items: Optional[List[str]]) -> Optional[List[str]]:
        """Convert all strings in a list to lowercase for case-insensitive column names.

        Used for $select, $orderby, $expand parameters where column names must be lowercase.
        """
        if not items:
            return items
        return [item.lower() if isinstance(item, str) else item for item in items]

    def __init__(
        self,
        auth,
        base_url: str,
        config=None,
        session=None,
    ) -> None:
        """Initialize the OData client.

        Sets up authentication, base URL, configuration, and internal caches.

        :param auth: Authentication manager providing ``_acquire_token(scope)`` that returns an object with ``access_token``.
        :type auth: ~PowerPlatform.Dataverse.core._auth._AuthManager
        :param base_url: Organization base URL (e.g. ``"https://<org>.crm.dynamics.com"``).
        :type base_url: ``str``
        :param config: Optional Dataverse configuration (HTTP retry, backoff, timeout, language code). If omitted ``DataverseConfig.from_env()`` is used.
        :type config: ~PowerPlatform.Dataverse.core.config.DataverseConfig | ``None``
        :param session: Optional ``requests.Session`` for HTTP connection pooling.
        :type session: :class:`requests.Session` | ``None``
        :raises ValueError: If ``base_url`` is empty after stripping.
        """
        self.auth = auth
        self.base_url = (base_url or "").rstrip("/")
        if not self.base_url:
            raise ValueError("base_url is required.")
        self.api = f"{self.base_url}/api/data/v9.2"
        self.config = (
            config
            or __import__(
                "PowerPlatform.Dataverse.core.config", fromlist=["DataverseConfig"]
            ).DataverseConfig.from_env()
        )
        self._http = _HttpClient(
            retries=self.config.http_retries,
            backoff=self.config.http_backoff,
            timeout=self.config.http_timeout,
            session=session,
        )
        # Cache: normalized table_schema_name (lowercase) -> entity set name (plural) resolved from metadata
        self._logical_to_entityset_cache: dict[str, str] = {}
        # Cache: normalized table_schema_name (lowercase) -> primary id attribute (e.g. accountid)
        self._logical_primaryid_cache: dict[str, str] = {}
        self._picklist_label_cache: dict[str, dict] = {}
        self._picklist_cache_ttl_seconds = 3600  # 1 hour TTL

    @contextmanager
    def _call_scope(self):
        """Context manager to generate a new correlation id for each SDK call scope."""
        shared_id = str(uuid.uuid4())
        token = _CALL_SCOPE_CORRELATION_ID.set(shared_id)
        try:
            yield shared_id
        finally:
            _CALL_SCOPE_CORRELATION_ID.reset(token)

    def close(self) -> None:
        """Close the OData client and release resources.

        Clears all internal caches and closes the underlying HTTP client.
        Safe to call multiple times.
        """
        self._logical_to_entityset_cache.clear()
        self._logical_primaryid_cache.clear()
        self._picklist_label_cache.clear()
        if self._http is not None:
            self._http.close()

    def _headers(self) -> Dict[str, str]:
        """Build standard OData headers with bearer auth."""
        scope = f"{self.base_url}/.default"
        token = self.auth._acquire_token(scope).access_token
        return {
            "Authorization": f"Bearer {token}",
            "Accept": "application/json",
            "Content-Type": "application/json",
            "OData-MaxVersion": "4.0",
            "OData-Version": "4.0",
            "User-Agent": _USER_AGENT,
        }

    def _merge_headers(self, headers: Optional[Dict[str, str]] = None) -> Dict[str, str]:
        base = self._headers()
        if not headers:
            return base
        merged = base.copy()
        merged.update(headers)
        return merged

    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, ...] = _DEFAULT_EXPECTED_STATUSES, **kwargs):
        request_context = _RequestContext.build(
            method,
            url,
            expected=expected,
            merge_headers=self._merge_headers,
            **kwargs,
        )

        r = self._raw_request(request_context.method, request_context.url, **request_context.kwargs)
        if r.status_code in request_context.expected:
            return r
        response_headers = getattr(r, "headers", {}) or {}
        body_excerpt = (getattr(r, "text", "") or "")[:200]
        svc_code = None
        msg = f"HTTP {r.status_code}"
        try:
            data = r.json() if getattr(r, "text", None) else {}
            if isinstance(data, dict):
                inner = data.get("error")
                if isinstance(inner, dict):
                    svc_code = inner.get("code")
                    imsg = inner.get("message")
                    if isinstance(imsg, str) and imsg.strip():
                        msg = imsg.strip()
                else:
                    imsg2 = data.get("message")
                    if isinstance(imsg2, str) and imsg2.strip():
                        msg = imsg2.strip()
        except Exception:
            pass
        sc = r.status_code
        subcode = _http_subcode(sc)
        request_id = (
            response_headers.get("x-ms-service-request-id")
            or response_headers.get("req_id")
            or response_headers.get("x-ms-request-id")
        )
        traceparent = response_headers.get("traceparent")
        ra = response_headers.get("Retry-After")
        retry_after = None
        if ra:
            try:
                retry_after = int(ra)
            except Exception:
                retry_after = None
        is_transient = _is_transient_status(sc)
        raise HttpError(
            msg,
            status_code=sc,
            subcode=subcode,
            service_error_code=svc_code,
            correlation_id=request_context.headers.get(
                "x-ms-correlation-id"
            ),  # this is a value set on client side, although it's logged on server side too
            client_request_id=request_context.headers.get(
                "x-ms-client-request-id"
            ),  # this is a value set on client side, although it's logged on server side too
            service_request_id=request_id,
            traceparent=traceparent,
            body_excerpt=body_excerpt,
            retry_after=retry_after,
            is_transient=is_transient,
        )

    def _execute_raw(self, req: _RawRequest, *, expected: tuple[int, ...] = _DEFAULT_EXPECTED_STATUSES):
        """Execute a ``_RawRequest`` and return the HTTP response.

        Encodes the pre-serialised body (if present) as UTF-8 and merges any
        per-request headers into the standard OData header set.
        """
        kwargs: Dict[str, Any] = {}
        if req.body is not None:
            kwargs["data"] = req.body.encode("utf-8")
        if req.headers:
            kwargs["headers"] = req.headers
        return self._request(req.method.lower(), req.url, expected=expected, **kwargs)

    # --- CRUD Internal functions ---
    def _create(self, entity_set: str, table_schema_name: str, record: Dict[str, Any]) -> str:
        """Create a single record and return its GUID.

        :param entity_set: Resolved entity set (plural) name.
        :type entity_set: ``str``
        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param record: Attribute payload mapped by logical column names.
        :type record: ``dict[str, Any]``

        :return: Created record GUID.
        :rtype: ``str``

        .. note::
           Relies on ``OData-EntityId`` (canonical) or ``Location`` response header. No response body parsing is performed. Raises ``RuntimeError`` if neither header contains a GUID.
        """
        r = self._execute_raw(self._build_create(entity_set, table_schema_name, record))
        ent_loc = r.headers.get("OData-EntityId") or r.headers.get("OData-EntityID")
        if ent_loc:
            m = _GUID_RE.search(ent_loc)
            if m:
                return m.group(0)
        loc = r.headers.get("Location")
        if loc:
            m = _GUID_RE.search(loc)
            if m:
                return m.group(0)
        header_keys = ", ".join(sorted(r.headers.keys()))
        raise RuntimeError(
            f"Create response missing GUID in OData-EntityId/Location headers (status={getattr(r,'status_code', '?')}). Headers: {header_keys}"
        )

    def _create_multiple(self, entity_set: str, table_schema_name: str, records: List[Dict[str, Any]]) -> List[str]:
        """Create multiple records using the collection-bound ``CreateMultiple`` action.

        :param entity_set: Resolved entity set (plural) name.
        :type entity_set: ``str``
        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param records: Payload dictionaries mapped by column schema names.
        :type records: ``list[dict[str, Any]]``

        :return: List of created record GUIDs (may be empty if response lacks IDs).
        :rtype: ``list[str]``

        .. note::
           Logical type stamping: if any payload omits ``@odata.type`` the client injects ``Microsoft.Dynamics.CRM.<table_logical_name>``. If all payloads already include ``@odata.type`` no modification occurs.
        """
        if not all(isinstance(r, dict) for r in records):
            raise TypeError("All items for multi-create must be dicts")
        r = self._execute_raw(self._build_create_multiple(entity_set, table_schema_name, records))
        try:
            body = r.json() if r.text else {}
        except ValueError:
            body = {}
        if not isinstance(body, dict):
            return []
        # Expected: { "Ids": [guid, ...] }
        ids = body.get("Ids")
        if isinstance(ids, list):
            return [i for i in ids if isinstance(i, str)]

        value = body.get("value")
        if isinstance(value, list):
            # Extract IDs if possible
            out: List[str] = []
            for item in value:
                if isinstance(item, dict):
                    # Heuristic: look for a property ending with 'id'
                    for k, v in item.items():
                        if isinstance(k, str) and k.lower().endswith("id") and isinstance(v, str) and len(v) >= 32:
                            out.append(v)
                            break
            return out
        return []

    def _build_alternate_key_str(self, alternate_key: Dict[str, Any]) -> str:
        """Build an OData alternate key segment from a mapping of key names to values.

        String values are single-quoted and escaped; all other values are rendered as-is.

        :param alternate_key: Mapping of alternate key attribute names to their values.
            Must be a non-empty dict with string keys.
        :type alternate_key: ``dict[str, Any]``

        :return: Comma-separated key=value pairs suitable for use in a URL segment.
        :rtype: ``str``

        :raises ValueError: If ``alternate_key`` is empty.
        :raises TypeError: If any key in ``alternate_key`` is not a string.
        """
        if not alternate_key:
            raise ValueError("alternate_key must be a non-empty dict")
        bad_keys = [k for k in alternate_key if not isinstance(k, str)]
        if bad_keys:
            raise TypeError(f"alternate_key keys must be strings; got: {bad_keys!r}")
        parts = []
        for k, v in alternate_key.items():
            k_lower = k.lower() if isinstance(k, str) else k
            if isinstance(v, str):
                v_escaped = self._escape_odata_quotes(v)
                parts.append(f"{k_lower}='{v_escaped}'")
            else:
                parts.append(f"{k_lower}={v}")
        return ",".join(parts)

    def _upsert(
        self,
        entity_set: str,
        table_schema_name: str,
        alternate_key: Dict[str, Any],
        record: Dict[str, Any],
    ) -> None:
        """Upsert a single record using an alternate key.

        Issues a PATCH request to ``{entity_set}({key_pairs})`` where ``key_pairs``
        is the OData alternate key segment built from ``alternate_key``. Creates the
        record if it does not exist; updates it if it does.

        :param entity_set: Resolved entity set (plural) name.
        :type entity_set: ``str``
        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param alternate_key: Mapping of alternate key attribute names to their values
            used to identify the target record in the URL.
        :type alternate_key: ``dict[str, Any]``
        :param record: Attribute payload to set on the record.
        :type record: ``dict[str, Any]``

        :return: ``None``
        :rtype: ``None``
        """
        record = self._lowercase_keys(record)
        record = self._convert_labels_to_ints(table_schema_name, record)
        key_str = self._build_alternate_key_str(alternate_key)
        url = f"{self.api}/{entity_set}({key_str})"
        self._request("patch", url, json=record, expected=(200, 201, 204))

    def _upsert_multiple(
        self,
        entity_set: str,
        table_schema_name: str,
        alternate_keys: List[Dict[str, Any]],
        records: List[Dict[str, Any]],
    ) -> None:
        """Upsert multiple records using the collection-bound ``UpsertMultiple`` action.

        Each target is formed by merging the corresponding alternate key fields and record
        fields. The ``@odata.type`` annotation is injected automatically if absent.

        :param entity_set: Resolved entity set (plural) name.
        :type entity_set: ``str``
        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param alternate_keys: List of alternate key dictionaries, one per record.
            Order is significant: ``alternate_keys[i]`` must correspond to ``records[i]``.
            Python ``list`` preserves insertion order, so the correspondence is guaranteed
            as long as both lists are built from the same source in the same order.
        :type alternate_keys: ``list[dict[str, Any]]``
        :param records: List of record payload dictionaries, one per record.
            Must be the same length as ``alternate_keys``.
        :type records: ``list[dict[str, Any]]``

        :return: ``None``
        :rtype: ``None``

        :raises ValueError: If ``alternate_keys`` and ``records`` differ in length, or if
            any record payload contains an alternate key field with a conflicting value.
        """
        if len(alternate_keys) != len(records):
            raise ValueError(
                f"alternate_keys and records must have the same length " f"({len(alternate_keys)} != {len(records)})"
            )
        logical_name = table_schema_name.lower()
        targets: List[Dict[str, Any]] = []
        for alt_key, record in zip(alternate_keys, records):
            alt_key_lower = self._lowercase_keys(alt_key)
            record_processed = self._lowercase_keys(record)
            record_processed = self._convert_labels_to_ints(table_schema_name, record_processed)
            conflicting = {
                k for k in set(alt_key_lower) & set(record_processed) if alt_key_lower[k] != record_processed[k]
            }
            if conflicting:
                raise ValueError(f"record payload conflicts with alternate_key on fields: {sorted(conflicting)!r}")
            if "@odata.type" not in record_processed:
                record_processed["@odata.type"] = f"Microsoft.Dynamics.CRM.{logical_name}"
            key_str = self._build_alternate_key_str(alt_key)
            record_processed["@odata.id"] = f"{entity_set}({key_str})"
            targets.append(record_processed)
        payload = {"Targets": targets}
        url = f"{self.api}/{entity_set}/Microsoft.Dynamics.CRM.UpsertMultiple"
        self._request("post", url, json=payload, expected=(200, 201, 204))

    # --- Derived helpers for high-level client ergonomics ---
    def _primary_id_attr(self, table_schema_name: str) -> str:
        """Return primary key attribute using metadata; error if unavailable."""
        cache_key = self._normalize_cache_key(table_schema_name)
        pid = self._logical_primaryid_cache.get(cache_key)
        if pid:
            return pid
        # Resolve metadata (populates _logical_primaryid_cache or raises if table_schema_name unknown)
        self._entity_set_from_schema_name(table_schema_name)
        pid2 = self._logical_primaryid_cache.get(cache_key)
        if pid2:
            return pid2
        raise RuntimeError(
            f"PrimaryIdAttribute not resolved for table_schema_name '{table_schema_name}'. Metadata did not include PrimaryIdAttribute."
        )

    def _update_by_ids(
        self, table_schema_name: str, ids: List[str], changes: Union[Dict[str, Any], List[Dict[str, Any]]]
    ) -> None:
        """Update many records by GUID list using the collection-bound ``UpdateMultiple`` action.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param ids: GUIDs of target records.
        :type ids: ``list[str]``
        :param changes: Broadcast patch (``dict``) applied to all IDs, or list of per-record patches (1:1 with ``ids``).
        :type changes: ``dict`` | ``list[dict]``

        :return: ``None``
        :rtype: ``None``
        """
        if not isinstance(ids, list):
            raise TypeError("ids must be list[str]")
        if not ids:
            return None
        pk_attr = self._primary_id_attr(table_schema_name)
        entity_set = self._entity_set_from_schema_name(table_schema_name)
        if isinstance(changes, dict):
            batch = [{pk_attr: rid, **changes} for rid in ids]
            self._update_multiple(entity_set, table_schema_name, batch)
            return None
        if not isinstance(changes, list):
            raise TypeError("changes must be dict or list[dict]")
        if len(changes) != len(ids):
            raise ValueError("Length of changes list must match length of ids list")
        batch: List[Dict[str, Any]] = []
        for rid, patch in zip(ids, changes):
            if not isinstance(patch, dict):
                raise TypeError("Each patch must be a dict")
            batch.append({pk_attr: rid, **patch})
        self._update_multiple(entity_set, table_schema_name, batch)
        return None

    def _delete_multiple(
        self,
        table_schema_name: str,
        ids: List[str],
    ) -> Optional[str]:
        """Delete many records by GUID list via the ``BulkDelete`` action.

        :param logical_name: Logical (singular) entity name.
        :type logical_name: ``str``
        :param ids: GUIDs of records to delete.
        :type ids: ``list[str]``

        :return: BulkDelete asynchronous job identifier when executed in bulk; ``None`` if no IDs provided or single deletes performed.
        :rtype: ``str`` | ``None``
        """
        targets = [rid for rid in ids if rid]
        if not targets:
            return None
        response = self._execute_raw(
            self._build_delete_multiple(table_schema_name, targets),
            expected=(200, 202, 204),
        )
        job_id = None
        try:
            body = response.json() if response.text else {}
        except ValueError:
            body = {}
        if isinstance(body, dict):
            job_id = body.get("JobId")
        return job_id

    def _format_key(self, key: str) -> str:
        k = key.strip()
        if k.startswith("(") and k.endswith(")"):
            return k
        # Escape single quotes in alternate key values
        if "=" in k and "'" in k:

            def esc(match):
                # match.group(1) is the key, match.group(2) is the value
                return f"{match.group(1)}='{self._escape_odata_quotes(match.group(2))}'"

            k = re.sub(r"(\w+)=\'([^\']*)\'", esc, k)
            return f"({k})"
        if len(k) == 36 and "-" in k:
            return f"({k})"
        return f"({k})"

    def _update(self, table_schema_name: str, key: str, data: Dict[str, Any]) -> None:
        """Update an existing record by GUID.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param key: Record GUID (with or without parentheses).
        :type key: ``str``
        :param data: Partial entity payload (attributes to patch).
        :type data: ``dict[str, Any]``
        :return: ``None``
        :rtype: ``None``
        """
        self._execute_raw(self._build_update(table_schema_name, key, data))

    def _update_multiple(self, entity_set: str, table_schema_name: str, records: List[Dict[str, Any]]) -> None:
        """Bulk update existing records via the collection-bound ``UpdateMultiple`` action.

        :param entity_set: Resolved entity set (plural) name.
        :type entity_set: ``str``
        :param table_schema_name: Schema name of the table, e.g. "new_MyTestTable".
        :type table_schema_name: ``str``
        :param records: List of patch dictionaries. Each must include the true primary key attribute (e.g. ``accountid``) and one or more fields to update.
        :type records: ``list[dict[str, Any]]``
        :return: ``None``
        :rtype: ``None``

        .. note::
           - Endpoint: ``POST /{entity_set}/Microsoft.Dynamics.CRM.UpdateMultiple`` with body ``{"Targets": [...]}``.
           - Transactional semantics: if any individual update fails, the entire request rolls back.
           - Response content is ignored; no stable contract for returned IDs/representations.
           - Caller must supply the correct primary key attribute (e.g. ``accountid``) in every record.
        """
        if not isinstance(records, list) or not records or not all(isinstance(r, dict) for r in records):
            raise TypeError("records must be a non-empty list[dict]")
        self._execute_raw(self._build_update_multiple_from_records(entity_set, table_schema_name, records))
        return None

    def _delete(self, table_schema_name: str, key: str) -> None:
        """Delete a record by GUID.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param key: Record GUID (with or without parentheses)
        :type key: ``str``

        :return: ``None``
        :rtype: ``None``
        """
        self._execute_raw(self._build_delete(table_schema_name, key))

    def _get(self, table_schema_name: str, key: str, select: Optional[List[str]] = None) -> Dict[str, Any]:
        """Retrieve a single record.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param key: Record GUID (with or without parentheses).
        :type key: ``str``
        :param select: Columns to select; joined with commas into $select.
        :type select: ``list[str]`` | ``None``

        :return: Retrieved record dictionary (may be empty if no selected attributes).
        :rtype: ``dict[str, Any]``
        """
        return self._execute_raw(self._build_get(table_schema_name, key, select=select)).json()

    def _get_multiple(
        self,
        table_schema_name: str,
        select: Optional[List[str]] = None,
        filter: Optional[str] = None,
        orderby: Optional[List[str]] = None,
        top: Optional[int] = None,
        expand: Optional[List[str]] = None,
        page_size: Optional[int] = None,
        count: bool = False,
        include_annotations: Optional[str] = None,
    ) -> Iterable[List[Dict[str, Any]]]:
        """Iterate records from an entity set, yielding one page (list of dicts) at a time.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param select: Columns to include (``$select``) or ``None``. Column names are automatically lowercased.
        :type select: ``list[str]`` | ``None``
        :param filter: OData ``$filter`` expression or ``None``. This is passed as-is without transformation. Users must provide lowercase logical column names (e.g., "statecode eq 0").
        :type filter: ``str`` | ``None``
        :param orderby: Order expressions (``$orderby``) or ``None``. Column names are automatically lowercased.
        :type orderby: ``list[str]`` | ``None``
        :param top: Max total records (applied on first request as ``$top``) or ``None``.
        :type top: ``int`` | ``None``
        :param expand: Navigation properties to expand (``$expand``) or ``None``. These are case-sensitive and passed as-is. Users must provide exact navigation property names from entity metadata.
        :type expand: ``list[str]`` | ``None``
        :param page_size: Per-page size hint via ``Prefer: odata.maxpagesize``.
        :type page_size: ``int`` | ``None``
        :param count: If ``True``, adds ``$count=true`` to include a total record count in the response.
        :type count: ``bool``
        :param include_annotations: OData annotation pattern for the ``Prefer: odata.include-annotations`` header (e.g. ``"*"`` or ``"OData.Community.Display.V1.FormattedValue"``), or ``None``.
        :type include_annotations: ``str`` | ``None``

        :return: Iterator yielding pages (each page is a ``list`` of record dicts).
        :rtype: ``Iterable[list[dict[str, Any]]]``
        """

        extra_headers: Dict[str, str] = {}
        prefer_parts: List[str] = []
        if page_size is not None:
            ps = int(page_size)
            if ps > 0:
                prefer_parts.append(f"odata.maxpagesize={ps}")
        if include_annotations:
            prefer_parts.append(f'odata.include-annotations="{include_annotations}"')
        if prefer_parts:
            extra_headers["Prefer"] = ",".join(prefer_parts)

        def _do_request(url: str, *, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
            headers = extra_headers if extra_headers else None
            r = self._request("get", url, headers=headers, params=params)
            try:
                return r.json()
            except ValueError:
                return {}

        entity_set = self._entity_set_from_schema_name(table_schema_name)
        base_url = f"{self.api}/{entity_set}"
        params: Dict[str, Any] = {}
        if select:
            # Lowercase column names for case-insensitive matching
            params["$select"] = ",".join(self._lowercase_list(select))
        if filter:
            # Filter is passed as-is; users must use lowercase column names in filter expressions
            params["$filter"] = filter
        if orderby:
            # Lowercase column names for case-insensitive matching
            params["$orderby"] = ",".join(self._lowercase_list(orderby))
        if expand:
            # Lowercase navigation property names for case-insensitive matching
            params["$expand"] = ",".join(expand)
        if top is not None:
            params["$top"] = int(top)
        if count:
            params["$count"] = "true"

        data = _do_request(base_url, params=params)
        items = data.get("value") if isinstance(data, dict) else None
        if isinstance(items, list) and items:
            yield [x for x in items if isinstance(x, dict)]

        next_link = None
        if isinstance(data, dict):
            next_link = data.get("@odata.nextLink") or data.get("odata.nextLink")

        while next_link:
            data = _do_request(next_link)
            items = data.get("value") if isinstance(data, dict) else None
            if isinstance(items, list) and items:
                yield [x for x in items if isinstance(x, dict)]
            next_link = data.get("@odata.nextLink") or data.get("odata.nextLink") if isinstance(data, dict) else None

    # --------------------------- SQL Custom API -------------------------
    def _query_sql(self, sql: str) -> list[dict[str, Any]]:
        """Execute a read-only SQL SELECT using the Dataverse Web API ``?sql=`` capability.

        :param sql: Single SELECT statement within the supported subset.
        :type sql: ``str``

        :return: Result rows (empty list if none).
        :rtype: ``list[dict[str, Any]]``

        :raises ValidationError: If ``sql`` is not a ``str`` or is empty.
        :raises MetadataError: If logical table name resolution fails.

        .. note::
           Endpoint form: ``GET /{entity_set}?sql=<encoded select>``. The client extracts the logical table name, resolves the entity set (metadata cached), then issues the request. Only a constrained SELECT subset is supported by the platform.
        """
        if not isinstance(sql, str):
            raise ValidationError("sql must be a string", subcode=VALIDATION_SQL_NOT_STRING)
        if not sql.strip():
            raise ValidationError("sql must be a non-empty string", subcode=VALIDATION_SQL_EMPTY)
        sql = sql.strip()
        r = self._execute_raw(self._build_sql(sql))
        try:
            body = r.json()
        except ValueError:
            return []

        # Collect first page
        results: list[dict[str, Any]] = []
        if isinstance(body, list):
            return [row for row in body if isinstance(row, dict)]
        if not isinstance(body, dict):
            return results

        value = body.get("value")
        if isinstance(value, list):
            results = [row for row in value if isinstance(row, dict)]

        # Follow pagination links until exhausted
        raw_link = body.get("@odata.nextLink") or body.get("odata.nextLink")
        next_link: str | None = raw_link if isinstance(raw_link, str) else None
        visited: set[str] = set()
        seen_cookies: set[str] = set()
        while next_link:
            # Guard 1: exact URL cycle (same next_link returned twice)
            if next_link in visited:
                warnings.warn(
                    f"SQL pagination stopped after {len(results)} rows — "
                    "the Dataverse server returned the same nextLink URL twice, "
                    "indicating an infinite pagination cycle. "
                    "Returning the rows collected so far. "
                    "To avoid pagination entirely, add a TOP clause to your query.",
                    RuntimeWarning,
                    stacklevel=4,
                )
                break
            visited.add(next_link)
            # Guard 2: server-side bug where pagingcookie does not advance between
            # pages (pagenumber increments but cookie GUIDs stay the same), which
            # causes an infinite loop even though URLs differ.
            cookie = _extract_pagingcookie(next_link)
            if cookie is not None:
                if cookie in seen_cookies:
                    warnings.warn(
                        f"SQL pagination stopped after {len(results)} rows — "
                        "the Dataverse server returned the same pagingcookie twice "
                        "(pagenumber incremented but the paging position did not advance). "
                        "This is a server-side bug. Returning the rows collected so far. "
                        "To avoid pagination entirely, add a TOP clause to your query.",
                        RuntimeWarning,
                        stacklevel=4,
                    )
                    break
                seen_cookies.add(cookie)
            try:
                page_resp = self._request("get", next_link)
            except Exception as exc:
                warnings.warn(
                    f"SQL pagination stopped after {len(results)} rows — "
                    f"the next-page request failed: {exc}. "
                    "Add a TOP clause to your query to limit results to a single page.",
                    RuntimeWarning,
                    stacklevel=5,
                )
                break
            try:
                page_body = page_resp.json()
            except ValueError as exc:
                warnings.warn(
                    f"SQL pagination stopped after {len(results)} rows — "
                    f"the next-page response was not valid JSON: {exc}. "
                    "Add a TOP clause to your query to limit results to a single page.",
                    RuntimeWarning,
                    stacklevel=5,
                )
                break
            if not isinstance(page_body, dict):
                break
            page_value = page_body.get("value")
            if not isinstance(page_value, list) or not page_value:
                break
            results.extend(row for row in page_value if isinstance(row, dict))
            raw_link = page_body.get("@odata.nextLink") or page_body.get("odata.nextLink")
            next_link = raw_link if isinstance(raw_link, str) else None

        return results

    @staticmethod
    def _extract_logical_table(sql: str) -> str:
        """Extract the logical table name after the first standalone FROM.

        Examples:
            SELECT * FROM account
            SELECT col1, startfrom FROM new_sampleitem WHERE col1 = 1

        """
        if not isinstance(sql, str):
            raise ValueError("sql must be a string")
        # Mask out single-quoted string literals to avoid matching FROM inside them.
        masked = re.sub(r"'([^']|'')*'", "'x'", sql)
        pattern = r"\bfrom\b\s+([A-Za-z0-9_]+)"  # minimal, single-line regex
        m = re.search(pattern, masked, flags=re.IGNORECASE)
        if not m:
            raise ValueError("Unable to determine table logical name from SQL (expected 'FROM <name>').")
        return m.group(1).lower()

    # ---------------------- Entity set resolution -----------------------
    def _entity_set_from_schema_name(self, table_schema_name: str) -> str:
        """Resolve entity set name (plural) from a schema name (singular) name using metadata.

        Caches results for subsequent queries. Case-insensitive.
        """
        if not table_schema_name:
            raise ValueError("table schema name required")

        # Use normalized (lowercase) key for cache lookup
        cache_key = self._normalize_cache_key(table_schema_name)
        cached = self._logical_to_entityset_cache.get(cache_key)
        if cached:
            return cached
        url = f"{self.api}/EntityDefinitions"
        # LogicalName in Dataverse is stored in lowercase, so we need to lowercase for the filter
        logical_lower = table_schema_name.lower()
        logical_escaped = self._escape_odata_quotes(logical_lower)
        params = {
            "$select": "LogicalName,EntitySetName,PrimaryIdAttribute",
            "$filter": f"LogicalName eq '{logical_escaped}'",
        }
        r = self._request("get", url, params=params)
        try:
            body = r.json()
            items = body.get("value", []) if isinstance(body, dict) else []
        except ValueError:
            items = []
        if not items:
            plural_hint = (
                " (did you pass a plural entity set name instead of the singular table schema name?)"
                if table_schema_name.endswith("s") and not table_schema_name.endswith("ss")
                else ""
            )
            raise MetadataError(
                f"Unable to resolve entity set for table schema name '{table_schema_name}'. Provide the singular table schema name.{plural_hint}",
                subcode=METADATA_ENTITYSET_NOT_FOUND,
            )
        md = items[0]
        es = md.get("EntitySetName")
        if not es:
            raise MetadataError(
                f"Metadata response missing EntitySetName for table schema name '{table_schema_name}'.",
                subcode=METADATA_ENTITYSET_NAME_MISSING,
            )
        self._logical_to_entityset_cache[cache_key] = es
        primary_id_attr = md.get("PrimaryIdAttribute")
        if isinstance(primary_id_attr, str) and primary_id_attr:
            self._logical_primaryid_cache[cache_key] = primary_id_attr
        return es

    # ---------------------- Table metadata helpers ----------------------
    def _label(self, text: str) -> Dict[str, Any]:
        lang = int(self.config.language_code)
        return {
            "@odata.type": "Microsoft.Dynamics.CRM.Label",
            "LocalizedLabels": [
                {
                    "@odata.type": "Microsoft.Dynamics.CRM.LocalizedLabel",
                    "Label": text,
                    "LanguageCode": lang,
                }
            ],
        }

    def _to_pascal(self, name: str) -> str:
        parts = re.split(r"[^A-Za-z0-9]+", name)
        return "".join(p[:1].upper() + p[1:] for p in parts if p)

    def _get_entity_by_table_schema_name(
        self,
        table_schema_name: str,
        headers: Optional[Dict[str, str]] = None,
    ) -> Optional[Dict[str, Any]]:
        """Get entity metadata by table schema name. Case-insensitive.

        Note: LogicalName is stored lowercase in Dataverse, so we lowercase the input
        for case-insensitive matching. The response includes SchemaName, LogicalName,
        EntitySetName, and MetadataId.
        """
        url = f"{self.api}/EntityDefinitions"
        # LogicalName is stored lowercase, so we lowercase the input for lookup
        logical_lower = table_schema_name.lower()
        logical_escaped = self._escape_odata_quotes(logical_lower)
        params = {
            "$select": "MetadataId,LogicalName,SchemaName,EntitySetName,PrimaryNameAttribute,PrimaryIdAttribute",
            "$filter": f"LogicalName eq '{logical_escaped}'",
        }
        r = self._request("get", url, params=params, headers=headers)
        items = r.json().get("value", [])
        return items[0] if items else None

    def _create_entity(
        self,
        table_schema_name: str,
        display_name: str,
        attributes: List[Dict[str, Any]],
        solution_unique_name: Optional[str] = None,
    ) -> Dict[str, Any]:
        url = f"{self.api}/EntityDefinitions"
        payload = {
            "@odata.type": "Microsoft.Dynamics.CRM.EntityMetadata",
            "SchemaName": table_schema_name,
            "DisplayName": self._label(display_name),
            "DisplayCollectionName": self._label(display_name + "s"),
            "Description": self._label(f"Custom entity for {display_name}"),
            "OwnershipType": "UserOwned",
            "HasActivities": False,
            "HasNotes": True,
            "IsActivity": False,
            "Attributes": attributes,
        }
        params = None
        if solution_unique_name:
            params = {"SolutionUniqueName": solution_unique_name}
        self._request("post", url, json=payload, params=params)
        ent = self._get_entity_by_table_schema_name(
            table_schema_name,
            headers={"Consistency": "Strong"},
        )
        if not ent or not ent.get("EntitySetName"):
            raise RuntimeError(
                f"Failed to create or retrieve entity '{table_schema_name}' (EntitySetName not available)."
            )
        if not ent.get("MetadataId"):
            raise RuntimeError(f"MetadataId missing after creating entity '{table_schema_name}'.")
        return ent

    def _get_attribute_metadata(
        self,
        entity_metadata_id: str,
        column_name: str,
        extra_select: Optional[str] = None,
    ) -> Optional[Dict[str, Any]]:
        # Convert to lowercase logical name for lookup
        logical_name = column_name.lower()
        attr_escaped = self._escape_odata_quotes(logical_name)
        url = f"{self.api}/EntityDefinitions({entity_metadata_id})/Attributes"
        select_fields = ["MetadataId", "LogicalName", "SchemaName"]
        if extra_select:
            for piece in extra_select.split(","):
                piece = piece.strip()
                if not piece or piece in select_fields:
                    continue
                if piece.startswith("@"):
                    continue
                if piece not in select_fields:
                    select_fields.append(piece)
        params = {
            "$select": ",".join(select_fields),
            "$filter": f"LogicalName eq '{attr_escaped}'",
        }
        r = self._request("get", url, params=params)
        try:
            body = r.json() if r.text else {}
        except ValueError:
            return None
        items = body.get("value") if isinstance(body, dict) else None
        if isinstance(items, list) and items:
            item = items[0]
            if isinstance(item, dict):
                return item
        return None

    def _wait_for_attribute_visibility(
        self,
        entity_set: str,
        attribute_name: str,
        delays: tuple = (0, 3, 10, 20),
    ) -> None:
        """Wait for a newly created attribute to become visible in the data API.

        After creating an attribute via the metadata API, there can be a delay before
        it becomes queryable in the data API. This method polls the entity set with
        the attribute in the $select clause until it succeeds or all delays are exhausted.
        """
        # Convert to lowercase logical name for URL
        logical_name = attribute_name.lower()
        probe_url = f"{self.api}/{entity_set}?$top=1&$select={logical_name}"
        last_error = None
        total_wait = sum(delays)

        for delay in delays:
            if delay:
                time.sleep(delay)
            try:
                self._request("get", probe_url)
                return
            except Exception as ex:
                last_error = ex
                continue

        # All retries exhausted - raise with context
        raise RuntimeError(
            f"Attribute '{logical_name}' did not become visible in the data API "
            f"after {total_wait} seconds (exhausted all retries)."
        ) from last_error

    # ---------------------- Enum / Option Set helpers ------------------
    def _build_localizedlabels_payload(self, translations: Dict[int, str]) -> Dict[str, Any]:
        """Build a Dataverse Label object from {<language_code>: <text>} entries.

        Ensures at least one localized label. Does not deduplicate language codes; last wins.
        """
        locs: List[Dict[str, Any]] = []
        for lang, text in translations.items():
            if not isinstance(lang, int):
                raise ValueError(f"Language code '{lang}' must be int")
            if not isinstance(text, str) or not text.strip():
                raise ValueError(f"Label for lang {lang} must be non-empty string")
            locs.append(
                {
                    "@odata.type": "Microsoft.Dynamics.CRM.LocalizedLabel",
                    "Label": text,
                    "LanguageCode": lang,
                }
            )
        if not locs:
            raise ValueError("At least one translation required")
        return {
            "@odata.type": "Microsoft.Dynamics.CRM.Label",
            "LocalizedLabels": locs,
        }

    def _enum_optionset_payload(
        self, column_schema_name: str, enum_cls: type[Enum], is_primary_name: bool = False
    ) -> Dict[str, Any]:
        """Create local (IsGlobal=False) PicklistAttributeMetadata from an Enum subclass.

        Supports translation mapping via optional class attribute `__labels__`:
            __labels__ = { 1033: { "Active": "Active", "Inactive": "Inactive" },
                           1036: { "Active": "Actif",  "Inactive": "Inactif" } }

        Keys inside per-language dict may be either enum member objects or their names.
        If a language lacks a label for a member, member.name is used as fallback.
        The client's configured language code is always ensured to exist.
        """
        all_member_items = list(enum_cls.__members__.items())
        if not all_member_items:
            raise ValueError(f"Enum {enum_cls.__name__} has no members")

        # Duplicate detection
        value_to_first_name: Dict[int, str] = {}
        for name, member in all_member_items:
            val = getattr(member, "value", None)
            # Defer non-int validation to later loop for consistency
            if val in value_to_first_name and value_to_first_name[val] != name:
                raise ValueError(
                    f"Duplicate enum value {val} in {enum_cls.__name__} (names: {value_to_first_name[val]}, {name})"
                )
            value_to_first_name[val] = name

        members = list(enum_cls)
        # Validate integer values
        for m in members:
            if not isinstance(m.value, int):
                raise ValueError(f"Enum member '{m.name}' has non-int value '{m.value}' (only int values supported)")

        raw_labels = getattr(enum_cls, "__labels__", None)
        labels_by_lang: Dict[int, Dict[str, str]] = {}
        if raw_labels is not None:
            if not isinstance(raw_labels, dict):
                raise ValueError("__labels__ must be a dict {lang:int -> {member: label}}")
            # Build a helper map for value -> member name to resolve raw int keys
            value_to_name = {m.value: m.name for m in members}
            for lang, mapping in raw_labels.items():
                if not isinstance(lang, int):
                    raise ValueError("Language codes in __labels__ must be ints")
                if not isinstance(mapping, dict):
                    raise ValueError(f"__labels__[{lang}] must be a dict of member names to strings")
                labels_by_lang.setdefault(lang, {})
                for k, v in mapping.items():
                    # Accept enum member object, its name, or raw int value (from class body reference)
                    if isinstance(k, enum_cls):
                        member_name = k.name
                    elif isinstance(k, int):
                        member_name = value_to_name.get(k)
                        if member_name is None:
                            raise ValueError(f"__labels__[{lang}] has int key {k} not matching any enum value")
                    else:
                        member_name = str(k)
                    if not isinstance(v, str) or not v.strip():
                        raise ValueError(f"Label for {member_name} lang {lang} must be non-empty string")
                    labels_by_lang[lang][member_name] = v

        config_lang = int(self.config.language_code)
        # Ensure config language appears (fallback to names)
        all_langs = set(labels_by_lang.keys()) | {config_lang}

        options: List[Dict[str, Any]] = []
        for m in sorted(members, key=lambda x: x.value):
            per_lang: Dict[int, str] = {}
            for lang in all_langs:
                label_text = labels_by_lang.get(lang, {}).get(m.name, m.name)
                per_lang[lang] = label_text
            options.append(
                {
                    "@odata.type": "Microsoft.Dynamics.CRM.OptionMetadata",
                    "Value": m.value,
                    "Label": self._build_localizedlabels_payload(per_lang),
                }
            )

        attr_label = column_schema_name.split("_")[-1]
        return {
            "@odata.type": "Microsoft.Dynamics.CRM.PicklistAttributeMetadata",
            "SchemaName": column_schema_name,
            "DisplayName": self._label(attr_label),
            "RequiredLevel": {"Value": "None"},
            "IsPrimaryName": bool(is_primary_name),
            "OptionSet": {
                "@odata.type": "Microsoft.Dynamics.CRM.OptionSetMetadata",
                "IsGlobal": False,
                "Options": options,
            },
        }

    def _normalize_picklist_label(self, label: str) -> str:
        """Normalize a label for case / diacritic insensitive comparison."""
        if not isinstance(label, str):
            return ""
        # Strip accents
        norm = unicodedata.normalize("NFD", label)
        norm = "".join(c for c in norm if unicodedata.category(c) != "Mn")
        # Collapse whitespace, lowercase
        norm = re.sub(r"\s+", " ", norm).strip().lower()
        return norm

    def _request_metadata_with_retry(self, method: str, url: str, **kwargs):
        """Fetch metadata with retries on transient errors."""
        max_attempts = 5
        backoff_seconds = 0.4
        for attempt in range(1, max_attempts + 1):
            try:
                return self._request(method, url, **kwargs)
            except HttpError as err:
                if getattr(err, "status_code", None) == 404:
                    if attempt < max_attempts:
                        time.sleep(backoff_seconds * (2 ** (attempt - 1)))
                        continue
                    raise RuntimeError(f"Metadata request failed after {max_attempts} retries (404): {url}") from err
                raise

    def _bulk_fetch_picklists(self, table_schema_name: str) -> None:
        """Fetch all picklist attributes and their options for a table in one API call.

        Uses collection-level PicklistAttributeMetadata cast to retrieve every picklist
        attribute on the table, including its OptionSet options. Populates the nested
        cache so that ``_convert_labels_to_ints`` resolves labels without further API calls.
        The Dataverse metadata API does not page results.
        """
        table_key = self._normalize_cache_key(table_schema_name)
        now = time.time()
        table_entry = self._picklist_label_cache.get(table_key)
        if isinstance(table_entry, dict) and (now - table_entry.get("ts", 0)) < self._picklist_cache_ttl_seconds:
            return

        table_esc = self._escape_odata_quotes(table_schema_name.lower())
        url = (
            f"{self.api}/EntityDefinitions(LogicalName='{table_esc}')"
            f"/Attributes/Microsoft.Dynamics.CRM.PicklistAttributeMetadata"
            f"?$select=LogicalName&$expand=OptionSet($select=Options)"
        )
        response = self._request_metadata_with_retry("get", url)
        body = response.json()
        items = body.get("value", []) if isinstance(body, dict) else []

        picklists: Dict[str, Dict[str, int]] = {}
        for item in items:
            if not isinstance(item, dict):
                continue
            ln = item.get("LogicalName", "").lower()
            if not ln:
                continue
            option_set = item.get("OptionSet") or {}
            options = option_set.get("Options") if isinstance(option_set, dict) else None
            mapping: Dict[str, int] = {}
            if isinstance(options, list):
                for opt in options:
                    if not isinstance(opt, dict):
                        continue
                    val = opt.get("Value")
                    if not isinstance(val, int):
                        continue
                    label_def = opt.get("Label") or {}
                    locs = label_def.get("LocalizedLabels")
                    if isinstance(locs, list):
                        for loc in locs:
                            if isinstance(loc, dict):
                                lab = loc.get("Label")
                                if isinstance(lab, str) and lab.strip():
                                    normalized = self._normalize_picklist_label(lab)
                                    mapping.setdefault(normalized, val)
            picklists[ln] = mapping

        self._picklist_label_cache[table_key] = {"ts": now, "picklists": picklists}

    def _convert_labels_to_ints(self, table_schema_name: str, record: Dict[str, Any]) -> Dict[str, Any]:
        """Return a copy of record with any labels converted to option ints.

        Heuristic: For each string value, attempt to resolve against picklist metadata.
        If attribute isn't a picklist or label not found, value left unchanged.

        On first encounter of a table, bulk-fetches all picklist attributes and
        their options in a single API call, then resolves labels from the warm cache.
        """
        resolved_record = record.copy()

        # Check if there are any string-valued candidates worth resolving
        has_candidates = any(
            isinstance(v, str) and v.strip() and isinstance(k, str) and "@odata." not in k
            for k, v in resolved_record.items()
        )
        if not has_candidates:
            return resolved_record

        # Bulk-fetch all picklists for this table (1 API call, cached for TTL)
        self._bulk_fetch_picklists(table_schema_name)

        # Resolve labels from the nested cache
        table_key = self._normalize_cache_key(table_schema_name)
        table_entry = self._picklist_label_cache.get(table_key)
        if not isinstance(table_entry, dict):
            return resolved_record
        picklists = table_entry.get("picklists", {})

        for k, v in resolved_record.items():
            if not isinstance(v, str) or not v.strip():
                continue
            if isinstance(k, str) and "@odata." in k:
                continue
            attr_key = self._normalize_cache_key(k)
            mapping = picklists.get(attr_key)
            if not isinstance(mapping, dict) or not mapping:
                continue
            norm = self._normalize_picklist_label(v)
            val = mapping.get(norm)
            if val is not None:
                resolved_record[k] = val
        return resolved_record

    def _attribute_payload(
        self, column_schema_name: str, dtype: Any, *, is_primary_name: bool = False
    ) -> Optional[Dict[str, Any]]:
        # Enum-based local option set support
        if isinstance(dtype, type) and issubclass(dtype, Enum):
            return self._enum_optionset_payload(column_schema_name, dtype, is_primary_name=is_primary_name)
        if not isinstance(dtype, str):
            raise ValueError(
                f"Unsupported column spec type for '{column_schema_name}': {type(dtype)} (expected str or Enum subclass)"
            )
        dtype_l = dtype.lower().strip()
        label = column_schema_name.split("_")[-1]
        if dtype_l in ("string", "text"):
            return {
                "@odata.type": "Microsoft.Dynamics.CRM.StringAttributeMetadata",
                "SchemaName": column_schema_name,
                "DisplayName": self._label(label),
                "RequiredLevel": {"Value": "None"},
                "MaxLength": 200,
                "FormatName": {"Value": "Text"},
                "IsPrimaryName": bool(is_primary_name),
            }
        if dtype_l in ("memo", "multiline"):
            return {
                "@odata.type": "Microsoft.Dynamics.CRM.MemoAttributeMetadata",
                "SchemaName": column_schema_name,
                "DisplayName": self._label(label),
                "RequiredLevel": {"Value": "None"},
                "MaxLength": 4000,
                "FormatName": {"Value": "Text"},
                "ImeMode": "Auto",
            }
        if dtype_l in ("int", "integer"):
            return {
                "@odata.type": "Microsoft.Dynamics.CRM.IntegerAttributeMetadata",
                "SchemaName": column_schema_name,
                "DisplayName": self._label(label),
                "RequiredLevel": {"Value": "None"},
                "Format": "None",
                "MinValue": -2147483648,
                "MaxValue": 2147483647,
            }
        if dtype_l in ("decimal", "money"):
            return {
                "@odata.type": "Microsoft.Dynamics.CRM.DecimalAttributeMetadata",
                "SchemaName": column_schema_name,
                "DisplayName": self._label(label),
                "RequiredLevel": {"Value": "None"},
                "MinValue": -100000000000.0,
                "MaxValue": 100000000000.0,
                "Precision": 2,
            }
        if dtype_l in ("float", "double"):
            return {
                "@odata.type": "Microsoft.Dynamics.CRM.DoubleAttributeMetadata",
                "SchemaName": column_schema_name,
                "DisplayName": self._label(label),
                "RequiredLevel": {"Value": "None"},
                "MinValue": -100000000000.0,
                "MaxValue": 100000000000.0,
                "Precision": 2,
            }
        if dtype_l in ("datetime", "date"):
            return {
                "@odata.type": "Microsoft.Dynamics.CRM.DateTimeAttributeMetadata",
                "SchemaName": column_schema_name,
                "DisplayName": self._label(label),
                "RequiredLevel": {"Value": "None"},
                "Format": "DateOnly",
                "ImeMode": "Inactive",
            }
        if dtype_l in ("bool", "boolean"):
            return {
                "@odata.type": "Microsoft.Dynamics.CRM.BooleanAttributeMetadata",
                "SchemaName": column_schema_name,
                "DisplayName": self._label(label),
                "RequiredLevel": {"Value": "None"},
                "OptionSet": {
                    "@odata.type": "Microsoft.Dynamics.CRM.BooleanOptionSetMetadata",
                    "TrueOption": {
                        "Value": 1,
                        "Label": self._label("True"),
                    },
                    "FalseOption": {
                        "Value": 0,
                        "Label": self._label("False"),
                    },
                    "IsGlobal": False,
                },
            }
        if dtype_l == "file":
            return {
                "@odata.type": "Microsoft.Dynamics.CRM.FileAttributeMetadata",
                "SchemaName": column_schema_name,
                "DisplayName": self._label(label),
                "RequiredLevel": {"Value": "None"},
            }
        return None

    def _get_table_info(self, table_schema_name: str) -> Optional[Dict[str, Any]]:
        """Return basic metadata for a custom table if it exists.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``

        :return: Metadata summary or ``None`` if not found.
        :rtype: ``dict[str, Any]`` | ``None``
        """
        ent = self._get_entity_by_table_schema_name(table_schema_name)
        if not ent:
            return None
        return {
            "table_schema_name": ent.get("SchemaName") or table_schema_name,
            "table_logical_name": ent.get("LogicalName"),
            "entity_set_name": ent.get("EntitySetName"),
            "metadata_id": ent.get("MetadataId"),
            "primary_name_attribute": ent.get("PrimaryNameAttribute"),
            "primary_id_attribute": ent.get("PrimaryIdAttribute"),
            "columns_created": [],
        }

    def _list_tables(
        self,
        filter: Optional[str] = None,
        select: Optional[List[str]] = None,
    ) -> List[Dict[str, Any]]:
        """List all non-private tables (``IsPrivate eq false``).

        :param filter: Optional additional OData ``$filter`` expression that is
            combined with the default ``IsPrivate eq false`` clause using
            ``and``.  For example, ``"SchemaName eq 'Account'"`` becomes
            ``"IsPrivate eq false and (SchemaName eq 'Account')"``.
            When ``None`` (the default), only the ``IsPrivate eq false`` filter
            is applied.
        :type filter: ``str`` or ``None``
        :param select: Optional list of property names to project via
            ``$select``.  Values are passed as-is (PascalCase) because
            ``EntityDefinitions`` uses PascalCase property names.
            When ``None`` (the default) or an empty list, no ``$select`` is
            applied and all properties are returned.  Passing a bare string
            raises ``TypeError``.
        :type select: ``list[str]`` or ``None``

        :return: Metadata entries for non-private tables (may be empty).
        :rtype: ``list[dict[str, Any]]``

        :raises HttpError: If the metadata request fails.
        """
        return self._execute_raw(self._build_list_entities(filter=filter, select=select)).json().get("value", [])

    def _delete_table(self, table_schema_name: str) -> None:
        """Delete a table by schema name.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``

        :return: ``None``
        :rtype: ``None``

        :raises MetadataError: If the table does not exist.
        :raises HttpError: If the delete request fails.
        """
        ent = self._get_entity_by_table_schema_name(table_schema_name)
        if not ent or not ent.get("MetadataId"):
            raise MetadataError(
                f"Table '{table_schema_name}' not found.",
                subcode=METADATA_TABLE_NOT_FOUND,
            )
        self._execute_raw(self._build_delete_entity(ent["MetadataId"]))

    # ------------------- Alternate key metadata helpers -------------------

    def _create_alternate_key(
        self,
        table_schema_name: str,
        key_name: str,
        columns: List[str],
        display_name_label=None,
    ) -> Dict[str, Any]:
        """Create an alternate key on a table.

        Issues ``POST EntityDefinitions(LogicalName='{logical_name}')/Keys``
        with ``EntityKeyMetadata`` payload.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param key_name: Schema name for the new alternate key.
        :type key_name: ``str``
        :param columns: List of column logical names that compose the key.
        :type columns: ``list[str]``
        :param display_name_label: Label for the key display name.
        :type display_name_label: ``Label`` or ``None``

        :return: Dictionary with ``metadata_id``, ``schema_name``, and ``key_attributes``.
        :rtype: ``dict[str, Any]``

        :raises MetadataError: If the table does not exist.
        :raises HttpError: If the Web API request fails.
        """
        ent = self._get_entity_by_table_schema_name(table_schema_name)
        if not ent or not ent.get("MetadataId"):
            raise MetadataError(
                f"Table '{table_schema_name}' not found.",
                subcode=METADATA_TABLE_NOT_FOUND,
            )

        logical_name = ent.get("LogicalName", table_schema_name.lower())
        url = f"{self.api}/EntityDefinitions(LogicalName='{logical_name}')/Keys"
        payload: Dict[str, Any] = {
            "SchemaName": key_name,
            "KeyAttributes": columns,
        }
        if display_name_label is not None:
            payload["DisplayName"] = display_name_label.to_dict()
        r = self._request("post", url, json=payload)
        metadata_id = self._extract_id_from_header(r.headers.get("OData-EntityId"))

        return {
            "metadata_id": metadata_id,
            "schema_name": key_name,
            "key_attributes": columns,
        }

    def _get_alternate_keys(self, table_schema_name: str) -> List[Dict[str, Any]]:
        """List all alternate keys on a table.

        Issues ``GET EntityDefinitions(LogicalName='{logical_name}')/Keys``.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``

        :return: List of raw ``EntityKeyMetadata`` dictionaries.
        :rtype: ``list[dict[str, Any]]``

        :raises MetadataError: If the table does not exist.
        :raises HttpError: If the Web API request fails.
        """
        ent = self._get_entity_by_table_schema_name(table_schema_name)
        if not ent or not ent.get("MetadataId"):
            raise MetadataError(
                f"Table '{table_schema_name}' not found.",
                subcode=METADATA_TABLE_NOT_FOUND,
            )

        logical_name = ent.get("LogicalName", table_schema_name.lower())
        url = f"{self.api}/EntityDefinitions(LogicalName='{logical_name}')/Keys"
        r = self._request("get", url)
        return r.json().get("value", [])

    def _delete_alternate_key(self, table_schema_name: str, key_id: str) -> None:
        """Delete an alternate key by metadata ID.

        Issues ``DELETE EntityDefinitions(LogicalName='{logical_name}')/Keys({key_id})``.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param key_id: Metadata GUID of the alternate key.
        :type key_id: ``str``

        :return: ``None``
        :rtype: ``None``

        :raises MetadataError: If the table does not exist.
        :raises HttpError: If the Web API request fails.
        """
        ent = self._get_entity_by_table_schema_name(table_schema_name)
        if not ent or not ent.get("MetadataId"):
            raise MetadataError(
                f"Table '{table_schema_name}' not found.",
                subcode=METADATA_TABLE_NOT_FOUND,
            )

        logical_name = ent.get("LogicalName", table_schema_name.lower())
        url = f"{self.api}/EntityDefinitions(LogicalName='{logical_name}')/Keys({key_id})"
        self._request("delete", url)

    def _create_table(
        self,
        table_schema_name: str,
        schema: Dict[str, Any],
        solution_unique_name: Optional[str] = None,
        primary_column_schema_name: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Create a custom table with specified columns.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param schema: Mapping of column name -> type spec (``str`` or ``Enum`` subclass).
        :type schema: ``dict[str, Any]``
        :param solution_unique_name: Optional solution container for the new table; if provided must be non-empty.
        :type solution_unique_name: ``str`` | ``None``
        :param primary_column_schema_name: Optional primary column schema name.
        :type primary_column_schema_name: ``str`` | ``None``

        :return: Metadata summary for the created table including created column schema names.
        :rtype: ``dict[str, Any]``

        :raises MetadataError: If the table already exists.
        :raises ValueError: If a column type is unsupported or ``solution_unique_name`` is empty.
        :raises TypeError: If ``solution_unique_name`` is not a ``str`` when provided.
        :raises HttpError: If underlying HTTP requests fail.
        """
        # Check if table already exists (case-insensitive)
        ent = self._get_entity_by_table_schema_name(table_schema_name)
        if ent:
            raise MetadataError(
                f"Table '{table_schema_name}' already exists.",
                subcode=METADATA_TABLE_ALREADY_EXISTS,
            )

        created_cols: List[str] = []

        # Use provided primary column name, or derive from table_schema_name prefix (e.g., "new_Product" -> "new_Name").
        # If no prefix detected, default to "new_Name"; server will validate overall table schema.
        if primary_column_schema_name:
            primary_attr_schema = primary_column_schema_name
        else:
            primary_attr_schema = (
                f"{table_schema_name.split('_',1)[0]}_Name" if "_" in table_schema_name else "new_Name"
            )

        attributes: List[Dict[str, Any]] = []
        attributes.append(self._attribute_payload(primary_attr_schema, "string", is_primary_name=True))
        for col_name, dtype in schema.items():
            payload = self._attribute_payload(col_name, dtype)
            if not payload:
                raise ValueError(f"Unsupported column type '{dtype}' for '{col_name}'.")
            attributes.append(payload)
            created_cols.append(col_name)

        if solution_unique_name is not None:
            if not isinstance(solution_unique_name, str):
                raise TypeError("solution_unique_name must be a string when provided")
            if not solution_unique_name:
                raise ValueError("solution_unique_name cannot be empty")

        metadata = self._create_entity(
            table_schema_name=table_schema_name,
            display_name=table_schema_name,
            attributes=attributes,
            solution_unique_name=solution_unique_name,
        )

        return {
            "table_schema_name": table_schema_name,
            "table_logical_name": metadata.get("LogicalName"),
            "entity_set_name": metadata.get("EntitySetName"),
            "metadata_id": metadata.get("MetadataId"),
            "primary_name_attribute": metadata.get("PrimaryNameAttribute"),
            "primary_id_attribute": metadata.get("PrimaryIdAttribute"),
            "columns_created": created_cols,
        }

    def _create_columns(
        self,
        table_schema_name: str,
        columns: Dict[str, Any],
    ) -> List[str]:
        """Create new columns on an existing table.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param columns: Mapping of column schema name -> type spec (``str`` or ``Enum`` subclass).
        :type columns: ``dict[str, Any]``

        :return: List of created column schema names.
        :rtype: ``list[str]``

        :raises TypeError: If ``columns`` is not a non-empty dict.
        :raises MetadataError: If the target table does not exist.
        :raises ValueError: If a column type is unsupported.
        :raises HttpError: If an underlying HTTP request fails.
        """
        if not isinstance(columns, dict) or not columns:
            raise TypeError("columns must be a non-empty dict[name -> type]")

        ent = self._get_entity_by_table_schema_name(table_schema_name)
        if not ent or not ent.get("MetadataId"):
            raise MetadataError(
                f"Table '{table_schema_name}' not found.",
                subcode=METADATA_TABLE_NOT_FOUND,
            )

        metadata_id = ent.get("MetadataId")
        created: List[str] = []
        needs_picklist_flush = False

        for column_name, column_type in columns.items():
            attr = self._attribute_payload(column_name, column_type)
            if not attr:
                raise ValidationError(
                    f"Unsupported column type '{column_type}' for column '{column_name}'.",
                    subcode=VALIDATION_UNSUPPORTED_COLUMN_TYPE,
                )
            if "OptionSet" in attr:
                needs_picklist_flush = True
            req = _RawRequest(
                method="POST",
                url=f"{self.api}/EntityDefinitions({metadata_id})/Attributes",
                body=json.dumps(attr, ensure_ascii=False),
            )
            self._execute_raw(req)
            created.append(column_name)

        if needs_picklist_flush:
            self._flush_cache("picklist")

        return created

    def _delete_columns(
        self,
        table_schema_name: str,
        columns: Union[str, List[str]],
    ) -> List[str]:
        """Delete one or more columns from a table.

        :param table_schema_name: Schema name of the table.
        :type table_schema_name: ``str``
        :param columns: Single column name or list of column names
        :type columns: ``str`` | ``list[str]``

        :return: List of deleted column schema names (empty if none removed).
        :rtype: ``list[str]``

        :raises TypeError: If ``columns`` is neither a ``str`` nor ``list[str]``.
        :raises ValueError: If any provided column name is empty.
        :raises MetadataError: If the table or a specified column does not exist.
        :raises RuntimeError: If column metadata lacks a required ``MetadataId``.
        :raises HttpError: If an underlying delete request fails.
        """
        if isinstance(columns, str):
            names = [columns]
        elif isinstance(columns, list):
            names = columns
        else:
            raise TypeError("columns must be str or list[str]")

        for name in names:
            if not isinstance(name, str) or not name.strip():
                raise ValueError("column names must be non-empty strings")

        ent = self._get_entity_by_table_schema_name(table_schema_name)
        if not ent or not ent.get("MetadataId"):
            raise MetadataError(
                f"Table '{table_schema_name}' not found.",
                subcode=METADATA_TABLE_NOT_FOUND,
            )

        # Use the actual SchemaName from the entity metadata
        entity_schema = ent.get("SchemaName") or table_schema_name
        metadata_id = ent.get("MetadataId")
        deleted: List[str] = []
        needs_picklist_flush = False

        for column_name in names:
            attr_meta = self._get_attribute_metadata(metadata_id, column_name, extra_select="@odata.type,AttributeType")
            if not attr_meta:
                raise MetadataError(
                    f"Column '{column_name}' not found on table '{entity_schema}'.",
                    subcode=METADATA_COLUMN_NOT_FOUND,
                )

            attr_metadata_id = attr_meta.get("MetadataId")
            if not attr_metadata_id:
                raise RuntimeError(f"Metadata incomplete for column '{column_name}' (missing MetadataId).")

            self._execute_raw(self._build_delete_column(metadata_id, attr_metadata_id))

            attr_type = attr_meta.get("@odata.type") or attr_meta.get("AttributeType")
            if isinstance(attr_type, str):
                attr_type_l = attr_type.lower()
                if "picklist" in attr_type_l or "optionset" in attr_type_l:
                    needs_picklist_flush = True

            deleted.append(column_name)

        if needs_picklist_flush:
            self._flush_cache("picklist")

        return deleted

    # ---------------------- _build_* methods (no HTTP) ---------------

    def _build_create(
        self,
        entity_set: str,
        table: str,
        data: Dict[str, Any],
        *,
        content_id: Optional[int] = None,
    ) -> _RawRequest:
        """Build a single-record POST request without sending it."""
        body = self._lowercase_keys(data)
        body = self._convert_labels_to_ints(table, body)
        return _RawRequest(
            method="POST",
            url=f"{self.api}/{entity_set}",
            body=json.dumps(body, ensure_ascii=False),
            content_id=content_id,
        )

    def _build_create_multiple(
        self,
        entity_set: str,
        table: str,
        records: List[Dict[str, Any]],
    ) -> _RawRequest:
        """Build a CreateMultiple POST request without sending it."""
        if not all(isinstance(r, dict) for r in records):
            raise TypeError("All items for multi-create must be dicts")
        logical_name = table.lower()
        enriched = []
        for r in records:
            r = self._lowercase_keys(r)
            r = self._convert_labels_to_ints(table, r)
            if "@odata.type" not in r:
                r = {**r, "@odata.type": f"Microsoft.Dynamics.CRM.{logical_name}"}
            enriched.append(r)
        return _RawRequest(
            method="POST",
            url=f"{self.api}/{entity_set}/Microsoft.Dynamics.CRM.CreateMultiple",
            body=json.dumps({"Targets": enriched}, ensure_ascii=False),
        )

    def _build_update(
        self,
        table: str,
        record_id: str,
        changes: Dict[str, Any],
        *,
        content_id: Optional[int] = None,
    ) -> _RawRequest:
        """Build a single-record PATCH request without sending it.

        ``record_id`` may be a ``"$n"`` content-ID reference; in that case the
        URL is the reference itself (resolved server-side within a changeset).
        """
        body = self._lowercase_keys(changes)
        body = self._convert_labels_to_ints(table, body)
        if record_id.startswith("$"):
            url = record_id
        else:
            entity_set = self._entity_set_from_schema_name(table)
            url = f"{self.api}/{entity_set}{self._format_key(record_id)}"
        return _RawRequest(
            method="PATCH",
            url=url,
            body=json.dumps(body, ensure_ascii=False),
            headers={"If-Match": "*"},
            content_id=content_id,
        )

    def _build_update_multiple_from_records(
        self,
        entity_set: str,
        table: str,
        records: List[Dict[str, Any]],
    ) -> _RawRequest:
        """Build an UpdateMultiple POST request from pre-assembled records.

        Each record must already contain the primary key attribute. This helper
        is shared by :meth:`_update_multiple` (which pre-assembles records) and
        :meth:`_build_update_multiple` (which assembles from ids + changes).
        """
        logical_name = table.lower()
        enriched = []
        for r in records:
            r = self._lowercase_keys(r)
            r = self._convert_labels_to_ints(table, r)
            if "@odata.type" not in r:
                r = {**r, "@odata.type": f"Microsoft.Dynamics.CRM.{logical_name}"}
            enriched.append(r)
        return _RawRequest(
            method="POST",
            url=f"{self.api}/{entity_set}/Microsoft.Dynamics.CRM.UpdateMultiple",
            body=json.dumps({"Targets": enriched}, ensure_ascii=False),
        )

    def _build_update_multiple(
        self,
        entity_set: str,
        table: str,
        ids: List[str],
        changes: Union[Dict[str, Any], List[Dict[str, Any]]],
    ) -> _RawRequest:
        """Build an UpdateMultiple POST request without sending it."""
        pk_attr = self._primary_id_attr(table)
        if isinstance(changes, dict):
            records = [{pk_attr: rid, **changes} for rid in ids]
        elif isinstance(changes, list):
            if len(changes) != len(ids):
                raise ValidationError(
                    "ids and changes lists must have equal length for paired update.",
                    subcode="ids_changes_length_mismatch",
                )
            records = [{pk_attr: rid, **ch} for rid, ch in zip(ids, changes)]
        else:
            raise ValidationError("changes must be a dict or list[dict].", subcode="invalid_changes_type")
        return self._build_update_multiple_from_records(entity_set, table, records)

    def _build_upsert(
        self,
        entity_set: str,
        table: str,
        alternate_key: Dict[str, Any],
        record: Dict[str, Any],
    ) -> _RawRequest:
        """Build a single-record PATCH upsert request without sending it.

        Unlike :meth:`_build_update`, no ``If-Match: *`` header is added so the
        server creates the record when it does not yet exist.
        """
        body = self._lowercase_keys(record)
        body = self._convert_labels_to_ints(table, body)
        key_str = self._build_alternate_key_str(alternate_key)
        url = f"{self.api}/{entity_set}({key_str})"
        return _RawRequest(
            method="PATCH",
            url=url,
            body=json.dumps(body, ensure_ascii=False),
        )

    def _build_upsert_multiple(
        self,
        entity_set: str,
        table: str,
        alternate_keys: List[Dict[str, Any]],
        records: List[Dict[str, Any]],
    ) -> _RawRequest:
        """Build an UpsertMultiple POST request without sending it."""
        if len(alternate_keys) != len(records):
            raise ValidationError(
                f"alternate_keys and records must have the same length " f"({len(alternate_keys)} != {len(records)})",
                subcode="upsert_length_mismatch",
            )
        logical_name = table.lower()
        targets: List[Dict[str, Any]] = []
        for alt_key, record in zip(alternate_keys, records):
            alt_key_lower = self._lowercase_keys(alt_key)
            record_processed = self._lowercase_keys(record)
            record_processed = self._convert_labels_to_ints(table, record_processed)
            conflicting = {
                k for k in set(alt_key_lower) & set(record_processed) if alt_key_lower[k] != record_processed[k]
            }
            if conflicting:
                raise ValidationError(
                    f"record payload conflicts with alternate_key on fields: {sorted(conflicting)!r}",
                    subcode="upsert_key_conflict",
                )
            if "@odata.type" not in record_processed:
                record_processed["@odata.type"] = f"Microsoft.Dynamics.CRM.{logical_name}"
            key_str = self._build_alternate_key_str(alt_key)
            record_processed["@odata.id"] = f"{entity_set}({key_str})"
            targets.append(record_processed)
        return _RawRequest(
            method="POST",
            url=f"{self.api}/{entity_set}/Microsoft.Dynamics.CRM.UpsertMultiple",
            body=json.dumps({"Targets": targets}, ensure_ascii=False),
        )

    def _build_delete(
        self,
        table: str,
        record_id: str,
        *,
        content_id: Optional[int] = None,
    ) -> _RawRequest:
        """Build a single-record DELETE request without sending it.

        ``record_id`` may be a ``"$n"`` content-ID reference.
        """
        if record_id.startswith("$"):
            url = record_id
        else:
            entity_set = self._entity_set_from_schema_name(table)
            url = f"{self.api}/{entity_set}{self._format_key(record_id)}"
        return _RawRequest(
            method="DELETE",
            url=url,
            headers={"If-Match": "*"},
            content_id=content_id,
        )

    def _build_delete_multiple(self, table: str, ids: List[str]) -> _RawRequest:
        """Build a BulkDelete POST request without sending it."""
        pk_attr = self._primary_id_attr(table)
        logical_name = table.lower()
        timestamp = datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")
        payload = {
            "JobName": f"Bulk delete {table} records @ {timestamp}",
            "SendEmailNotification": False,
            "ToRecipients": [],
            "CCRecipients": [],
            "RecurrencePattern": "",
            "StartDateTime": timestamp,
            "QuerySet": [
                {
                    "@odata.type": "Microsoft.Dynamics.CRM.QueryExpression",
                    "EntityName": logical_name,
                    "ColumnSet": {
                        "@odata.type": "Microsoft.Dynamics.CRM.ColumnSet",
                        "AllColumns": False,
                        "Columns": [],
                    },
                    "Criteria": {
                        "@odata.type": "Microsoft.Dynamics.CRM.FilterExpression",
                        "FilterOperator": "And",
                        "Conditions": [
                            {
                                "@odata.type": "Microsoft.Dynamics.CRM.ConditionExpression",
                                "AttributeName": pk_attr,
                                "Operator": "In",
                                "Values": [{"Value": rid, "Type": "System.Guid"} for rid in ids],
                            }
                        ],
                    },
                }
            ],
        }
        return _RawRequest(
            method="POST",
            url=f"{self.api}/BulkDelete",
            body=json.dumps(payload, ensure_ascii=False),
        )

    def _build_get(
        self,
        table: str,
        record_id: str,
        *,
        select: Optional[List[str]] = None,
    ) -> _RawRequest:
        """Build a single-record GET request without sending it."""
        entity_set = self._entity_set_from_schema_name(table)
        url = f"{self.api}/{entity_set}{self._format_key(record_id)}"
        if select:
            url += "?$select=" + ",".join(self._lowercase_list(select))
        return _RawRequest(method="GET", url=url)

    def _build_create_entity(
        self,
        table: str,
        columns: Dict[str, Any],
        solution: Optional[str] = None,
        primary_column: Optional[str] = None,
    ) -> _RawRequest:
        """Build an EntityDefinitions POST request without sending it."""
        if primary_column:
            primary_attr = primary_column
        else:
            primary_attr = f"{table.split('_', 1)[0]}_Name" if "_" in table else "new_Name"
        attributes = [self._attribute_payload(primary_attr, "string", is_primary_name=True)]
        for col_name, dtype in columns.items():
            attr = self._attribute_payload(col_name, dtype)
            if not attr:
                raise ValidationError(
                    f"Unsupported column type '{dtype}' for column '{col_name}'.",
                    subcode=VALIDATION_UNSUPPORTED_COLUMN_TYPE,
                )
            attributes.append(attr)
        body = {
            "@odata.type": "Microsoft.Dynamics.CRM.EntityMetadata",
            "SchemaName": table,
            "DisplayName": self._label(table),
            "DisplayCollectionName": self._label(table + "s"),
            "Description": self._label(f"Custom entity for {table}"),
            "OwnershipType": "UserOwned",
            "HasActivities": False,
            "HasNotes": True,
            "IsActivity": False,
            "Attributes": attributes,
        }
        url = f"{self.api}/EntityDefinitions"
        if solution:
            url += f"?SolutionUniqueName={solution}"
        return _RawRequest(
            method="POST",
            url=url,
            body=json.dumps(body, ensure_ascii=False),
        )

    def _build_delete_entity(self, metadata_id: str) -> _RawRequest:
        """Build an EntityDefinitions DELETE request without sending it."""
        return _RawRequest(
            method="DELETE",
            url=f"{self.api}/EntityDefinitions({metadata_id})",
            headers={"If-Match": "*"},
        )

    def _build_get_entity(self, table: str) -> _RawRequest:
        """Build an EntityDefinitions GET request without sending it."""
        logical = self._escape_odata_quotes(table.lower())
        return _RawRequest(
            method="GET",
            url=(
                f"{self.api}/EntityDefinitions"
                f"?$select=MetadataId,LogicalName,SchemaName,EntitySetName,PrimaryNameAttribute,PrimaryIdAttribute"
                f"&$filter=LogicalName eq '{logical}'"
            ),
        )

    def _build_list_entities(
        self,
        *,
        filter: Optional[str] = None,
        select: Optional[List[str]] = None,
    ) -> _RawRequest:
        """Build an EntityDefinitions list GET request without sending it."""
        base_filter = "IsPrivate eq false"
        if filter:
            combined_filter = f"{base_filter} and ({filter})"
        else:
            combined_filter = base_filter
        url = f"{self.api}/EntityDefinitions?$filter={combined_filter}"
        if select is not None and isinstance(select, str):
            raise TypeError("select must be a list of property names, not a bare string")
        if select:
            url += "&$select=" + ",".join(select)
        return _RawRequest(method="GET", url=url)

    def _build_create_column(
        self,
        entity_metadata_id: str,
        col_name: str,
        dtype: Any,
    ) -> _RawRequest:
        """Build an Attributes POST request for one column without sending it."""
        attr = self._attribute_payload(col_name, dtype)
        if not attr:
            raise ValidationError(
                f"Unsupported column type '{dtype}' for column '{col_name}'.",
                subcode=VALIDATION_UNSUPPORTED_COLUMN_TYPE,
            )
        return _RawRequest(
            method="POST",
            url=f"{self.api}/EntityDefinitions({entity_metadata_id})/Attributes",
            body=json.dumps(attr, ensure_ascii=False),
        )

    def _build_delete_column(
        self,
        entity_metadata_id: str,
        col_metadata_id: str,
    ) -> _RawRequest:
        """Build an Attributes DELETE request for one column without sending it."""
        return _RawRequest(
            method="DELETE",
            url=f"{self.api}/EntityDefinitions({entity_metadata_id})/Attributes({col_metadata_id})",
            headers={"If-Match": "*"},
        )

    @staticmethod
    def _build_lookup_field_models(
        referencing_table: str,
        lookup_field_name: str,
        referenced_table: str,
        *,
        display_name: Optional[str] = None,
        description: Optional[str] = None,
        required: bool = False,
        cascade_delete: str = CASCADE_BEHAVIOR_REMOVE_LINK,
        language_code: int = 1033,
    ) -> tuple:
        """Build a (lookup, relationship) pair for a lookup field creation.

        Returns ``(LookupAttributeMetadata, OneToManyRelationshipMetadata)``.
        Used by both the batch resolver and ``TableOperations.create_lookup_field``
        to avoid duplicating the metadata assembly logic.
        """
        lookup = LookupAttributeMetadata(
            schema_name=lookup_field_name,
            display_name=Label(
                localized_labels=[
                    LocalizedLabel(
                        label=display_name or referenced_table,
                        language_code=language_code,
                    )
                ]
            ),
            required_level="ApplicationRequired" if required else "None",
        )
        if description:
            lookup.description = Label(
                localized_labels=[LocalizedLabel(label=description, language_code=language_code)]
            )
        rel_name = f"{referenced_table}_{referencing_table}_{lookup_field_name}"
        relationship = OneToManyRelationshipMetadata(
            schema_name=rel_name,
            referenced_entity=referenced_table,
            referencing_entity=referencing_table,
            referenced_attribute=f"{referenced_table}id",
            cascade_configuration=CascadeConfiguration(delete=cascade_delete),
        )
        return lookup, relationship

    def _build_create_relationship(
        self,
        body: Dict[str, Any],
        *,
        solution: Optional[str] = None,
    ) -> _RawRequest:
        """Build a RelationshipDefinitions POST request without sending it."""
        headers: Dict[str, str] = {}
        if solution:
            headers["MSCRM.SolutionUniqueName"] = solution
        return _RawRequest(
            method="POST",
            url=f"{self.api}/RelationshipDefinitions",
            body=json.dumps(body, ensure_ascii=False),
            headers=headers or None,
        )

    def _build_delete_relationship(self, relationship_id: str) -> _RawRequest:
        """Build a RelationshipDefinitions DELETE request without sending it."""
        return _RawRequest(
            method="DELETE",
            url=f"{self.api}/RelationshipDefinitions({relationship_id})",
            headers={"If-Match": "*"},
        )

    def _build_get_relationship(self, schema_name: str) -> _RawRequest:
        """Build a RelationshipDefinitions GET request without sending it."""
        escaped = self._escape_odata_quotes(schema_name)
        return _RawRequest(
            method="GET",
            url=f"{self.api}/RelationshipDefinitions?$filter=SchemaName eq '{escaped}'",
        )

    def _build_sql(self, sql: str) -> _RawRequest:
        """Build a SQL query GET request without sending it.

        Resolves the entity set from the table name in the SQL statement via
        :meth:`_extract_logical_table`, then embeds the SQL as a URL-encoded
        ``?sql=`` query parameter.

        Uses ``urllib.parse.quote`` (``%20`` for spaces) rather than
        ``urllib.parse.urlencode`` (``+`` for spaces).  Both are accepted by
        Dataverse and ``%20`` is the canonical RFC 3986 encoding for query-
        string values.

        :param sql: SELECT statement (non-empty string; caller is responsible
            for validation).
        """
        logical = self._extract_logical_table(sql)
        entity_set = self._entity_set_from_schema_name(logical)
        return _RawRequest(
            method="GET",
            url=f"{self.api}/{entity_set}?sql={_url_quote(sql, safe='')}",
        )

    # ---------------------- Cache maintenance -------------------------
    def _flush_cache(
        self,
        kind,
    ) -> int:
        """Flush cached client metadata/state.

        :param kind: Cache kind to flush (only ``"picklist"`` supported).
        :type kind: ``str``
        :return: Number of cache entries removed.
        :rtype: ``int``
        :raises ValidationError: If ``kind`` is unsupported.
        """
        k = (kind or "").strip().lower()
        if k != "picklist":
            raise ValidationError(
                f"Unsupported cache kind '{kind}' (only 'picklist' is implemented)",
                subcode=VALIDATION_UNSUPPORTED_CACHE_KIND,
            )

        removed = len(self._picklist_label_cache)
        self._picklist_label_cache.clear()
        return removed