rest-api-patterns

🔍 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

pip install rapidfuzz requests

def _base_api(
    request: str,                      # API endpoint path
    client: str = "fabric",            # Client type
    method: str = "get",               # HTTP method
    payload: Optional[str] = None,     # Request body
    status_codes: Optional[int] = 200, # Expected status codes
    uses_pagination: bool = False,     # Enable pagination
    lro_return_json: bool = False,     # Wait for LRO, return JSON
    lro_return_status_code: bool = False,  # Wait for LRO, return status
    lro_return_df: bool = False,       # Wait for LRO, return DataFrame
):

# ❌ WRONG - response is a Response object, not a dict
response = _base_api(request=f"/v1/workspaces/{workspace_id}/items/{item_id}")
name = response.get("displayName")  # AttributeError: 'Response' object has no attribute 'get'
 
# ✅ CORRECT - call .json() to get the dict
response = _base_api(request=f"/v1/workspaces/{workspace_id}/items/{item_id}").json()
name = response.get("displayName")  # Works!

from sempy_labs._helper_functions import _base_api
 
response = _base_api(
    request=f"/v1/workspaces/{workspace_id}/items",
    client="fabric_sp",
)
 
data = response.json()

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",
)

_base_api(
    request=f"/v1/workspaces/{workspace_id}/items/{item_id}",
    method="delete",
    client="fabric_sp",
)

payload = {
    "displayName": new_name,
}
 
_base_api(
    request=f"/v1/workspaces/{workspace_id}/items/{item_id}",
    method="patch",
    payload=payload,
    client="fabric_sp",
)

from sempy_labs._helper_functions import _base_api, _create_dataframe
 
columns = {
    "Id": "string",
    "Name": "string",
}
df = _create_dataframe(columns=columns)
 
# Get all pages
responses = _base_api(
    request=f"/v1/workspaces/{workspace_id}/items",
    uses_pagination=True,
    client="fabric_sp",
)
 
# Process all responses
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

result = _base_api(
    request=f"/v1/workspaces/{workspace_id}/items/{item_id}/getDefinition",
    method="post",
    lro_return_json=True,
    client="fabric_sp",
)
 
# Result contains the final JSON response
definition = result.get("definition")

status = _base_api(
    request=f"/v1/workspaces/{workspace_id}/items",
    method="post",
    payload=payload,
    lro_return_status_code=True,
    client="fabric_sp",
)
 
# status is the final HTTP status code
if status == 200:
    print("Operation completed successfully")

# Accept 200, 201, or 202 as success
response = _base_api(
    request=url,
    method="post",
    payload=payload,
    status_codes=[200, 201, 202],
    client="fabric_sp",
)

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

from sempy_labs._helper_functions import _build_url
 
url = "/v1/admin/workspaces"
params = {
    "capacityId": capacity_id,
    "state": "Active",
}
 
url = _build_url(url, params)
# Result: "/v1/admin/workspaces?capacityId=xxx&state=Active"
 
responses = _base_api(
    request=url,
    uses_pagination=True,
    client="fabric_sp",
)

# List items in workspace
f"/v1/workspaces/{workspace_id}/items"
 
# Get specific item
f"/v1/workspaces/{workspace_id}/items/{item_id}"
 
# Item operations
f"/v1/workspaces/{workspace_id}/items/{item_id}/getDefinition"
f"/v1/workspaces/{workspace_id}/items/{item_id}/updateDefinition"

# Admin workspaces
"/v1/admin/workspaces"
 
# Admin items
"/v1/admin/items"
 
# Capacities
"/v1/admin/capacities"

# Groups (workspaces)
f"/v1.0/myorg/groups/{workspace_id}/..."
 
# Datasets
f"/v1.0/myorg/groups/{workspace_id}/datasets/{dataset_id}/..."
 
# Reports
f"/v1.0/myorg/groups/{workspace_id}/reports/{report_id}/..."

# Fabric capacities
f"https://management.azure.com/subscriptions/{subscription_id}/providers/Microsoft.Fabric/capacities"
 
# Resource groups
f"https://management.azure.com/subscriptions/{subscription_id}/resourceGroups/{resource_group}"

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,
)

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)

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)

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

response = _base_api(
    request=url,
    client="fabric_sp",
)
 
print(f"Status: {response.status_code}")
print(f"Headers: {response.headers}")
print(f"Body: {response.json()}")

print(f"Making request to: {url}")
print(f"Payload: {payload}")