feat: SDK-native column constraints (create, update, typed read) — Layer 1 of cutting raw-urllib usage (#206)

# SDK-native column constraints — Layer 1 of reducing raw-`urllib` usage

## The problem this fixes

In the Dataverse plugin's telemetry, a large share of traffic is **raw
`urllib`** (hand-rolled HTTP) instead of the managed SDK. The single
biggest driver is **schema metadata** on the `dv-metadata` skill: agents
were forced to hand-build calls to `EntityDefinitions/Attributes`
because the SDK **could not express column constraints**. Concretely,
three operations had no SDK path:

1. **Create** a column with constraints (`MaxLength`,
`MinValue`/`MaxValue`, `Format`, `RequiredLevel`, a custom
`DisplayName`).
2. **Update** an existing column's constraints (requires the
counter-intuitive `PUT`-not-`PATCH` plus a derived `@odata.type`
discriminator).
3. **Read** type-specific fields back (needs a second request with an
`@odata.type` URL cast).

So for a routine ask like *"add a Rating 1–5 and a 2000-char Comment"*,
the agent dropped to raw `urllib`, hit the metadata quirks, and looped.
This PR gives all three an SDK path, so agents stay on the managed
client (which carries auth, retries, and telemetry attribution) and
**never hand-roll the metadata Web API**.

> **Scope: this is Layer 1 (SDK / product fixes).** Layer 2 — **fixing
the skills** (updating `dv-metadata` to teach the new SDK spec and stop
describing the raw Web-API workaround) — is a separate follow-up in the
plugin repo and depends on this PR shipping.

## What's in this PR

Four changes, all backward-compatible, sync + async parity.

### 1. Create with constraints (closes #194)

`tables.create` / `add_columns` now accept a **dict** column spec
alongside the existing `str` / `Enum`:

```python
client.tables.create("cfb_CustomerFeedback", {
    "cfb_Name":    "string",                                   # unchanged
    "cfb_Rating":  {"type": "int",  "min_value": 1, "max_value": 5},
    "cfb_Comment": {"type": "memo", "max_length": 2000, "display_name": "Comment"},
})
```

Override keys: `max_length`, `min_value`, `max_value`, `precision`,
`format`, `required`, `display_name`.

### 2. Update an existing column (closes #202)

New `tables.update_column` / `update_columns`. The SDK reads the typed
attribute, applies the overrides, and issues the `PUT` with the right
`@odata.type` internally — the caller never sees a verb or a
discriminator:

```python
client.tables.update_column("cfb_CustomerFeedback", "cfb_Comment",
                            {"max_length": 4000, "display_name": "Customer Comment"})
```

### 3. Typed reads in one request (closes #203)

`typed=True` on `list_columns` / `get_column`, backed by
`RetrieveMetadataChanges` with property projection — type-specific
fields come back in a **single** request, no `@odata.type` URL cast:

```python
col  = client.tables.get_column("account", "emailaddress1", typed=True)   # col["MaxLength"], etc.
cols = client.tables.list_columns("account", typed=True)
```

### 4. Attribution allowlist

Add `dv-overview` + `erp-xpp` to `OperationContext._ALLOWED_SKILLS`
(parity with the plugin's `scripts/auth.py`, which allows all 10).
Previously `get_client('dv-overview'|'erp-xpp')` raised `ValueError` on
the SDK side, and the plugin silently fell back to raw `urllib` for
those two skills.

### Compatibility

`str` / `Enum` column specs and every existing signature are unchanged;
the only new parameter is a keyword-only `typed=False` on
`list_columns`. No CHANGELOG / version edit (per CONTRIBUTING —
maintainers handle those at release).

## Testing

**Unit** — full suite green: **2437 passed** (sync + async). New
coverage: dict-spec overrides, `update_column` (PUT verb + `@odata.type`
+ override application + error paths), `RetrieveMetadataChanges` typed
reads (request shape, projection, criteria), `get_column(typed=True)`,
and allowlist parity.

**Live end-to-end** against a Dataverse org (`*.crm10.dynamics.com`) —
**12/12 checks pass**:

| Check | Result |
|---|---|
| #194 create table w/ bounded int (1–5) + 2000-char renamed memo | PASS
|
| #203 typed read → `MinValue=1`, `MaxValue=5`, `MaxLength=2000` (one
request) | PASS |
| #203 `list_columns(typed=True)` returns typed fields | PASS |
| #202 `update_column` memo `2000 → 4000`, re-read confirms | PASS |
| core CRUD (create / retrieve / update / list / delete) — no regression
| PASS |

