From 80020114eb0ea5b4b0612b18e1ac6768225b502c Mon Sep 17 00:00:00 2001 From: Max Wang Date: Mon, 25 Aug 2025 10:00:28 -0700 Subject: [PATCH 1/3] api seems to be working. need to check tests a bit --- examples/quickstart_custom_api.py | 337 ++++++++++++++++++++++++++ src/dataverse_sdk/odata.py | 387 +++++++++++++++++++++++++++++- 2 files changed, 723 insertions(+), 1 deletion(-) create mode 100644 examples/quickstart_custom_api.py diff --git a/examples/quickstart_custom_api.py b/examples/quickstart_custom_api.py new file mode 100644 index 0000000..4128c0f --- /dev/null +++ b/examples/quickstart_custom_api.py @@ -0,0 +1,337 @@ +import sys +from pathlib import Path +import traceback +import time +import requests + +# Add src to PYTHONPATH for local runs; insert at position 0 so local code overrides any installed package +src_path = str(Path(__file__).resolve().parents[1] / "src") +if src_path not in sys.path: + sys.path.insert(0, src_path) + +from dataverse_sdk import DataverseClient # noqa: E402 +from azure.identity import InteractiveBrowserCredential # noqa: E402 + +"""Quickstart: Custom API lifecycle (create -> add params -> invoke -> update -> delete). + +Two operating modes: +1. Plug-in backed (you supply PLUGIN_TYPENAME): request + response property, plug-in sets the output. +2. Plug-in-less (business event style): ONLY a request parameter is created. Invocation will succeed + (HTTP 204 / empty body or {{}}) even though no plug-in logic runs. This matches docs stating a + custom API does not strictly require a plug-in (it can just raise events). In this mode we omit + response properties because without server logic they cannot be populated and may cause confusion. + +Below we auto-detect: if PLUGIN_TYPENAME is blank we skip creating the response property and use a +simple string parameter value. If a plug-in name is provided we also create a response property. +""" + +# ---------------- Configuration ---------------- +base_url = "https://aurorabapenv0f528.crm10.dynamics.com" # <-- change to your environment +CUSTOM_API_UNIQUE_NAME = "new_EchoMessage" # Must be globally unique in the org +REQUEST_PARAM_UNIQUE = "new_EchoMessage_Message" +RESPONSE_PROP_UNIQUE = "new_EchoMessage_Response" +CLEANUP = True # Set True to delete the Custom API at the end +PLUGIN_TYPENAME = "" # e.g. "Contoso.Plugins.EchoMessagePlugin" (leave blank for plug-in-less mode) +INCLUDE_RESPONSE_PROPERTY = bool(PLUGIN_TYPENAME) # Only create a response property when a plug-in can populate it +PUBLISH_STRATEGY = "auto" # auto | skip | force. force = call PublishAllXml, auto = poll metadata first + +# ------------------------------------------------ +client = DataverseClient(base_url=base_url, credential=InteractiveBrowserCredential()) +odata = client._get_odata() # low-level client exposing custom API helpers + +# Small helpers: call logging and step pauses + +def log_call(call: str) -> None: + print({"call": call}) + +def plan(call: str) -> None: + print({"plan": call}) + +# Simple generic backoff (same style as other quickstarts) + +def backoff_retry(op, *, delays=(0, 2, 5), retry_http_statuses=(429, 500, 502, 503, 504)): + last_exc = None + for d in delays: + if d: + time.sleep(d) + try: + return op() + except Exception as ex: # noqa: BLE001 + last_exc = ex + if isinstance(ex, requests.exceptions.HTTPError): + code = getattr(getattr(ex, "response", None), "status_code", None) + if code in retry_http_statuses: + continue + break + if last_exc: + raise last_exc + +# 1) List existing custom APIs with our prefix for context +print("List existing custom APIs (prefix=new_):") +try: + plan("odata.list_custom_apis(filter_expr=uniquename startswith 'new_')") + existing = backoff_retry(lambda: odata.list_custom_apis(filter_expr="startswith(uniquename,'new_')")) + print({"count": len(existing)}) + for item in existing[:5]: # show a few + print(" -", item.get("uniquename"), "isfunction=" + str(item.get("isfunction"))) +except Exception as e: # noqa: BLE001 + print(f"List custom APIs failed: {e}") + +# 2) Create the Custom API if absent +print("Recreate Custom API fresh (delete if exists then create):") +existing_api = odata.get_custom_api(unique_name=CUSTOM_API_UNIQUE_NAME) +if existing_api: + plan("odata.delete_custom_api(existing)") + try: + backoff_retry(lambda: odata.delete_custom_api(unique_name=CUSTOM_API_UNIQUE_NAME)) + print({"deleted_prior": True}) + # Brief pause to allow backend cleanup + time.sleep(2) + except Exception as del_ex: # noqa: BLE001 + print({"delete_prior_error": str(del_ex)}) + +plan("odata.create_custom_api (inline request parameter + optional response property)") +try: + # Parameter type codes (subset): 10=String, 7=Integer, 6=Float. We use String for plug-in-less clarity. + request_parameters = [{ + "uniquename": REQUEST_PARAM_UNIQUE, + "name": "Message", + "displayname": "Message", + "type": 6, # Int32 (common & simple) + "description": "Integer message to echo / raise event with", + "isoptional": False, + }] + response_properties = [] + if INCLUDE_RESPONSE_PROPERTY: + response_properties.append({ + "uniquename": RESPONSE_PROP_UNIQUE, + "name": "ResponseMessage", + "displayname": "ResponseMessage", + "type": 6, # Int32 response + "description": "Echoed integer (set by plug-in)", + }) + + api_meta = backoff_retry(lambda: odata.create_custom_api( + unique_name=CUSTOM_API_UNIQUE_NAME, + name="Echo Message", + description="Echo sample (metadata only) created by SDK quickstart.", + is_function=False, + binding_type="Global", + request_parameters=request_parameters, + response_properties=response_properties, + )) + print({"created": True, "customapiid": api_meta.get("customapiid")}) +except Exception as e: + print("Create Custom API failed:") + traceback.print_exc() + resp = getattr(e, 'response', None) + if resp is not None: + try: + print({"status": resp.status_code, "body": resp.text[:2000]}) + except Exception: + pass + sys.exit(1) +created_this_run = True + +customapiid = api_meta.get("customapiid") if api_meta else None +if not customapiid: + print("Missing customapiid; cannot continue") + sys.exit(1) + +# Publish customizations so the action metadata is available for invocation (required for freshly created APIs) +print("Ensure custom API metadata is available:") + +def _action_in_metadata(action_name: str) -> bool: + try: + # Must include auth headers; previously we overwrote them causing 401 + md_resp = odata._request( + "get", + f"{odata.api}/$metadata", + headers={**odata._headers(), "Accept": "application/xml"}, + ) + if md_resp.status_code == 200: + txt = md_resp.text + return f"Name=\"{action_name}\"" in txt + except Exception: # noqa: BLE001 + return False + return False + +def wait_for_action(action_name: str, timeout_sec: int = 60, interval: float = 2.0) -> bool: + start = time.time() + while time.time() - start < timeout_sec: + if _action_in_metadata(action_name): + return True + time.sleep(interval) + return _action_in_metadata(action_name) + +published = False +if PUBLISH_STRATEGY == "skip": + print({"publish_strategy": "skip"}) +elif PUBLISH_STRATEGY in ("auto", "force"): + if PUBLISH_STRATEGY == "auto": + # First attempt: see if already present (often immediate) + if _action_in_metadata(CUSTOM_API_UNIQUE_NAME): + print({"publish_strategy": "auto", "metadata_present": True}) + published = True + else: + print({"publish_strategy": "auto", "metadata_present": False, "action": "polling"}) + if wait_for_action(CUSTOM_API_UNIQUE_NAME, timeout_sec=20, interval=2): + print({"metadata_present_after_poll": True}) + published = True + # Fallback (auto when still not present, or explicit force): attempt PublishAllXml with timeout + if not published: + try: + plan("POST PublishAllXml (timeout=15s)") + pub_url = f"{odata.api}/PublishAllXml" + # Direct requests call so we can enforce timeout + r_pub = requests.post(pub_url, headers=odata._headers(), json={}, timeout=15) + if r_pub.status_code not in (200, 204): + r_pub.raise_for_status() + print({"published": True, "status": r_pub.status_code}) + # Short propagation wait + poll again + time.sleep(3) + if wait_for_action(CUSTOM_API_UNIQUE_NAME, timeout_sec=25, interval=2): + print({"metadata_present_after_publish": True}) + published = True + else: + print({"metadata_present_after_publish": False, "hint": "Invocation retry logic will attempt anyway."}) + except requests.exceptions.Timeout: + print({"published": False, "error": "PublishAllXml timeout (15s)", "hint": "Proceeding; action may still become available."}) + except Exception as pub_ex: # noqa: BLE001 + print({"published": False, "error": str(pub_ex)}) +else: + print({"publish_strategy": PUBLISH_STRATEGY, "warning": "Unknown strategy value"}) + +# Optional: bind an existing plug-in type so invocation returns something meaningful. +# Provide the fully-qualified class name in PLUGIN_TYPENAME above. The plug-in must +# set OutputParameters["ResponseMessage"] (or whatever your response property name is). +if PLUGIN_TYPENAME: + print("Attempt plug-in bind:") + try: + plan(f"lookup plugintype '{PLUGIN_TYPENAME}' then patch custom api") + # Lookup plugintypeid by typename + url = f"{odata.api}/plugintypes" + params = {"$select": "plugintypeid,typename", "$filter": f"typename eq '{PLUGIN_TYPENAME}'"} + r = odata._request("get", url, headers=odata._headers(), params=params) + r.raise_for_status() + vals = r.json().get("value", []) + if vals: + plugintypeid = vals[0]["plugintypeid"] + log_call("odata.update_custom_api (attach plugintype)") + patched = odata.update_custom_api(unique_name=CUSTOM_API_UNIQUE_NAME, changes={ + "plugintypeid@odata.bind": f"/plugintypes({plugintypeid})" + }) + print({"plugin_attached": True, "plugintypeid": plugintypeid}) + else: + print({"plugin_attached": False, "reason": "Plugin typename not found"}) + except Exception as ex: # noqa: BLE001 + resp = getattr(ex, 'response', None) + body = None + if resp is not None: + try: + body = resp.text[:800] + except Exception: # noqa: BLE001 + body = None + print({"plugin_attach_error": str(ex), "body": body}) + +# 3) (Re)List parameters / response properties for visibility +print("Parameters / Response Properties:") +try: + params = odata.list_custom_api_request_parameters(customapiid) + props = odata.list_custom_api_response_properties(customapiid) + print({"parameters": [p.get("name") for p in params], "responses": [p.get("name") for p in props]}) +except Exception as e: # noqa: BLE001 + print(f"List params/props failed: {e}") + +# 4) Invoke the Custom API (will only succeed if backed by server logic) +print("Invoke Custom API:") +time.sleep(1) +try: + # Fetch $metadata to inspect the expected parameter names (diagnostic aid) + try: + meta_url = f"{odata.api}/$metadata" + md_resp = odata._request( + "get", + meta_url, + headers={**odata._headers(), "Accept": "application/xml"}, + ) + md_resp.raise_for_status() + md_text = md_resp.text + # Extract the Action definition snippet for debugging + snippet = None + idx = md_text.find(f"Name=\"{CUSTOM_API_UNIQUE_NAME}\"") + if idx != -1: + snippet = md_text[max(0, idx-200): idx+400] + if snippet: + print({"metadata_snippet": snippet.replace('\n', ' ')[:400]}) + except Exception as md_ex: # noqa: BLE001 + print({"metadata_fetch_error": str(md_ex)}) + + base_message = 123 # Int matches parameter type 6 + # Prefer the unique name first (proved to work in plug-in-less mode); keep logical name fallback for completeness + candidate_param_names = [REQUEST_PARAM_UNIQUE, "Message"] + last_error = None + for pname in candidate_param_names: + for attempt in range(1,4): # up to 3 attempts each name for propagation + invoke_payload = {pname: base_message} + plan(f"attempt {attempt} param '{pname}' -> odata.call_custom_api('{CUSTOM_API_UNIQUE_NAME}', {invoke_payload})") + def invoke(): + return odata.call_custom_api(CUSTOM_API_UNIQUE_NAME, invoke_payload) + try: + result = invoke() + print({"invoked": True, "result": result, "mode": "plugin" if INCLUDE_RESPONSE_PROPERTY else "plugin-less", "used_param": pname, "attempt": attempt}) + raise SystemExit # exit double loop cleanly + except requests.exceptions.HTTPError as ex: # noqa: PERF203 + last_error = ex + resp = getattr(ex, 'response', None) + body = None + if resp is not None: + try: + body = resp.text[:300] + except Exception: # noqa: BLE001 + body = None + if resp is not None and resp.status_code == 400 and "not a valid parameter" in (body or ""): + # Wait and retry + time.sleep(2 + attempt) + continue + if resp is not None and resp.status_code == 400 and "Int32" in (body or ""): + print({"hint": "Server expects Int32; ensure payload is int (it is). If still failing, metadata not published yet."}) + time.sleep(2) + continue + print({"attempt": pname, "error": str(ex), "body": body}) + time.sleep(2) + continue + if last_error: + raise last_error +except SystemExit: + pass # Successful invocation path signaled via SystemExit raise above +except Exception as e: # Invocation may legitimately fail without a plug-in + resp = getattr(e, 'response', None) + body = None + if resp is not None: + try: + body = resp.text[:1500] + except Exception: + body = None + print({"invoked": False, "error": str(e), "body": body, "hint": "If 400, verify parameter Type code & payload match; for plug-in-less mode only request param should be present."}) + +# 5) Update description (demonstrate patch) +print("Update Custom API description:") +try: + plan("odata.update_custom_api(unique_name, changes={'description': 'Updated via quickstart'})") + updated = backoff_retry(lambda: odata.update_custom_api(unique_name=CUSTOM_API_UNIQUE_NAME, changes={"description": "Updated via quickstart"})) + print({"updated": True, "description": updated.get("description")}) +except Exception as e: # noqa: BLE001 + print({"updated": False, "error": str(e)}) + +# 6) Conditional cleanup +if CLEANUP and created_this_run: + print("Cleanup: delete Custom API created in this run") + try: + plan("odata.delete_custom_api(unique_name)") + backoff_retry(lambda: odata.delete_custom_api(unique_name=CUSTOM_API_UNIQUE_NAME)) + print({"deleted": True}) + except Exception as e: # noqa: BLE001 + print({"deleted": False, "error": str(e)}) +else: + print({"cleanup": False, "reason": "CLEANUP flag False or pre-existing API"}) diff --git a/src/dataverse_sdk/odata.py b/src/dataverse_sdk/odata.py index 5bf39cf..b2b9b46 100644 --- a/src/dataverse_sdk/odata.py +++ b/src/dataverse_sdk/odata.py @@ -1,8 +1,9 @@ from __future__ import annotations -from typing import Any, Dict, Optional, List +from typing import Any, Dict, Optional, List, Iterable import re import json +from urllib.parse import urlencode from .http import HttpClient @@ -301,3 +302,387 @@ def create_table(self, tablename: str, schema: Dict[str, str]) -> Dict[str, Any] "metadata_id": metadata_id, "columns_created": created_cols, } + + # ---------------------- Custom API (metadata) ---------------------- + def list_custom_apis(self, select: Optional[Iterable[str]] = None, filter_expr: Optional[str] = None) -> List[Dict[str, Any]]: + """List Custom APIs (metadata records). + + Parameters + ---------- + select : iterable of str, optional + Specific columns to select (e.g. ["customapiid","uniquename","isfunction"]). + filter_expr : str, optional + OData $filter expression. + """ + params: Dict[str, Any] = {"$select": ",".join(select) if select else "customapiid,uniquename,isfunction,bindingtype"} + if filter_expr: + params["$filter"] = filter_expr + url = f"{self.api}/customapis" + r = self._request("get", url, headers=self._headers(), params=params) + r.raise_for_status() + return r.json().get("value", []) + + def _get_custom_api(self, *, unique_name: Optional[str] = None, customapiid: Optional[str] = None) -> Optional[Dict[str, Any]]: + """Internal helper to fetch a single Custom API by unique name or id.""" + if not unique_name and not customapiid: + raise ValueError("Provide unique_name or customapiid") + if customapiid: + url = f"{self.api}/customapis({customapiid})" + r = self._request("get", url, headers=self._headers()) + if r.status_code == 404: + return None + r.raise_for_status() + return r.json() + # unique name path + flt = f"uniquename eq '{unique_name}'" + items = self.list_custom_apis(filter_expr=flt) + if not items: + return None + return items[0] + + def get_custom_api(self, unique_name: Optional[str] = None, customapiid: Optional[str] = None) -> Optional[Dict[str, Any]]: + """Public accessor for a Custom API metadata record.""" + return self._get_custom_api(unique_name=unique_name, customapiid=customapiid) + + def create_custom_api( + self, + *, + unique_name: str, + name: str, + description: Optional[str] = None, + is_function: bool = False, + binding_type: str | int = "Global", + bound_entity_logical_name: Optional[str] = None, + allowed_custom_processing_step_type: int = 0, + execute_privilege_name: Optional[str] = None, + payload: Optional[Dict[str, Any]] = None, + request_parameters: Optional[List[Dict[str, Any]]] = None, + response_properties: Optional[List[Dict[str, Any]]] = None, + plugin_type_id: Optional[str] = None, + is_private: bool = False, + is_customizable: Optional[bool] = None, + ) -> Dict[str, Any]: + """Create a Dataverse Custom API metadata record. + + Parameters + ---------- + unique_name : str + Unique name (publisher prefix + name), e.g. ``new_Echo``. + name : str + Friendly display/primary name. + description : str, optional + Description text. + is_function : bool, default False + When True creates a function (GET); otherwise an action (POST). + binding_type : str | int, default "Global" + One of ``Global``, ``Entity``, ``EntityCollection`` (or 0/1/2). + bound_entity_logical_name : str, optional + Logical name required when binding_type is Entity or EntityCollection. + allowed_custom_processing_step_type : int, default 0 + Allowed custom processing step type (0 = None, 1 = Plug-in, etc. per platform option set) – typically leave 0. + execute_privilege_name : str, optional + Privilege name required to execute (rare; use to gate execution by security role privilege). + payload : dict, optional + Raw body overrides/extra fields to merge; values here win only if not already set by convenience params. + request_parameters : list[dict], optional + Inline definitions for CustomAPIRequestParameters. + (e.g. uniquename, name, displayname, description, type (option set int), isoptional, logicalentityname, iscustomizable={"Value": bool}). + response_properties : list[dict], optional + Inline definitions for CustomAPIResponseProperties (same shape as request parameters, minus isoptional). + plugin_type_id : str, optional + GUID of an existing plugintype to bind via ``PluginTypeId@odata.bind`` so the API executes that plug-in. + is_private : bool, default False + Marks the Custom API as private (hidden from some discovery scenarios). + is_customizable : bool, optional + When provided wraps into ``{"Value": bool}`` to set the metadata customizability flag. + + Notes + ----- + This does not register any plug-in code; invocation will only succeed if server logic exists. + """ + if self._get_custom_api(unique_name=unique_name): + raise RuntimeError(f"Custom API '{unique_name}' already exists") + body: Dict[str, Any] = payload.copy() if payload else {} + body.setdefault("uniquename", unique_name) + body.setdefault("name", name) + body.setdefault("displayname", name) + if description: + body.setdefault("description", description) + body.setdefault("isfunction", bool(is_function)) + body.setdefault("isprivate", bool(is_private)) + # bindingtype expects an int (0=Global,1=Entity,2=EntityCollection) + if isinstance(binding_type, str): + bt_map = {"global": 0, "entity": 1, "entitycollection": 2} + bt_key = binding_type.lower().strip() + binding_type_value = bt_map.get(bt_key) + if binding_type_value is None: + raise ValueError("binding_type must be one of Global, Entity, EntityCollection or an int 0/1/2") + body.setdefault("bindingtype", binding_type_value) + else: + body.setdefault("bindingtype", binding_type) + if bound_entity_logical_name: + body.setdefault("boundentitylogicalname", bound_entity_logical_name) + body.setdefault("allowedcustomprocessingsteptype", allowed_custom_processing_step_type) + if execute_privilege_name: + body.setdefault("executeprivilegename", execute_privilege_name) + if is_customizable is not None: + body.setdefault("iscustomizable", {"Value": bool(is_customizable)}) + if plugin_type_id: + body.setdefault("PluginTypeId@odata.bind", f"/plugintypes({plugin_type_id})") + + if request_parameters: + body["CustomAPIRequestParameters"] = request_parameters + if response_properties: + body["CustomAPIResponseProperties"] = response_properties + + url = f"{self.api}/customapis" + headers = self._headers().copy() + headers["Prefer"] = "return=representation" + r = self._request("post", url, headers=headers, json=body) + r.raise_for_status() + created: Dict[str, Any] + if r.status_code == 204 or not r.content: + # Representation not returned; do a lookup by unique name + created = self._get_custom_api(unique_name=unique_name) or {"uniquename": unique_name} + else: + try: + created = r.json() + except Exception: # noqa: BLE001 + created = {"uniquename": unique_name} + return created + + def update_custom_api(self, *, unique_name: Optional[str] = None, customapiid: Optional[str] = None, changes: Dict[str, Any]) -> Dict[str, Any]: + """Patch an existing Custom API.""" + rec = self._get_custom_api(unique_name=unique_name, customapiid=customapiid) + if not rec: + raise RuntimeError("Custom API not found") + cid = rec.get("customapiid") + url = f"{self.api}/customapis({cid})" + headers = self._headers().copy() + headers["If-Match"] = "*" + # Prefer return representation + headers["Prefer"] = "return=representation" + r = self._request("patch", url, headers=headers, json=changes) + r.raise_for_status() + return r.json() + + def delete_custom_api(self, *, unique_name: Optional[str] = None, customapiid: Optional[str] = None) -> None: + rec = self._get_custom_api(unique_name=unique_name, customapiid=customapiid) + if not rec: + return + cid = rec.get("customapiid") + url = f"{self.api}/customapis({cid})" + headers = self._headers().copy() + headers["If-Match"] = "*" + r = self._request("delete", url, headers=headers) + if r.status_code not in (200, 204, 404): + r.raise_for_status() + + # ---------------------- Custom API invocation ---------------------- + def call_custom_api(self, name: str, parameters: Optional[Dict[str, Any]] = None, *, is_function: Optional[bool] = None) -> Any: + """Invoke a custom API by its unique name. + + Parameters + ---------- + name : str + Unique name of the custom API. + parameters : dict, optional + Key/value pairs of parameters. For functions these are sent as query string; for actions in the body. + is_function : bool, optional + If not provided, is_function lookup is performed. + """ + params = parameters or {} + # Determine if function or action + fn_flag = is_function + if fn_flag is None: + meta = self._get_custom_api(unique_name=name) + if not meta: + raise RuntimeError(f"Custom API '{name}' not found") + fn_flag = bool(meta.get("isfunction")) + try: + if fn_flag: + # Function -> GET with query string parameters (primitive only) + if params: + def format_val(v: Any) -> str: + if isinstance(v, str): + return f"'{v.replace("'", "''")}'" + if isinstance(v, bool): + return "true" if v else "false" + return str(v) + inner = ",".join(f"{k}={format_val(v)}" for k, v in params.items()) + url = f"{self.api}/{name}({inner})" + else: + url = f"{self.api}/{name}()" + r = self._request("get", url, headers=self._headers()) + else: + # Action -> POST with JSON body (parameters serialized directly) + url = f"{self.api}/{name}" + r = self._request("post", url, headers=self._headers(), json=params if params else None) + r.raise_for_status() + except Exception as ex: + # Try to surface server diagnostic if available + resp = getattr(ex, 'response', None) + if resp is not None: + try: + detail = resp.text[:1000] + raise RuntimeError(f"Custom API call failed ({resp.status_code}) body={detail}") from ex + except Exception: + pass + raise + # Some custom APIs return no content (204) + if r.status_code == 204 or not r.content: + return None + ct = r.headers.get("Content-Type", "") + if "application/json" in ct: + try: + return r.json() + except Exception: # noqa: BLE001 + return r.text + return r.text + + # ----------------- Custom API request parameters ------------------- + _DATA_TYPE_MAP = { + "boolean": 0, + "datetime": 1, + "decimal": 2, + "entity": 3, + "entitycollection": 4, + "float": 5, + "int": 6, + "integer": 6, + "money": 7, + "picklist": 8, + "string": 9, + "stringarray": 10, + "guid": 11, + "entityreference": 12, + "entityreferencecollection": 13, + "bigint": 14, + } + + def _resolve_data_type(self, data_type: Any) -> Any: + if isinstance(data_type, int): + return data_type + if isinstance(data_type, str): + k = data_type.lower().strip() + if k in self._DATA_TYPE_MAP: + return self._DATA_TYPE_MAP[k] + return data_type # let server validate + + def list_custom_api_request_parameters(self, customapiid: str) -> List[Dict[str, Any]]: + params = { + "$select": "customapirequestparameterid,uniquename,name,type,isoptional", + "$filter": f"_customapiid_value eq {customapiid}", + } + url = f"{self.api}/customapirequestparameters" + r = self._request("get", url, headers=self._headers(), params=params) + r.raise_for_status() + return r.json().get("value", []) + + def create_custom_api_request_parameter( + self, + *, + customapiid: Optional[str] = None, + custom_api_unique_name: Optional[str] = None, + unique_name: str, + name: Optional[str] = None, + data_type: Any = "string", + description: Optional[str] = None, + is_optional: bool = False, + logical_entity_name: Optional[str] = None, + extra: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """Create a Custom API Request Parameter. + + If customapiid is not supplied, custom_api_unique_name is resolved first. + data_type can be an int (raw option set) or a friendly string (see _DATA_TYPE_MAP). + """ + if not customapiid: + if not custom_api_unique_name: + raise ValueError("Provide customapiid or custom_api_unique_name") + api_meta = self._get_custom_api(unique_name=custom_api_unique_name) + if not api_meta: + raise RuntimeError(f"Custom API '{custom_api_unique_name}' not found") + customapiid = api_meta.get("customapiid") + body: Dict[str, Any] = (extra or {}).copy() + body.setdefault("uniquename", unique_name) + body.setdefault("name", name or unique_name) + body.setdefault("displayname", name or unique_name) + body.setdefault("isoptional", bool(is_optional)) + body.setdefault("type", self._resolve_data_type(data_type)) + if description: + body.setdefault("description", description) + if logical_entity_name: + body.setdefault("logicalentityname", logical_entity_name) + # Associate to parent Custom API via navigation property + body.setdefault("customapiid@odata.bind", f"/customapis({customapiid})") + url = f"{self.api}/customapirequestparameters" + r = self._request("post", url, headers=self._headers(), json=body) + r.raise_for_status() + return r.json() + + def delete_custom_api_request_parameter(self, request_parameter_id: str) -> None: + url = f"{self.api}/customapirequestparameters({request_parameter_id})" + headers = self._headers().copy() + headers["If-Match"] = "*" + r = self._request("delete", url, headers=headers) + if r.status_code not in (200, 204, 404): + r.raise_for_status() + + # --------------- Custom API response properties -------------------- + def list_custom_api_response_properties(self, customapiid: str) -> List[Dict[str, Any]]: + params = { + "$select": "customapiresponsepropertyid,uniquename,name,type", + "$filter": f"_customapiid_value eq {customapiid}", + } + url = f"{self.api}/customapiresponseproperties" + r = self._request("get", url, headers=self._headers(), params=params) + r.raise_for_status() + return r.json().get("value", []) + + def create_custom_api_response_property( + self, + *, + customapiid: Optional[str] = None, + custom_api_unique_name: Optional[str] = None, + unique_name: str, + name: Optional[str] = None, + data_type: Any = "string", + description: Optional[str] = None, + logical_entity_name: Optional[str] = None, + extra: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """Create a Custom API Response Property. + + If customapiid is not supplied, custom_api_unique_name is resolved first. + data_type can be an int (raw option set) or a friendly string (see _DATA_TYPE_MAP). + """ + if not customapiid: + if not custom_api_unique_name: + raise ValueError("Provide customapiid or custom_api_unique_name") + api_meta = self._get_custom_api(unique_name=custom_api_unique_name) + if not api_meta: + raise RuntimeError(f"Custom API '{custom_api_unique_name}' not found") + customapiid = api_meta.get("customapiid") + body: Dict[str, Any] = (extra or {}).copy() + body.setdefault("uniquename", unique_name) + body.setdefault("name", name or unique_name) + body.setdefault("displayname", name or unique_name) + body.setdefault("type", self._resolve_data_type(data_type)) + if description: + body.setdefault("description", description) + if logical_entity_name: + body.setdefault("logicalentityname", logical_entity_name) + body.setdefault("customapiid@odata.bind", f"/customapis({customapiid})") + url = f"{self.api}/customapiresponseproperties" + r = self._request("post", url, headers=self._headers(), json=body) + r.raise_for_status() + return r.json() + + def delete_custom_api_response_property(self, response_property_id: str) -> None: + url = f"{self.api}/customapiresponseproperties({response_property_id})" + headers = self._headers().copy() + headers["If-Match"] = "*" + r = self._request("delete", url, headers=headers) + if r.status_code not in (200, 204, 404): + r.raise_for_status() From 2ec1de2f260060255756aaa325cb87b5cbfccc07 Mon Sep 17 00:00:00 2001 From: Max Wang Date: Thu, 28 Aug 2025 16:02:00 -0700 Subject: [PATCH 2/3] working version --- README.md | 5 + examples/quickstart_custom_api.py | 235 ++++++++++++------------------ src/dataverse_sdk/odata.py | 127 ++++------------ 3 files changed, 126 insertions(+), 241 deletions(-) diff --git a/README.md b/README.md index 0bd8ada..05e5eba 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,7 @@ A minimal Python SDK to use Microsoft Dataverse as a database for Azure AI Found - OData CRUD — Thin wrappers over Dataverse Web API (create/get/update/delete). - Metadata helpers — Create/inspect/delete simple custom tables (EntityDefinitions + Attributes). - Pandas helpers — Convenience DataFrame oriented wrappers for quick prototyping/notebooks. +- Custom API — CRUD for Custom API that wraps over Dataverse Web API - Auth — Azure Identity (`TokenCredential`) injection. ## Features @@ -13,6 +14,7 @@ A minimal Python SDK to use Microsoft Dataverse as a database for Azure AI Found - Simple `DataverseClient` facade for CRUD, SQL (read-only), and table metadata. - SQL-over-API: T-SQL routed through Custom API endpoint (no ODBC / TDS driver required). - Table metadata ops: create simple custom tables with primitive columns (string/int/decimal/float/datetime/bool) and delete them. +- Custom API support for CRUD and invoking - Optional pandas integration (`PandasODataClient`) for DataFrame based create / get / query. Auth: @@ -139,7 +141,9 @@ Notes: - For CRUD methods that take a record id, pass the GUID string (36-char hyphenated). Parentheses around the GUID are accepted but not required. - SQL is routed through the Custom API named in `DataverseConfig.sql_api_name` (default: `McpExecuteSqlQuery`). +### Custom API functionalities +See `examples/quickstart_custom_api.py` for a Custom API workflow from create -> read- > call -> update -> delete. ### Pandas helpers @@ -152,6 +156,7 @@ VS Code Tasks ## Limitations / Future Work - No batching, upsert, or association operations yet. - Minimal retry policy in library (network-error only); examples include additional backoff for transient Dataverse consistency. +- Custom API SDK doesn't support adding service logic like Plug-in currently, so it can only function like business events. ## Contributing diff --git a/examples/quickstart_custom_api.py b/examples/quickstart_custom_api.py index 4128c0f..dbec695 100644 --- a/examples/quickstart_custom_api.py +++ b/examples/quickstart_custom_api.py @@ -9,31 +9,31 @@ if src_path not in sys.path: sys.path.insert(0, src_path) -from dataverse_sdk import DataverseClient # noqa: E402 -from azure.identity import InteractiveBrowserCredential # noqa: E402 - -"""Quickstart: Custom API lifecycle (create -> add params -> invoke -> update -> delete). - -Two operating modes: -1. Plug-in backed (you supply PLUGIN_TYPENAME): request + response property, plug-in sets the output. -2. Plug-in-less (business event style): ONLY a request parameter is created. Invocation will succeed - (HTTP 204 / empty body or {{}}) even though no plug-in logic runs. This matches docs stating a - custom API does not strictly require a plug-in (it can just raise events). In this mode we omit - response properties because without server logic they cannot be populated and may cause confusion. - -Below we auto-detect: if PLUGIN_TYPENAME is blank we skip creating the response property and use a -simple string parameter value. If a plug-in name is provided we also create a response property. -""" +from dataverse_sdk import DataverseClient +from azure.identity import InteractiveBrowserCredential # ---------------- Configuration ---------------- base_url = "https://aurorabapenv0f528.crm10.dynamics.com" # <-- change to your environment CUSTOM_API_UNIQUE_NAME = "new_EchoMessage" # Must be globally unique in the org REQUEST_PARAM_UNIQUE = "new_EchoMessage_Message" RESPONSE_PROP_UNIQUE = "new_EchoMessage_Response" -CLEANUP = True # Set True to delete the Custom API at the end -PLUGIN_TYPENAME = "" # e.g. "Contoso.Plugins.EchoMessagePlugin" (leave blank for plug-in-less mode) -INCLUDE_RESPONSE_PROPERTY = bool(PLUGIN_TYPENAME) # Only create a response property when a plug-in can populate it PUBLISH_STRATEGY = "auto" # auto | skip | force. force = call PublishAllXml, auto = poll metadata first +# Parameter type codes (subset): 10=String, 7=Int32, 6=Float. Using Int32 for this run. +REQUEST_PARAMETERS = [{ + "uniquename": REQUEST_PARAM_UNIQUE, + "name": "Message", + "displayname": "Message", + "type": 7, # Int32 + "description": "Integer value to echo / raise event with", + "isoptional": False, +}] +RESPONSE_PROPERTIES = [{ + "uniquename": RESPONSE_PROP_UNIQUE, + "name": "ResponseMessage", + "displayname": "ResponseMessage", + "type": 10, # String response + "description": "Echoed string", +}] # ------------------------------------------------ client = DataverseClient(base_url=base_url, credential=InteractiveBrowserCredential()) @@ -56,7 +56,7 @@ def backoff_retry(op, *, delays=(0, 2, 5), retry_http_statuses=(429, 500, 502, 5 time.sleep(d) try: return op() - except Exception as ex: # noqa: BLE001 + except Exception as ex: last_exc = ex if isinstance(ex, requests.exceptions.HTTPError): code = getattr(getattr(ex, "response", None), "status_code", None) @@ -66,18 +66,16 @@ def backoff_retry(op, *, delays=(0, 2, 5), retry_http_statuses=(429, 500, 502, 5 if last_exc: raise last_exc -# 1) List existing custom APIs with our prefix for context -print("List existing custom APIs (prefix=new_):") +# 1) Check if target Custom API exists +print("Check target Custom API existence:") try: - plan("odata.list_custom_apis(filter_expr=uniquename startswith 'new_')") - existing = backoff_retry(lambda: odata.list_custom_apis(filter_expr="startswith(uniquename,'new_')")) - print({"count": len(existing)}) - for item in existing[:5]: # show a few - print(" -", item.get("uniquename"), "isfunction=" + str(item.get("isfunction"))) -except Exception as e: # noqa: BLE001 - print(f"List custom APIs failed: {e}") + plan("odata.get_custom_api(unique_name)") + existing = backoff_retry(lambda: odata.get_custom_api(unique_name=CUSTOM_API_UNIQUE_NAME)) + print({"exists": bool(existing)}) +except Exception as e: + print(f"Existence check failed: {e}") -# 2) Create the Custom API if absent +# 2) Create the Custom API, remove the existing one first if present print("Recreate Custom API fresh (delete if exists then create):") existing_api = odata.get_custom_api(unique_name=CUSTOM_API_UNIQUE_NAME) if existing_api: @@ -87,40 +85,30 @@ def backoff_retry(op, *, delays=(0, 2, 5), retry_http_statuses=(429, 500, 502, 5 print({"deleted_prior": True}) # Brief pause to allow backend cleanup time.sleep(2) - except Exception as del_ex: # noqa: BLE001 + except Exception as del_ex: print({"delete_prior_error": str(del_ex)}) -plan("odata.create_custom_api (inline request parameter + optional response property)") +plan("odata.create_custom_api (inline request parameter + response property)") try: - # Parameter type codes (subset): 10=String, 7=Integer, 6=Float. We use String for plug-in-less clarity. - request_parameters = [{ - "uniquename": REQUEST_PARAM_UNIQUE, - "name": "Message", - "displayname": "Message", - "type": 6, # Int32 (common & simple) - "description": "Integer message to echo / raise event with", - "isoptional": False, - }] - response_properties = [] - if INCLUDE_RESPONSE_PROPERTY: - response_properties.append({ - "uniquename": RESPONSE_PROP_UNIQUE, - "name": "ResponseMessage", - "displayname": "ResponseMessage", - "type": 6, # Int32 response - "description": "Echoed integer (set by plug-in)", - }) - api_meta = backoff_retry(lambda: odata.create_custom_api( unique_name=CUSTOM_API_UNIQUE_NAME, name="Echo Message", description="Echo sample (metadata only) created by SDK quickstart.", is_function=False, binding_type="Global", - request_parameters=request_parameters, - response_properties=response_properties, + request_parameters=REQUEST_PARAMETERS, + response_properties=RESPONSE_PROPERTIES, )) - print({"created": True, "customapiid": api_meta.get("customapiid")}) + print({ + "created": True, + "message": "Created Custom API with the following parameters", + "unique_name": CUSTOM_API_UNIQUE_NAME, + "customapiid": api_meta.get("customapiid"), + "description": "Echo sample (metadata only) created by SDK quickstart.", + "is_function": False, + "request_parameters": [p.get("name") for p in REQUEST_PARAMETERS], + "response_properties": [p.get("name") for p in RESPONSE_PROPERTIES] + }) except Exception as e: print("Create Custom API failed:") traceback.print_exc() @@ -131,19 +119,35 @@ def backoff_retry(op, *, delays=(0, 2, 5), retry_http_statuses=(429, 500, 502, 5 except Exception: pass sys.exit(1) -created_this_run = True customapiid = api_meta.get("customapiid") if api_meta else None if not customapiid: print("Missing customapiid; cannot continue") sys.exit(1) +# 3) Read back the Custom API metadata just created +print("Read Custom API metadata:") +try: + plan("odata.get_custom_api(unique_name)") + read_back = backoff_retry(lambda: odata.get_custom_api(unique_name=CUSTOM_API_UNIQUE_NAME)) + if read_back: + # Display a concise subset of fields + subset = {k: read_back.get(k) for k in [ + "customapiid", "uniquename", "isfunction", "bindingtype", "allowedcustomprocessingsteptype", "isprivate", "executeprivilegename", "description" + ]} + subset["request_param_count"] = len(REQUEST_PARAMETERS) + subset["response_prop_count"] = len(RESPONSE_PROPERTIES) + print({"read_back": subset}) + else: + print({"read_back": None}) +except Exception as e: + print({"read_custom_api_error": str(e)}) + # Publish customizations so the action metadata is available for invocation (required for freshly created APIs) print("Ensure custom API metadata is available:") def _action_in_metadata(action_name: str) -> bool: try: - # Must include auth headers; previously we overwrote them causing 401 md_resp = odata._request( "get", f"{odata.api}/$metadata", @@ -152,7 +156,7 @@ def _action_in_metadata(action_name: str) -> bool: if md_resp.status_code == 200: txt = md_resp.text return f"Name=\"{action_name}\"" in txt - except Exception: # noqa: BLE001 + except Exception: return False return False @@ -197,108 +201,60 @@ def wait_for_action(action_name: str, timeout_sec: int = 60, interval: float = 2 print({"metadata_present_after_publish": False, "hint": "Invocation retry logic will attempt anyway."}) except requests.exceptions.Timeout: print({"published": False, "error": "PublishAllXml timeout (15s)", "hint": "Proceeding; action may still become available."}) - except Exception as pub_ex: # noqa: BLE001 + except Exception as pub_ex: print({"published": False, "error": str(pub_ex)}) else: print({"publish_strategy": PUBLISH_STRATEGY, "warning": "Unknown strategy value"}) -# Optional: bind an existing plug-in type so invocation returns something meaningful. -# Provide the fully-qualified class name in PLUGIN_TYPENAME above. The plug-in must -# set OutputParameters["ResponseMessage"] (or whatever your response property name is). -if PLUGIN_TYPENAME: - print("Attempt plug-in bind:") - try: - plan(f"lookup plugintype '{PLUGIN_TYPENAME}' then patch custom api") - # Lookup plugintypeid by typename - url = f"{odata.api}/plugintypes" - params = {"$select": "plugintypeid,typename", "$filter": f"typename eq '{PLUGIN_TYPENAME}'"} - r = odata._request("get", url, headers=odata._headers(), params=params) - r.raise_for_status() - vals = r.json().get("value", []) - if vals: - plugintypeid = vals[0]["plugintypeid"] - log_call("odata.update_custom_api (attach plugintype)") - patched = odata.update_custom_api(unique_name=CUSTOM_API_UNIQUE_NAME, changes={ - "plugintypeid@odata.bind": f"/plugintypes({plugintypeid})" - }) - print({"plugin_attached": True, "plugintypeid": plugintypeid}) - else: - print({"plugin_attached": False, "reason": "Plugin typename not found"}) - except Exception as ex: # noqa: BLE001 - resp = getattr(ex, 'response', None) - body = None - if resp is not None: - try: - body = resp.text[:800] - except Exception: # noqa: BLE001 - body = None - print({"plugin_attach_error": str(ex), "body": body}) - -# 3) (Re)List parameters / response properties for visibility -print("Parameters / Response Properties:") +# 4) (Re)List parameters / response properties for visibility +print("List Parameters / Response Properties:") try: params = odata.list_custom_api_request_parameters(customapiid) props = odata.list_custom_api_response_properties(customapiid) print({"parameters": [p.get("name") for p in params], "responses": [p.get("name") for p in props]}) -except Exception as e: # noqa: BLE001 +except Exception as e: print(f"List params/props failed: {e}") -# 4) Invoke the Custom API (will only succeed if backed by server logic) +# 5) Invoke the Custom API print("Invoke Custom API:") -time.sleep(1) try: - # Fetch $metadata to inspect the expected parameter names (diagnostic aid) - try: - meta_url = f"{odata.api}/$metadata" - md_resp = odata._request( - "get", - meta_url, - headers={**odata._headers(), "Accept": "application/xml"}, - ) - md_resp.raise_for_status() - md_text = md_resp.text - # Extract the Action definition snippet for debugging - snippet = None - idx = md_text.find(f"Name=\"{CUSTOM_API_UNIQUE_NAME}\"") - if idx != -1: - snippet = md_text[max(0, idx-200): idx+400] - if snippet: - print({"metadata_snippet": snippet.replace('\n', ' ')[:400]}) - except Exception as md_ex: # noqa: BLE001 - print({"metadata_fetch_error": str(md_ex)}) - - base_message = 123 # Int matches parameter type 6 - # Prefer the unique name first (proved to work in plug-in-less mode); keep logical name fallback for completeness - candidate_param_names = [REQUEST_PARAM_UNIQUE, "Message"] + base_message = 42 # Matches Int32 parameter type + candidate_param_names = [REQUEST_PARAM_UNIQUE] last_error = None for pname in candidate_param_names: - for attempt in range(1,4): # up to 3 attempts each name for propagation + for attempt in range(1,4): # up to 3 attempts each name for propagation / publish delay invoke_payload = {pname: base_message} plan(f"attempt {attempt} param '{pname}' -> odata.call_custom_api('{CUSTOM_API_UNIQUE_NAME}', {invoke_payload})") def invoke(): return odata.call_custom_api(CUSTOM_API_UNIQUE_NAME, invoke_payload) try: result = invoke() - print({"invoked": True, "result": result, "mode": "plugin" if INCLUDE_RESPONSE_PROPERTY else "plugin-less", "used_param": pname, "attempt": attempt}) + print({"invoked": True, "message": "note the None in new_EchoMessage_Response is expected as there is no server logic attached to the workflow", "result": result, "used_param": pname, "attempt": attempt}) raise SystemExit # exit double loop cleanly - except requests.exceptions.HTTPError as ex: # noqa: PERF203 + except requests.exceptions.HTTPError as ex: last_error = ex resp = getattr(ex, 'response', None) + status = getattr(resp, 'status_code', None) body = None if resp is not None: try: - body = resp.text[:300] - except Exception: # noqa: BLE001 + body = resp.text[:600] + except Exception: body = None - if resp is not None and resp.status_code == 400 and "not a valid parameter" in (body or ""): - # Wait and retry + body_lc = (body or "").lower() + # Handle not yet routable (sdkmessage) 404 specially + if status == 404 and 'sdkmessage' in body_lc: + print({"retry": True, "reason": "404 sdkmessage not found (known issue where the custom api exists but metadata is not updated yet)", "attempt": attempt}) + time.sleep(2 + attempt) + continue + if status == 400 and "not a valid parameter" in body_lc: time.sleep(2 + attempt) continue - if resp is not None and resp.status_code == 400 and "Int32" in (body or ""): - print({"hint": "Server expects Int32; ensure payload is int (it is). If still failing, metadata not published yet."}) + if status == 400 and "int32" in body_lc: + print({"hint": "Server expects Int32; payload is int. Likely metadata publish delay."}) time.sleep(2) continue - print({"attempt": pname, "error": str(ex), "body": body}) + print({"attempt": pname, "error": str(ex), "status": status, "body": body}) time.sleep(2) continue if last_error: @@ -315,23 +271,20 @@ def invoke(): body = None print({"invoked": False, "error": str(e), "body": body, "hint": "If 400, verify parameter Type code & payload match; for plug-in-less mode only request param should be present."}) -# 5) Update description (demonstrate patch) -print("Update Custom API description:") +# 6) Update custom api +print("Update Custom API:") try: plan("odata.update_custom_api(unique_name, changes={'description': 'Updated via quickstart'})") updated = backoff_retry(lambda: odata.update_custom_api(unique_name=CUSTOM_API_UNIQUE_NAME, changes={"description": "Updated via quickstart"})) print({"updated": True, "description": updated.get("description")}) -except Exception as e: # noqa: BLE001 +except Exception as e: print({"updated": False, "error": str(e)}) -# 6) Conditional cleanup -if CLEANUP and created_this_run: - print("Cleanup: delete Custom API created in this run") - try: - plan("odata.delete_custom_api(unique_name)") - backoff_retry(lambda: odata.delete_custom_api(unique_name=CUSTOM_API_UNIQUE_NAME)) - print({"deleted": True}) - except Exception as e: # noqa: BLE001 - print({"deleted": False, "error": str(e)}) -else: - print({"cleanup": False, "reason": "CLEANUP flag False or pre-existing API"}) +# 7) Cleanup +print("Cleanup: delete Custom API created in this run") +try: + plan("odata.delete_custom_api(unique_name)") + backoff_retry(lambda: odata.delete_custom_api(unique_name=CUSTOM_API_UNIQUE_NAME)) + print({"deleted": True}) +except Exception as e: + print({"deleted": False, "error": str(e)}) diff --git a/src/dataverse_sdk/odata.py b/src/dataverse_sdk/odata.py index b2b9b46..889c074 100644 --- a/src/dataverse_sdk/odata.py +++ b/src/dataverse_sdk/odata.py @@ -335,10 +335,19 @@ def _get_custom_api(self, *, unique_name: Optional[str] = None, customapiid: Opt return r.json() # unique name path flt = f"uniquename eq '{unique_name}'" - items = self.list_custom_apis(filter_expr=flt) + # First, get just the id via a narrow select, then fetch full record + items = self.list_custom_apis(select=["customapiid"], filter_expr=flt) if not items: return None - return items[0] + cid = items[0].get("customapiid") + if not cid: + return None + url = f"{self.api}/customapis({cid})" + r = self._request("get", url, headers=self._headers()) + if r.status_code == 404: + return None + r.raise_for_status() + return r.json() def get_custom_api(self, unique_name: Optional[str] = None, customapiid: Optional[str] = None) -> Optional[Dict[str, Any]]: """Public accessor for a Custom API metadata record.""" @@ -452,7 +461,22 @@ def create_custom_api( return created def update_custom_api(self, *, unique_name: Optional[str] = None, customapiid: Optional[str] = None, changes: Dict[str, Any]) -> Dict[str, Any]: - """Patch an existing Custom API.""" + """Update an existing Custom API metadata record. + + Parameters + ---------- + unique_name : str, optional + The ``uniquename`` of the Custom API (e.g. ``new_EchoMessage``). Provide this OR ``customapiid``. + customapiid : str, optional + The GUID of the Custom API. Provide this OR ``unique_name``. If both are supplied, ``customapiid`` takes precedence. + changes : dict + A mapping of field names to new values. These are sent directly in the PATCH body. + + Returns + ------- + dict + The updated Custom API record (server representation) because ``Prefer: return=representation`` is used. + """ rec = self._get_custom_api(unique_name=unique_name, customapiid=customapiid) if not rec: raise RuntimeError("Custom API not found") @@ -579,56 +603,6 @@ def list_custom_api_request_parameters(self, customapiid: str) -> List[Dict[str, r.raise_for_status() return r.json().get("value", []) - def create_custom_api_request_parameter( - self, - *, - customapiid: Optional[str] = None, - custom_api_unique_name: Optional[str] = None, - unique_name: str, - name: Optional[str] = None, - data_type: Any = "string", - description: Optional[str] = None, - is_optional: bool = False, - logical_entity_name: Optional[str] = None, - extra: Optional[Dict[str, Any]] = None, - ) -> Dict[str, Any]: - """Create a Custom API Request Parameter. - - If customapiid is not supplied, custom_api_unique_name is resolved first. - data_type can be an int (raw option set) or a friendly string (see _DATA_TYPE_MAP). - """ - if not customapiid: - if not custom_api_unique_name: - raise ValueError("Provide customapiid or custom_api_unique_name") - api_meta = self._get_custom_api(unique_name=custom_api_unique_name) - if not api_meta: - raise RuntimeError(f"Custom API '{custom_api_unique_name}' not found") - customapiid = api_meta.get("customapiid") - body: Dict[str, Any] = (extra or {}).copy() - body.setdefault("uniquename", unique_name) - body.setdefault("name", name or unique_name) - body.setdefault("displayname", name or unique_name) - body.setdefault("isoptional", bool(is_optional)) - body.setdefault("type", self._resolve_data_type(data_type)) - if description: - body.setdefault("description", description) - if logical_entity_name: - body.setdefault("logicalentityname", logical_entity_name) - # Associate to parent Custom API via navigation property - body.setdefault("customapiid@odata.bind", f"/customapis({customapiid})") - url = f"{self.api}/customapirequestparameters" - r = self._request("post", url, headers=self._headers(), json=body) - r.raise_for_status() - return r.json() - - def delete_custom_api_request_parameter(self, request_parameter_id: str) -> None: - url = f"{self.api}/customapirequestparameters({request_parameter_id})" - headers = self._headers().copy() - headers["If-Match"] = "*" - r = self._request("delete", url, headers=headers) - if r.status_code not in (200, 204, 404): - r.raise_for_status() - # --------------- Custom API response properties -------------------- def list_custom_api_response_properties(self, customapiid: str) -> List[Dict[str, Any]]: params = { @@ -639,50 +613,3 @@ def list_custom_api_response_properties(self, customapiid: str) -> List[Dict[str r = self._request("get", url, headers=self._headers(), params=params) r.raise_for_status() return r.json().get("value", []) - - def create_custom_api_response_property( - self, - *, - customapiid: Optional[str] = None, - custom_api_unique_name: Optional[str] = None, - unique_name: str, - name: Optional[str] = None, - data_type: Any = "string", - description: Optional[str] = None, - logical_entity_name: Optional[str] = None, - extra: Optional[Dict[str, Any]] = None, - ) -> Dict[str, Any]: - """Create a Custom API Response Property. - - If customapiid is not supplied, custom_api_unique_name is resolved first. - data_type can be an int (raw option set) or a friendly string (see _DATA_TYPE_MAP). - """ - if not customapiid: - if not custom_api_unique_name: - raise ValueError("Provide customapiid or custom_api_unique_name") - api_meta = self._get_custom_api(unique_name=custom_api_unique_name) - if not api_meta: - raise RuntimeError(f"Custom API '{custom_api_unique_name}' not found") - customapiid = api_meta.get("customapiid") - body: Dict[str, Any] = (extra or {}).copy() - body.setdefault("uniquename", unique_name) - body.setdefault("name", name or unique_name) - body.setdefault("displayname", name or unique_name) - body.setdefault("type", self._resolve_data_type(data_type)) - if description: - body.setdefault("description", description) - if logical_entity_name: - body.setdefault("logicalentityname", logical_entity_name) - body.setdefault("customapiid@odata.bind", f"/customapis({customapiid})") - url = f"{self.api}/customapiresponseproperties" - r = self._request("post", url, headers=self._headers(), json=body) - r.raise_for_status() - return r.json() - - def delete_custom_api_response_property(self, response_property_id: str) -> None: - url = f"{self.api}/customapiresponseproperties({response_property_id})" - headers = self._headers().copy() - headers["If-Match"] = "*" - r = self._request("delete", url, headers=headers) - if r.status_code not in (200, 204, 404): - r.raise_for_status() From 8c216bfff5e50bcf6e81cd295fb5020c4f2eec09 Mon Sep 17 00:00:00 2001 From: Max Wang Date: Thu, 28 Aug 2025 16:06:27 -0700 Subject: [PATCH 3/3] update to align the org url input as other quickstarts --- examples/quickstart_custom_api.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/examples/quickstart_custom_api.py b/examples/quickstart_custom_api.py index dbec695..4679fe0 100644 --- a/examples/quickstart_custom_api.py +++ b/examples/quickstart_custom_api.py @@ -13,7 +13,16 @@ from azure.identity import InteractiveBrowserCredential # ---------------- Configuration ---------------- -base_url = "https://aurorabapenv0f528.crm10.dynamics.com" # <-- change to your environment +if not sys.stdin.isatty(): + print("Interactive input required for org URL. Run this script in a TTY.") + sys.exit(1) +entered = input("Enter Dataverse org URL (e.g. https://yourorg.crm.dynamics.com): ").strip() +if not entered: + print("No URL entered; exiting.") + sys.exit(1) +base_url = entered.rstrip('/') +client = DataverseClient(base_url=base_url, credential=InteractiveBrowserCredential()) + CUSTOM_API_UNIQUE_NAME = "new_EchoMessage" # Must be globally unique in the org REQUEST_PARAM_UNIQUE = "new_EchoMessage_Message" RESPONSE_PROP_UNIQUE = "new_EchoMessage_Response"