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 new file mode 100644 index 0000000..4679fe0 --- /dev/null +++ b/examples/quickstart_custom_api.py @@ -0,0 +1,299 @@ +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 +from azure.identity import InteractiveBrowserCredential + +# ---------------- Configuration ---------------- +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" +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()) +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: + 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) Check if target Custom API exists +print("Check target Custom API existence:") +try: + 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, 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: + 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: + print({"delete_prior_error": str(del_ex)}) + +plan("odata.create_custom_api (inline request parameter + response property)") +try: + 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, + "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() + 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) + +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: + 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: + 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: + print({"published": False, "error": str(pub_ex)}) +else: + print({"publish_strategy": PUBLISH_STRATEGY, "warning": "Unknown strategy value"}) + +# 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: + print(f"List params/props failed: {e}") + +# 5) Invoke the Custom API +print("Invoke Custom API:") +try: + 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 / 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, "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: + 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[:600] + except Exception: + body = None + 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 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), "status": status, "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."}) + +# 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: + print({"updated": False, "error": str(e)}) + +# 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 5bf39cf..889c074 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,314 @@ 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}'" + # 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 + 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.""" + 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]: + """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") + 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", []) + + # --------------- 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", [])