For contrast, the **unmodified** SDK raises `ValueError: Unsupported
column spec type ... <class 'dict'>` on the very same create — the exact
gap this PR closes.

Author: arorashivam96

.azdo/ci-pr.yaml

@@ -39,7 +39,14 @@ extends:
             versionSpec: '3.12'
             addToPath: true
           displayName: 'Use Python 3.12'
-        
+
+        # 1ES managed pools have no public PyPI egress; route pip through the
+        # internal Azure Artifacts feed (PyPI upstream), same as the Official pipeline.
+        - task: PipAuthenticate@1
+          inputs:
+            artifactFeeds: 'OneCRM/DPX-Tools-Upstream'
+          displayName: 'Authenticate pip to internal feed'
+
         - script: |
             python -m pip install --upgrade pip
             python -m pip install flake8 black build diff-cover

README.md

@@ -111,7 +111,7 @@ df = results.to_dataframe()          # same rows as a pandas DataFrame
 
 ### Define and evolve schema
 
-`client.tables` creates and inspects tables, columns, relationships, and alternate keys. Column types are simple strings (`"string"`, `"int"`, `"decimal"`, `"money"`, `"datetime"`, `"bool"`, `"memo"`); pass an `IntEnum` subclass to create a choice column. `tables.get` returns `None` when the table does not exist, which makes schema setup idempotent. Every table gets a primary name column automatically — `<prefix>_Name` unless you pass `primary_column` — so do not list it in `columns`.
+`client.tables` creates and inspects tables, columns, relationships, and alternate keys. Column types are simple strings (`"string"`, `"int"`, `"decimal"`, `"money"`, `"datetime"`, `"bool"`, `"memo"`); pass an `IntEnum` subclass to create a choice column, or a dict spec (e.g. `{"type": "memo", "max_length": 2000}`) to set constraints like length, range, format, required level, or display name. `tables.get` returns `None` when the table does not exist, which makes schema setup idempotent. Every table gets a primary name column automatically — `<prefix>_Name` unless you pass `primary_column` — so do not list it in `columns`.
 
 ```python
 # Create a custom table with typed columns
@@ -136,6 +136,30 @@ client.tables.create_lookup_field(
 )
 ```
 
+Columns can carry constraints, be updated in place, and be read back with their
+type-specific metadata in a single request:
+
+```python
+# Create columns with constraints -- pass a dict spec instead of a bare type.
+# Keys: max_length, min_value, max_value, precision, format, required, display_name.
+client.tables.create("new_Feedback", {
+    "new_Rating":  {"type": "int",  "min_value": 1, "max_value": 5},
+    "new_Comment": {"type": "memo", "max_length": 2000, "display_name": "Comment"},
+})
+
+# Update an existing column's constraints (same override keys as create).
+client.tables.update_column("new_Feedback", "new_Comment", {"max_length": 4000})
+client.tables.update_columns("new_Feedback", {
+    "new_Rating":  {"max_value": 10},
+    "new_Comment": {"display_name": "Customer Comment"},
+})
+
+# Read a column (or all columns) with typed fields in one request (typed=True).
+col = client.tables.get_column("new_Feedback", "new_Comment", typed=True)
+print(col["MaxLength"])  # -> 4000
+cols = client.tables.list_columns("new_Feedback", typed=True)
+```
+
 Relationship methods take logical names, which are always the schema name in lowercase. For choice columns, many-to-many relationships, and alternate keys, see [Customize tables and columns](https://learn.microsoft.com/power-apps/developer/data-platform/sdk-python/metadata) and [Manage table relationships](https://learn.microsoft.com/power-apps/developer/data-platform/sdk-python/relationships).
 
 ### Work in bulk

src/PowerPlatform/Dataverse/aio/core/_async_http.py

@@ -115,13 +115,13 @@ async def _request(self, method: str, url: str, **kwargs: Any) -> _AsyncResponse
             raise RuntimeError("No aiohttp.ClientSession set. Set _session before making requests.")
 
         # If no timeout is provided, use the user-specified default timeout if set;
-        # otherwise, apply per-method defaults (120s for POST/PATCH/DELETE, 10s for others).
+        # otherwise, apply per-method defaults (120s for POST/PUT/PATCH/DELETE, 10s for others).
         if "timeout" not in kwargs:
             if self.default_timeout is not None:
                 t = self.default_timeout
             else:
                 m = (method or "").lower()
-                t = _TIMEOUT_WRITE_METHODS if m in ("post", "patch", "delete") else _TIMEOUT_READ_METHODS
+                t = _TIMEOUT_WRITE_METHODS if m in ("post", "put", "patch", "delete") else _TIMEOUT_READ_METHODS
             kwargs["timeout"] = aiohttp.ClientTimeout(total=t)
 
         # Log outbound request once (before retry loop).

src/PowerPlatform/Dataverse/aio/data/_async_odata.py

@@ -44,6 +44,7 @@
     _USER_AGENT,
     _DEFAULT_EXPECTED_STATUSES,
     _RequestContext,
+    _TYPED_COLUMN_PROPERTIES,
 )
 
 
