| name | rest-api-patterns |
| description | Guide for implementing REST API wrapper functions. Use this when adding new API wrappers or troubleshooting API calls. |
REST API Patterns
This skill covers the patterns and utilities for implementing REST API wrapper functions in Semantic Link Labs.
When to Use This Skill
Use this skill when you need to:
- Implement new REST API wrappers
- Understand the
_base_api helper function
- Handle pagination, long-running operations, or errors
- Debug API-related issues
Finding API Documentation
Before implementing a wrapper, use the API search tool to find the relevant documentation:
Using search_public_api_doc.py
cd .claude/skills/rest-api-patterns/scripts
python search_public_api_doc.py "dataset refresh"
python search_public_api_doc.py "create item" --source fabric
python search_public_api_doc.py "gateway" --source powerbi
python search_public_api_doc.py "workspace" --limit 10
Example Output
🔍 Searching for: 'dataset refresh' in Fabric + Power BI
================================================================================
📥 Fetching Microsoft Fabric TOC...
✅ Loaded 15 top-level categories from Microsoft Fabric
📥 Fetching Power BI TOC...
✅ Loaded 20 top-level categories from Power BI
Found 5 results:
1. [POWERBI] Datasets - Refresh Dataset In Group (score: 95.0)
URL: https://learn.microsoft.com/en-us/rest/api/power-bi/datasets/refresh-dataset-in-group
Path: Datasets > Refresh Dataset In Group
2. [POWERBI] Datasets - Get Refresh History In Group (score: 90.0)
URL: https://learn.microsoft.com/en-us/rest/api/power-bi/datasets/get-refresh-history-in-group
Path: Datasets > Get Refresh History In Group
Requirements
The script requires rapidfuzz and requests:
pip install rapidfuzz requests
API Documentation References
API Client Architecture
Semantic Link Labs uses sempy.fabric.FabricRestClient as the underlying HTTP client, wrapped by the _base_api helper function.
Key Components
| Component | Purpose |
|---|
_base_api | Main helper for all API calls |
FabricRestClient | HTTP client from sempy |
pagination | Handles paginated responses |
lro | Handles long-running operations |
The _base_api Function
Located in src/sempy_labs/_helper_functions.py, this is the standard way to make API calls.
Function Signature
def _base_api(
request: str,
client: str = "fabric",
method: str = "get",
payload: Optional[str] = None,
status_codes: Optional[int] = 200,
uses_pagination: bool = False,
lro_return_json: bool = False,
lro_return_status_code: bool = False,
lro_return_df: bool = False,
):
Client Types
| Client | Use Case | Authentication |
|---|
fabric | Standard Fabric API | Default notebook credentials |
fabric_sp | Fabric API with SP support | Service Principal or default |
azure | Azure Resource Manager | Service Principal |
graph | Microsoft Graph | Service Principal |
onelake | OneLake storage | Storage token |
Return Types
The _base_api function returns different types depending on the parameters used:
| Parameters | Return Type | How to Access Data |
|---|
| Default (no special flags) | Response object | Call .json() to get dict |
uses_pagination=True | list[dict] | Iterate over list, each item has .get("value", []) |
lro_return_json=True | dict | Access directly, already parsed JSON |
lro_return_status_code=True | int | HTTP status code |
lro_return_df=True | DataFrame | Use directly |
⚠️ COMMON MISTAKE: Forgetting to call .json() on the response for simple GET requests.
response = _base_api(request=f"/v1/workspaces/{workspace_id}/items/{item_id}")
name = response.get("displayName")
response = _base_api(request=f"/v1/workspaces/{workspace_id}/items/{item_id}").json()
name = response.get("displayName")
Common API Patterns
Simple GET Request
from sempy_labs._helper_functions import _base_api
response = _base_api(
request=f"/v1/workspaces/{workspace_id}/items",
client="fabric_sp",
)
data = response.json()
POST Request with Payload
payload = {
"displayName": name,
"description": description,
}
response = _base_api(
request=f"/v1/workspaces/{workspace_id}/items",
method="post",
payload=payload,
status_codes=[201, 202],
client="fabric_sp",
)
DELETE Request
_base_api(
request=f"/v1/workspaces/{workspace_id}/items/{item_id}",
method="delete",
client="fabric_sp",
)
PATCH Request
payload = {
"displayName": new_name,
}
_base_api(
request=f"/v1/workspaces/{workspace_id}/items/{item_id}",
method="patch",
payload=payload,
client="fabric_sp",
)
Handling Pagination
For APIs that return paginated results:
from sempy_labs._helper_functions import _base_api, _create_dataframe
columns = {
"Id": "string",
"Name": "string",
}
df = _create_dataframe(columns=columns)
responses = _base_api(
request=f"/v1/workspaces/{workspace_id}/items",
uses_pagination=True,
client="fabric_sp",
)
rows = []
for r in responses:
for item in r.get("value", []):
rows.append({
"Id": item.get("id"),
"Name": item.get("displayName"),
})
if rows:
df = pd.DataFrame(rows)
return df
Handling Long-Running Operations (LRO)
Some APIs return 202 Accepted and require polling for completion.
Return JSON When Complete
result = _base_api(
request=f"/v1/workspaces/{workspace_id}/items/{item_id}/getDefinition",
method="post",
lro_return_json=True,
client="fabric_sp",
)
definition = result.get("definition")
Return Status Code
status = _base_api(
request=f"/v1/workspaces/{workspace_id}/items",
method="post",
payload=payload,
lro_return_status_code=True,
client="fabric_sp",
)
if status == 200:
print("Operation completed successfully")
Error Handling
Expected Status Codes
Specify expected status codes to avoid exceptions:
response = _base_api(
request=url,
method="post",
payload=payload,
status_codes=[200, 201, 202],
client="fabric_sp",
)
FabricHTTPException
When status code doesn't match, FabricHTTPException is raised:
from sempy.fabric.exceptions import FabricHTTPException
try:
response = _base_api(
request=f"/v1/workspaces/{workspace_id}/items/{item_id}",
client="fabric_sp",
)
except FabricHTTPException as e:
if e.response.status_code == 404:
print(f"Item not found")
else:
raise
Building URLs with Parameters
Use the _build_url helper for query parameters:
from sempy_labs._helper_functions import _build_url
url = "/v1/admin/workspaces"
params = {
"capacityId": capacity_id,
"state": "Active",
}
url = _build_url(url, params)
responses = _base_api(
request=url,
uses_pagination=True,
client="fabric_sp",
)
API Endpoint Patterns
Fabric Core API
f"/v1/workspaces/{workspace_id}/items"
f"/v1/workspaces/{workspace_id}/items/{item_id}"
f"/v1/workspaces/{workspace_id}/items/{item_id}/getDefinition"
f"/v1/workspaces/{workspace_id}/items/{item_id}/updateDefinition"
Fabric Admin API
"/v1/admin/workspaces"
"/v1/admin/items"
"/v1/admin/capacities"
Power BI REST API
f"/v1.0/myorg/groups/{workspace_id}/..."
f"/v1.0/myorg/groups/{workspace_id}/datasets/{dataset_id}/..."
f"/v1.0/myorg/groups/{workspace_id}/reports/{report_id}/..."
Azure Resource Manager
f"https://management.azure.com/subscriptions/{subscription_id}/providers/Microsoft.Fabric/capacities"
f"https://management.azure.com/subscriptions/{subscription_id}/resourceGroups/{resource_group}"
Authentication Headers
For non-Fabric clients (Azure, Graph), use _get_headers:
from sempy_labs._authentication import _get_headers
import sempy_labs._authentication as auth
headers = _get_headers(auth.token_provider.get(), audience="azure")
response = requests.get(
url,
headers=headers,
)
Creating Result DataFrames
Use _create_dataframe for consistent empty DataFrames:
from sempy_labs._helper_functions import _create_dataframe
columns = {
"Id": "string",
"Name": "string",
"Type": "string",
"Created Date": "datetime",
"Size": "int",
}
df = _create_dataframe(columns=columns)
Updating DataFrame Types
from sempy_labs._helper_functions import _update_dataframe_datatypes
column_map = {
"Created Date": "datetime",
"Size": "int",
"Is Active": "bool",
}
_update_dataframe_datatypes(df, column_map)
Complete Example
from sempy._utils._log import log
from sempy_labs._helper_functions import (
resolve_workspace_name_and_id,
_base_api,
_create_dataframe,
_build_url,
)
import sempy_labs._icons as icons
from typing import Optional
from uuid import UUID
import pandas as pd
@log
def list_my_items(
item_type: Optional[str] = None,
workspace: Optional[str | UUID] = None,
) -> pd.DataFrame:
"""
Lists items in a workspace.
This is a wrapper function for the following API: `Items - List Items <https://learn.microsoft.com/rest/api/fabric/core/items/list-items>`_.
Service Principal Authentication is supported.
Parameters
----------
item_type : str, default=None
Filter by item type.
workspace : str | uuid.UUID, default=None
The Fabric workspace name or ID.
Defaults to None which resolves to the workspace of the attached lakehouse.
Returns
-------
pandas.DataFrame
A pandas dataframe showing items in the workspace.
"""
(workspace_name, workspace_id) = resolve_workspace_name_and_id(workspace)
columns = {
"Id": "string",
"Name": "string",
"Type": "string",
}
df = _create_dataframe(columns=columns)
url = f"/v1/workspaces/{workspace_id}/items"
params = {}
if item_type:
params["type"] = item_type
if params:
url = _build_url(url, params)
responses = _base_api(
request=url,
uses_pagination=True,
client="fabric_sp",
)
rows = []
for r in responses:
for item in r.get("value", []):
rows.append({
"Id": item.get("id"),
"Name": item.get("displayName"),
"Type": item.get("type"),
})
if rows:
df = pd.DataFrame(rows)
return df
Debugging API Calls
Print Response Details
response = _base_api(
request=url,
client="fabric_sp",
)
print(f"Status: {response.status_code}")
print(f"Headers: {response.headers}")
print(f"Body: {response.json()}")
Check Request Being Made
Add temporary debug prints:
print(f"Making request to: {url}")
print(f"Payload: {payload}")
API Documentation References