Summary
(If you don't have Ten Minutes)
1. If you have a lot of LakeHouses in your Fabric environment, and want to quickly audit them, or build a pipeline that will gather the information about Workspace, Lake House, and Table, the final code is at the end.
2. Two tools that may be used - SemPy library & direct Fabric API.
3. SemPy provides simplicity, API - provides customized scenarios.
4. Automation - it is easy. No need to manually look for duplicated tables or the other necessary info. This article begins this topic in depth.
5. If SemPy is used, the default version should be upgraded to get the maximum benefits.
Section 1. Introduction
I often face challenges when I, before ingesting a new table, need to understand the current assets, to avoid creating redundant pipelines. If the Fabric has only a few lakehouses, it is not a big challenge. But what if the company’s data assets are built in the Data Mesh style, where each business domain is represented as a separate lakehouse, with interchangeable connections, and the company has 50-200 lakehouses? In that or similar cases, quickly identifying the proper table involves a lot of manual work. Documentation may be inaccurate. And that’s where the automation of that may significantly reduce total manual work on that.
This article is the beginning of a deeper dive into Fabric Automation using APIs and native libraries. Thinking broadly, we may cover many other cases later, in addition to the table audit.
We will use two types of solutions: the SemPy library and the native API for Fabric (v1).
1.1. SemPy - Abstraction Layer over Fabric API
First of all, let’s start with a simple possible solution - the SemPy library. Let’s structure that.
Why the SemPy library:
- Microsoft natively supports it
- It provides simplicity over the core operations on top of the API
- It is natively included in MS Fabric (but a manual update is needed; the installed version is too old)
What are the possible limitations of the SemPy library:
- The new features may have some lag in delivery compared to the API
- Complex scenarios may be limited
1.2. Solution with API
What we need to take into account, if we consider using an API to automate things in Fabric:
- API provides the most complex options to manage Fabric
- API is stable and reliable, meaning you don’t need to rewrite the automations each month
- API is not so complex, taking into account that it is well documented, and LLMs can learn from that to assist you
What are the factors to consider before using an API instead of SemPy:
- When using API, you need to understand its pitfalls, such as pagination and throttling (SemPy may handle it for you)
- With API, your code will be 20-30% heavier (when the common functions are modularized and moved to a common notebook)
1.3. Sempy vs API
So, if your goal is Fabric Governance Automation, both options are worth considering. Choose SemPy if simplicity is a priority. Choose API if you need a strong foundation and the power.
Now, let’s consider the solution we can build to begin automating Fabric Governance.
Simple observation of the tables in the specific workspace, using SemPy:
%pip install --upgrade semantic-link
# For convenience, move it to a separate cell
import sempy.fabric as fabric
import pandas as pd
LH = "LH_AuditMe_NoSchema" # replace by yours, with NO SCHEMA enabled
df = fabric.lakehouse.list_lakehouse_tables(
lakehouse=LH,
workspace=fabric.get_workspace_id()
)
display(df)
The same, but done using native API:
import requests
from notebookutils import mssparkutils
from urllib.parse import urlparse, urlencode, parse_qsl, urlunparse
from pyspark.sql.types import StructType, StructField, StringType
TARGET_LAKEHOUSE = "LH_AuditMe_NoSchema"
access_token = mssparkutils.credentials.getToken("https://api.fabric.microsoft.com")
api_headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
API_ROOT = "https://api.fabric.microsoft.com/v1"
workspace_id = spark.conf.get("trident.workspace.id")
def get_all_pages(url, result_key=("value", "data")):
if isinstance(result_key, str):
result_key = (result_key,)
all_items = []
while url:
resp = requests.get(url, headers=api_headers, timeout=30)
resp.raise_for_status()
body = resp.json()
items = next((body[k] for k in result_key if k in body), [])
all_items.extend(items)
if body.get("continuationUri"):
url = body["continuationUri"]
elif body.get("continuationToken"):
parsed = urlparse(url)
params = dict(parse_qsl(parsed.query))
params["continuationToken"] = body["continuationToken"]
url = urlunparse(parsed._replace(query=urlencode(params)))
else:
url = None
return all_items
# Find the target lakehouse
lakehouses = get_all_pages(f"{API_ROOT}/workspaces/{workspace_id}/lakehouses")
lakehouse = next((lh for lh in lakehouses if lh["displayName"] == TARGET_LAKEHOUSE), None)
if not lakehouse:
raise ValueError(f"Lakehouse '{TARGET_LAKEHOUSE}' not found in current workspace.")
lakehouse_id = lakehouse["id"]
print(f"Found lakehouse: {TARGET_LAKEHOUSE} ({lakehouse_id})")
# Get tables
tables = get_all_pages(
f"{API_ROOT}/workspaces/{workspace_id}/lakehouses/{lakehouse_id}/tables",
result_key="data"
)
records = [
{
"table_name": t.get("name"),
"table_type": t.get("type"),
"location": t.get("location"),
"format": t.get("format"),
}
for t in tables
]
schema = StructType([
StructField("table_name", StringType(), True),
StructField("table_type", StringType(), True),
StructField("location", StringType(), True),
StructField("format", StringType(), True),
])
df = spark.createDataFrame(records, schema=schema)
df.createOrReplaceTempView("lakehouse_tables")
print(f"View 'lakehouse_tables' created — {df.count()} tables found.")
display(df)
So, you may see, the SemPy solution is much more elegant and shorter. Also, it nicely handles pagination - a hidden API thing that may provide false results if not processed properly.
Section 2. From Theory to Practical Implementation
Solution requirements
- Gives the list of tables across all workspaces and lakehouses, to audit, find duplicates, monitor, etc.
Solution limitation
- Doesn't support lakehouses with schema enabled (because it is still in Public Preview, and the API doesn't support it yet)
Prerequisites
- Microsoft Fabric Notebook must be created
- "%pip install --upgrade semantic-link" must be run at the beginning, for the SemPy version of the solution
2.1. API Version
Copy and paste that code into the Spark Notebook. It returns a dataframe with the list of tables in each workspace, as well as the list of Lakehouses that weren't processed due to issues (most probably the schema-enabled ones).
import requests
from notebookutils import mssparkutils
from urllib.parse import urlparse, urlencode, parse_qsl, urlunparse
from pyspark.sql.types import StructType, StructField, StringType
access_token = mssparkutils.credentials.getToken(
"https://api.fabric.microsoft.com"
)
api_headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json"
}
API_ROOT = "https://api.fabric.microsoft.com/v1"
# -----------------------------------------------------------
# Generic paginated GET — works for all Fabric list endpoints
# -----------------------------------------------------------
def _get_with_retry(url, headers, timeout=30, retries=3, backoff=1.0):
for attempt in range(retries + 1):
resp = requests.get(url, headers=headers, timeout=timeout)
if resp.status_code < 500 and resp.status_code != 429:
resp.raise_for_status()
return resp
if attempt == retries:
resp.raise_for_status()
wait = float(resp.headers.get("Retry-After", backoff * (2 ** attempt)))
time.sleep(wait)
def get_all_pages(url, result_key=("value", "data")):
"""
result_key: a string, or a tuple of keys tried in order.
The first key present in the response body is used.
"""
if isinstance(result_key, str):
result_key = (result_key,)
all_items = []
while url:
body = _get_with_retry(url, api_headers).json()
items = next((body[k] for k in result_key if k in body), [])
all_items.extend(items)
if body.get("continuationUri"):
url = body["continuationUri"]
elif body.get("continuationToken"):
parsed = urlparse(url)
params = dict(parse_qsl(parsed.query))
params["continuationToken"] = body["continuationToken"]
url = urlunparse(parsed._replace(query=urlencode(params)))
else:
url = None
return all_items
# -----------------------------------------------------------
# 1. Get ALL workspaces (paginated)
# -----------------------------------------------------------
workspace_list = get_all_pages(f"{API_ROOT}/workspaces", result_key="value")
print(f"Discovered {len(workspace_list)} workspaces\n")
inventory_records = []
schema_enabled_lakehouses = []
failed_workspaces = []
# -----------------------------------------------------------
# 2. Loop through workspaces → lakehouses → tables
# -----------------------------------------------------------
for workspace in workspace_list:
workspace_id = workspace["id"]
workspace_name = workspace["displayName"]
print(f"Processing workspace: {workspace_name}")
try:
# Get ALL lakehouses in this workspace (paginated)
lakehouse_list = get_all_pages(
f"{API_ROOT}/workspaces/{workspace_id}/lakehouses",
result_key="value"
)
for lakehouse in lakehouse_list:
lakehouse_id = lakehouse["id"]
lakehouse_name = lakehouse["displayName"]
# Check if lakehouse is schema-enabled
properties = lakehouse.get("properties", {})
if (
properties.get("defaultSchema") is not None
or properties.get("enableSchemas", False)
):
schema_enabled_lakehouses.append({
"workspace_name": workspace_name,
"lakehouse_name": lakehouse_name,
})
print(f" ⚠ Skipped '{lakehouse_name}' — schema-enabled lakehouse (REST API not supported)")
continue
# Get ALL tables in this lakehouse (paginated)
table_list = get_all_pages(
f"{API_ROOT}/workspaces/{workspace_id}/lakehouses/{lakehouse_id}/tables",
result_key="data"
)
for table in table_list:
inventory_records.append({
"workspace_name": workspace_name,
"lakehouse_name": lakehouse_name,
"table_name": table.get("name"),
"table_type": table.get("type"),
"location": table.get("location"),
"format": table.get("format")
})
print(f" ✓ Workspace processed successfully\n")
except Exception as ex:
failed_workspaces.append({
"workspace_name": workspace_name,
"error": str(ex),
})
print(f" ✗ Workspace failed: {workspace_name}")
print(f" Error: {ex}\n")
inventory_schema = StructType([
StructField("workspace_name", StringType(), True),
StructField("lakehouse_name", StringType(), True),
StructField("table_name", StringType(), True),
StructField("table_type", StringType(), True),
StructField("location", StringType(), True),
StructField("format", StringType(), True),
])
if inventory_records:
spark_df = spark.createDataFrame(inventory_records, schema=inventory_schema)
spark_df.createOrReplaceTempView("fabric_lakehouse_inventory")
print("A view fabric_lakehouse_inventory created")
print(f"Total tables indexed: {spark_df.count()}")
else:
print("No lakehouses or tables found.")
if schema_enabled_lakehouses:
print(f"\n⚠ Skipped {len(schema_enabled_lakehouses)} schema-enabled lakehouse(s):")
for lh in schema_enabled_lakehouses:
print(f" - {lh['workspace_name']} → {lh['lakehouse_name']}")
print(" (The REST API /tables endpoint does not support schema-enabled lakehouses yet)")
if failed_workspaces:
print(f"\n✗ Failed to process {len(failed_workspaces)} workspace(s):")
for fw in failed_workspaces:
print(f" - {fw['workspace_name']}: {fw['error']}")
query_text = """SELECT
workspace_name,
lakehouse_name,
table_name,
table_type,
location,
format
FROM fabric_lakehouse_inventory
"""
if schema_enabled_lakehouses:
display(spark.sql(query_text))
print("Use that query to get data:")
print("****************")
print("%%sql")
print(query_text)2.2. SemPy-Based Script
If you prefer the SemPy version of the script, use that version of the solution. It is functionally identical, but utilizes the SemPy library.
%pip install --upgrade semantic-link
import sempy.fabric as fabric
import pandas as pd
from pyspark.sql.types import StructType, StructField, StringType
all_tables = []
failed = []
workspaces = fabric.list_workspaces()
for _, ws in workspaces.iterrows():
ws_id = ws["Id"]
ws_name = ws["Name"]
try:
lakehouses = fabric.list_items(workspace=ws_id, item_type="Lakehouse")
for _, lh in lakehouses.iterrows():
lh_name = lh["Display Name"]
try:
df = fabric.lakehouse.list_lakehouse_tables(
lakehouse=lh_name,
workspace=ws_id
)
df.insert(0, "workspace_name", ws_name)
df.insert(1, "lakehouse_name", lh_name)
all_tables.append(df)
except Exception:
failed.append({"workspace": ws_name, "lakehouse": lh_name})
except Exception:
failed.append({"workspace": ws_name, "lakehouse": "N/A"})
if all_tables:
result = pd.concat(all_tables, ignore_index=True)
# Normalize column names to lowercase with underscores
result.columns = [c.lower().replace(" ", "_") for c in result.columns]
inventory = result.rename(columns={
"name": "table_name",
"type": "table_type",
"format": "format",
"location": "location",
})[["workspace_name", "lakehouse_name", "table_name", "table_type", "location", "format"]]
spark_df = spark.createDataFrame(inventory.astype(str))
spark_df.createOrReplaceTempView("fabric_lakehouse_inventory")
print(f"Total tables indexed: {spark_df.count()}")
print("View fabric_lakehouse_inventory created. Query it with:")
q = "SELECT workspace_name, lakehouse_name, table_name, table_type, location, format FROM fabric_lakehouse_inventory"
print("%%sql")
print(q)
display(spark.sql("SELECT workspace_name, lakehouse_name, table_name, table_type, location, format FROM fabric_lakehouse_inventory"))
else:
print("No tables found.")
if failed:
print(f"\nFailed lakehouses ({len(failed)}):")
display(pd.DataFrame(failed))tables audit
Section 3. Next Steps & Conclusion
API in Fabric is powerful. SemPy gives a great level of abstraction on that, and simplifies the development and level of effort. On the other hand, API gives a better grip on the solution, and may be beneficial on scale, but is more complex and requires more attention to the details, such as pagination. Also, API may be used if more complex scenarios are needed, or if the SemPy library hasn't been updated to the latest feature.
These scripts reviewed here may be used manually when you need to quickly get the information about the tables. Also, it may be used automatically when you run it as a pipeline and update a governing table. This second scenario may also allow you to monitor changes in table structure over time.
Resources
Fabric REST API: https://learn.microsoft.com/en-us/rest/api/fabric/articles/
Pagination: https://learn.microsoft.com/en-us/rest/api/fabric/articles/pagination
Throttling: https://learn.microsoft.com/en-us/rest/api/fabric/articles/throttling
Repo with examples: https://github.com/dataassets1/fabric-pulse
SemPy documentation: https://learn.microsoft.com/en-us/python/api/semantic-link-sempy/sempy.fabric.lakehouse?view=semantic-link-python
Advocate III