Forum Discussion

SnoekLaurens's avatar
SnoekLaurens
Advocate I
1 year ago
Solved

Notebook semantic link (sempy) extract report usage

Hi, does anyone know if there is a way to use Python semantic link package to extract report usage form all reports that are in a workspace or connected to a certain shared dataset in power bi Servic...
data engineering
notebook
  • SnoekLaurens's avatar
    SnoekLaurens
    1 year ago

    Already found it 🙂

     

    This works for me:

     

    import sempy.fabric as fabric
    import json

    method = 'GET'
    start_time = "'2024-12-11T00:55:00.000Z'"
    end_time = "'2024-12-11T12:00:00.000Z'"
    filter = "$filter=Activity eq 'viewreport'"
    query = '&'.join(['startDateTime='+start_time,'endDateTime='+end_time,filter])
    full_url = url + '?' + query

    client = fabric.PowerBIRestClient()

    def get_activity_events(url😞
        response = client.request(method, url)
        data = response.json()
       
        # Process the current batch of events
        events = data.get('activityEventEntities', [])
       
        # Check for continuation
        continuation_uri = data.get('continuationUri')
        while continuation_uri:
            response = client.request(method, continuation_uri)
            data = response.json()
            events.extend(data.get('activityEventEntities', []))
            continuation_uri = data.get('continuationUri')
       
        return events

    events = get_activity_events(full_url)
    print(json.dumps(events, indent=4))
govindarajan_d's avatar
govindarajan_d
Super User
1 year ago

SnoekLaurens, I don't think the semantic model that powers the Report Usage is directly accessible right now. One another way I would think of is using the REST API Get Activity Events (requires admin role though). But it is not as detailed as Report Usage, for e.g. getting the page level information is not possible. 

 

But calling a PBI REST API is much easier in Spark notebook inside Fabric, as you can get the token using mssparkutils. 

 

import requests
from pyspark.sql import SparkSession
import json
from notebookutils import mssparkutils as msu

def pbi_rest_api(method, uri, payload=None):
    """
    Makes a REST API call to Power BI. Token is automatically generated from mssparkutils based on the user id running the notebook

    Args:
        method (str): The HTTP method ('GET' or 'POST').
        uri (str): The later part of the REST API URL.
        payload (dict): The payload, if it is a POST request.

    Returns:
        Response object: The response from the REST API.
    """

    endpoint = "https://api.powerbi.com/v1.0/myorg"
    url = f"{endpoint}/{uri}"

    headers = {
        "Authorization": "Bearer " + msu.credentials.getToken("pbi"),
        "Content-Type": "application/json"
    }
    session = requests.Session()
    try:
        response = session.request(method, url, headers=headers, json=payload, timeout=120)
        response.raise_for_status()
        return response.json()
    except Exception as e:
        print(f"Exception raised {e}")

 

 

SnoekLaurens's avatar
SnoekLaurens
Advocate I
1 year ago

What do I need to add to get the results correctly shown in a dataframe?

My results don't seem to be the report views but it shows me the continuationToken

 

  • SnoekLaurens's avatar
    SnoekLaurens
    Advocate I
    1 year ago

    Already found it 🙂

     

    This works for me:

     

    import sempy.fabric as fabric
    import json

    method = 'GET'
    start_time = "'2024-12-11T00:55:00.000Z'"
    end_time = "'2024-12-11T12:00:00.000Z'"
    filter = "$filter=Activity eq 'viewreport'"
    query = '&'.join(['startDateTime='+start_time,'endDateTime='+end_time,filter])
    full_url = url + '?' + query

    client = fabric.PowerBIRestClient()

    def get_activity_events(url😞
        response = client.request(method, url)
        data = response.json()
       
        # Process the current batch of events
        events = data.get('activityEventEntities', [])
       
        # Check for continuation
        continuation_uri = data.get('continuationUri')
        while continuation_uri:
            response = client.request(method, continuation_uri)
            data = response.json()
            events.extend(data.get('activityEventEntities', []))
            continuation_uri = data.get('continuationUri')
       
        return events

    events = get_activity_events(full_url)
    print(json.dumps(events, indent=4))