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 Service?

Or is there any other good alternative to do this easily?

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

7 Replies

  • spencer_sa's avatar
    spencer_sa
    Impactful Individual
    1 year ago

    One thing I've been doing is using the Fabric Capacity Metrics App that installs a semantic model (and a report) that contains the usage metrics for the previous 14(?) days.
    https://learn.microsoft.com/en-us/fabric/enterprise/metrics-app-install?tabs=1st

    You can then connect to that semantic model with semantic_link_labs aka sempy and extract data from these;

    df = spark.createDataFrame(fabric.read_table(dataset = datasetID, table = table, workspace = workspaceID))

    Useful tables are Items, MetricsByItemandOperationandHour, MetricsByItemandOperationandDay.
    Be aware the data does not persist beyond 14 days, so you'll need to stash any historic data in a suitable place like a lakehouse.

    If this post helps, please Accept it as Solution to help other members find it.

    • SnoekLaurens's avatar
      SnoekLaurens
      Advocate I
      1 year ago

      Hi spencer_sa,

      I got the data from the tables you adviced and loaded them into my lakehouse.

      Now next step is to understand how I can get the report usage from that data. Should these be specific items? I guess it would only work if the reports are also in a workspace with fabric capacity?

      • 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}")