@@ -920,6 +921,7 @@ async def _list_columns(
         *,
         select: Optional[List[str]] = None,
         filter: Optional[str] = None,
+        typed: bool = False,
     ) -> List[Dict[str, Any]]:
         """List all attribute (column) definitions for a table.
 
@@ -942,6 +944,14 @@ async def _list_columns(
         :raises MetadataError: If the table is not found.
         :raises HttpError: If the Web API request fails.
         """
+        if typed:
+            if filter:
+                raise ValueError(
+                    "The 'filter' argument is not supported with typed=True: RetrieveMetadataChanges "
+                    "projects typed fields but does not accept an OData $filter. Use typed=False to "
+                    "filter, or drop filter when typed=True."
+                )
+            return await self._retrieve_metadata_changes(table_schema_name, properties=select)
         ent = await self._get_entity_by_table_schema_name(table_schema_name)
         if not ent or not ent.get("MetadataId"):
             raise MetadataError(
@@ -958,6 +968,138 @@ async def _list_columns(
         r = await self._request("get", url, params=params)
         return (r.json()).get("value", [])
 
+    async def _retrieve_metadata_changes(
+        self,
+        table_schema_name: str,
+        *,
+        attribute_logical_name: Optional[str] = None,
+        properties: Optional[List[str]] = None,
+    ) -> List[Dict[str, Any]]:
+        """Return typed column metadata in one request (RetrieveMetadataChanges, #203)."""
+        props = properties or _TYPED_COLUMN_PROPERTIES
+        attr_query: Dict[str, Any] = {"Properties": {"PropertyNames": props}}
+        if attribute_logical_name:
+            attr_query["Criteria"] = {
+                "Conditions": [
+                    {
+                        "PropertyName": "LogicalName",
+                        "ConditionOperator": "Equals",
+                        "Value": {"Type": "System.String", "Value": attribute_logical_name.lower()},
+                    }
+                ]
+            }
+        query: Dict[str, Any] = {
+            "Properties": {"PropertyNames": ["LogicalName", "Attributes"]},
+            "Criteria": {
+                "Conditions": [
+                    {
+                        "PropertyName": "LogicalName",
+                        "ConditionOperator": "Equals",
+                        "Value": {"Type": "System.String", "Value": table_schema_name.lower()},
+                    }
+                ]
+            },
+            "AttributeQuery": attr_query,
+        }
+        url = f"{self.api}/RetrieveMetadataChanges(Query=@p1)"
+        r = await self._request_metadata_with_retry("get", url, params={"@p1": json.dumps(query, ensure_ascii=False)})
+        entities = (r.json()).get("EntityMetadata", [])
+        if not entities:
+            raise MetadataError(
+                f"Table '{table_schema_name}' not found.",
+                subcode=METADATA_TABLE_NOT_FOUND,
+            )
+        return entities[0].get("Attributes", []) or []
+
+    async def _get_column(
+        self,
+        table_schema_name: str,
+        column_name: str,
+        *,
+        typed: bool = False,
+        select: Optional[List[str]] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """Return one column's metadata (typed via RetrieveMetadataChanges when ``typed``)."""
+        if typed:
+            attrs = await self._retrieve_metadata_changes(
+                table_schema_name, attribute_logical_name=column_name, properties=select
+            )
+            return attrs[0] if attrs else None
+        ent = await 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["MetadataId"]
+        logical = self._escape_odata_quotes(column_name.lower())
+        url = f"{self.api}/EntityDefinitions({metadata_id})/Attributes(LogicalName='{logical}')"
+        params: Dict[str, str] = {}
+        if select:
+            params["$select"] = ",".join(select)
+        try:
+            r = await self._request("get", url, params=params)
+            return r.json()
+        except HttpError as err:
+            if getattr(err, "status_code", None) == 404:
+                return None
+            raise
+
+    async def _update_attribute(
+        self,
+        table_schema_name: str,
+        column_name: str,
+        overrides: Dict[str, Any],
+    ) -> str:
+        """Update constraints on an existing column: retrieve the full attr -> PUT it back (#202).
+
+        Follows the documented column-update contract -- ``PUT`` the *entire*
+        current attribute definition so untouched properties are preserved -- and
+        sends ``MSCRM.MergeLabels: true`` so localized labels in other languages
+        survive.
+        """
+        self._validate_column_overrides(column_name, overrides)
+        ent = await 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["MetadataId"]
+        logical = self._escape_odata_quotes(column_name.lower())
+        attr_url = f"{self.api}/EntityDefinitions({metadata_id})/Attributes(LogicalName='{logical}')"
+        try:
+            resp = await self._request("get", attr_url, params={"$select": "MetadataId,SchemaName"})
+            existing = resp.json()
+        except HttpError as err:
+            if getattr(err, "status_code", None) == 404:
+                raise MetadataError(
+                    f"Column '{column_name}' not found on table '{table_schema_name}'.",
+                    subcode=METADATA_COLUMN_NOT_FOUND,
+                ) from err
+            raise
+        attr_metadata_id = existing.get("MetadataId")
+        odata_type = str(existing.get("@odata.type", "")).lstrip("#")
+        if not attr_metadata_id or not odata_type:
+            raise MetadataError(
+                f"Column '{column_name}' not found on table '{table_schema_name}'.",
+                subcode=METADATA_COLUMN_NOT_FOUND,
+            )
+        full = (await self._request("get", f"{attr_url}/{odata_type}")).json()
+        body: Dict[str, Any] = {k: v for k, v in full.items() if not str(k).startswith("@odata.")}
+        body["@odata.type"] = odata_type
+        body["MetadataId"] = attr_metadata_id
+        body.setdefault("SchemaName", existing.get("SchemaName", column_name))
+        self._apply_column_overrides(body, overrides)
+        req = _RawRequest(
+            method="PUT",
+            url=f"{self.api}/EntityDefinitions({metadata_id})/Attributes({attr_metadata_id})",
+            body=json.dumps(body, ensure_ascii=False),
+            headers={"MSCRM.MergeLabels": "true"},
+        )
+        await self._execute_raw(req)
+        return column_name
+
     async def _wait_for_attribute_visibility(
         self,
         entity_set: str,

src/PowerPlatform/Dataverse/aio/operations/async_tables.py

@@ -87,7 +87,13 @@ async def create(
             (or ``"money"``), ``"float"`` (or ``"double"``), ``"datetime"``
             (or ``"date"``), ``"bool"`` (or ``"boolean"``), ``"file"``, and
             ``Enum`` subclasses
-            (for local option sets).
+            (for local option sets). A value may instead be a dict that pairs
+            a base ``type`` with constraint overrides -- ``max_length``,
+            ``min_value``, ``max_value``, ``precision``, ``format``,
+            ``required`` (``"None"``/``"ApplicationRequired"``/``"Recommended"``),
+            and ``display_name`` -- e.g.
+            ``{"type": "int", "min_value": 1, "max_value": 5}`` or
+            ``{"type": "memo", "max_length": 2000, "display_name": "Comment"}``.
         :type columns: :class:`dict`
         :param solution: Optional solution unique name that should own the new
             table. When omitted the table is created in the default solution.
@@ -705,6 +711,7 @@ async def list_columns(
         *,
         select: Optional[List[str]] = None,
         filter: Optional[str] = None,
+        typed: bool = False,
     ) -> List[Dict[str, Any]]:
         """List all attribute (column) definitions for a table.
 
@@ -714,9 +721,15 @@ async def list_columns(
         :param select: Optional list of property names to project via
             ``$select``.  Values are passed as-is (PascalCase).
         :type select: list[str] or None
-        :param filter: Optional OData ``$filter`` expression.  For example,
-            ``"AttributeType eq 'String'"`` returns only string columns.
+        :param filter: Optional OData ``$filter`` expression (only when
+            ``typed=False``).  For example, ``"AttributeType eq 'String'"``
+            returns only string columns.
         :type filter: :class:`str` or None
+        :param typed: When ``True``, read via ``RetrieveMetadataChanges`` so
+            type-specific fields (``MaxLength``, ``MinValue``/``MaxValue``, ...)
+            come back in one request without the ``@odata.type`` URL cast.
+            ``filter`` is not supported in this mode.
+        :type typed: :class:`bool`
 
         :return: List of raw attribute metadata dictionaries.
         :rtype: list[dict[str, typing.Any]]
@@ -725,6 +738,8 @@ async def list_columns(
             If the table is not found.
         :raises ~PowerPlatform.Dataverse.core.errors.HttpError:
             If the Web API request fails.
+        :raises ValueError:
+            If ``filter`` is combined with ``typed=True``.
 
         Example::
 
@@ -746,7 +761,59 @@ async def list_columns(
             )
         """
         async with self._client._scoped_odata() as od:
-            return await od._list_columns(table, select=select, filter=filter)
+            return await od._list_columns(table, select=select, filter=filter, typed=typed)
+
+    async def get_column(
+        self,
+        table: str,
+        column: str,
+        *,
+        typed: bool = False,
+        select: Optional[List[str]] = None,
+    ) -> Optional[Dict[str, Any]]:
+        """Return a single column's metadata, or ``None`` if it does not exist.
+
+        When ``typed`` is ``True``, reads via ``RetrieveMetadataChanges`` so
+        type-specific fields come back in one request (no ``@odata.type`` cast).
+        """
+        async with self._client._scoped_odata() as od:
+            return await od._get_column(table, column, typed=typed, select=select)
+
+    async def update_column(
+        self,
+        table: str,
+        column: str,
+        spec: Dict[str, Any],
+    ) -> str:
+        """Update constraints on an existing column.
+
+        Accepts the same override keys as the dict column spec in
+        :meth:`create` (``max_length``, ``min_value``, ``max_value``,
+        ``precision``, ``format``, ``required``, ``display_name``). The SDK
+        reads the current typed attribute and issues the update internally;
+        callers never pass an HTTP verb or ``@odata.type``.
+        """
+        async with self._client._scoped_odata() as od:
+            return await od._update_attribute(table, column, spec)
+
+    async def update_columns(
+        self,
+        table: str,
+        columns: Dict[str, Dict[str, Any]],
+    ) -> List[str]:
+        """Update constraints on multiple existing columns."""
+        if not isinstance(columns, dict) or not columns:
+            raise TypeError("columns must be a non-empty dict of {column: overrides}")
+        updated: List[str] = []
+        async with self._client._scoped_odata() as od:
+            # Validate every spec up front so a bad entry can't leave the table
+            # partially modified by earlier successful updates.
+            for col, spec in columns.items():
+                od._validate_column_overrides(col, spec)
+            for col, spec in columns.items():
+                await od._update_attribute(table, col, spec)
+                updated.append(col)
+        return updated
 
     # ------------------------------------------------- list_relationships
 

src/PowerPlatform/Dataverse/core/_http.py

@@ -76,14 +76,14 @@ def _request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
         :raises requests.exceptions.RequestException: If all retry attempts fail.
         """
         # If no timeout is provided, use the user-specified default timeout if set;
-        # otherwise, apply per-method defaults (120s for POST/PATCH/DELETE, 10s for others).
+        # otherwise, apply per-method defaults (120s for POST/PUT/PATCH/DELETE, 10s for others).
         if "timeout" not in kwargs:
             if self.default_timeout is not None:
                 kwargs["timeout"] = self.default_timeout
             else:
                 m = (method or "").lower()
                 kwargs["timeout"] = (
-                    _TIMEOUT_WRITE_METHODS if m in ("post", "patch", "delete") else _TIMEOUT_READ_METHODS
+                    _TIMEOUT_WRITE_METHODS if m in ("post", "put", "patch", "delete") else _TIMEOUT_READ_METHODS
                 )
 
         # Log outbound request once (before retry loop).

src/PowerPlatform/Dataverse/core/config.py

@@ -30,13 +30,15 @@
 _ALLOWED_KEYS = frozenset({"app", "skill", "agent"})
 _ALLOWED_SKILLS = frozenset(
     {
+        "dv-overview",
         "dv-connect",
         "dv-data",
         "dv-query",
         "dv-metadata",
         "dv-solution",
         "dv-admin",
         "dv-security",
+        "erp-xpp",
         "unknown",
     }
 )