Hello Fabric Community,
I’ve been struggling with the complex execution context in Fabric (p.s. Who's Calling?).
I created a notebook to deploy a (template) Semantic Model from a central workspace to a target workspace using sempy_labs.labs.
When I execute this notebook using a user identity, it works ✅
When I execute the notebook inside a pipeline with a connection (authenticated using a Service Principal), it fails ❌ with the following error (though the SP has contributor access on both source and target WS):
FabricHTTPException
403 Forbidden for URL: https://api.fabric.microsoft.com/v1.0/myorg/groups/xxxxxxxx-yyyyy-zzzzz-000000/datasets
Headers: {'Content-Length': '0', 'Strict-Transport-Security': 'max-age=31536000; includeSubDomains', 'X-Frame-Options': 'deny', 'X-Content-Type-Options': 'nosniff', 'Access-Control-Expose-Headers': 'RequestId', 'RequestId': 'a9bd2bed-ff7b-4c1c-b28d-551727ebe685', 'Date': 'Mon, 20 Apr 2026 18:56:10 GMT'}
When I execute the notebook inside a data pipeline without a connection, it occasionally works, but mostly throws this error:
WorkspaceNotFoundException
Workspace 'xxxxxxxx-yyyyy-zzzzz-000000' not found.
Below is the code:
import sempy_labs as labs
from sempy_labs import directlake
import sempy_labs.report as rep
import time
import logging
import sys
import os
import requests
# all the variables and parameters are collected and passed correctly (ignore them)
# Helper functions
def get_workspace_id(workspace_name: str | None):
if not workspace_name:
return None
df = fabric.list_workspaces()
row = df.loc[df["Name"] == workspace_name, "Id"]
return row.iloc[0] if not row.empty else None
def get_item_id(workspace_id, item_name, item_type):
df = fabric.list_items(workspace=workspace_id)
row = df.loc[
(df["Type"] == item_type) &
(df["Display Name"] == item_name),
"Id"
]
return row.iloc[0] if not row.empty else None
def semantic_model_exists(workspace_id, dataset_name):
return get_item_id(workspace_id, dataset_name, "SemanticModel") is not None
def report_exists(workspace_id, report_name):
return get_item_id(workspace_id, report_name, "Report") is not None
def stage(name):
logger.info(f"=== {name} ===")
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(message)s",
handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger("deployment")
def main():
stage("Resolve Context")
target_workspace_id = workspaceId
lakehouse_id = lakehouseId
if not target_workspace_id:
raise ValueError(f"Workspace not found: {TARGET_WORKSPACE_NAME}")
if not lakehouse_id:
raise ValueError(f"Lakehouse not found: {LAKEHOUSE_NAME}")
stage("Check Existing Assets")
sm_exists = semantic_model_exists(
target_workspace_id,
TARGET_SEMANTICMODEL_NAME
)
rep_exists = report_exists(
target_workspace_id,
SOURCE_REPORT_NAME
)
logger.info(f"Semantic model exists: {sm_exists}")
logger.info(f"Report exists: {rep_exists}")
# --------------------------------------------------
# SHORT CIRCUIT
# --------------------------------------------------
if sm_exists and rep_exists:
logger.info("Nothing to deploy")
else:
if not sm_exists:
stage("Deploy Semantic Model")
labs.deploy_semantic_model(
source_dataset=SOURCE_DATASET_NAME,
source_workspace=SOURCE_WORKSPACE_NAME,
target_dataset=TARGET_SEMANTICMODEL_NAME,
target_workspace=TARGET_WORKSPACE_NAME,
refresh_target_dataset=False,
overwrite=True
)
directlake.update_direct_lake_model_connection(
dataset=TARGET_SEMANTICMODEL_NAME,
workspace=TARGET_WORKSPACE_NAME,
source=LAKEHOUSE_NAME,
source_type="Lakehouse",
source_workspace=TARGET_WORKSPACE_NAME,
use_sql_endpoint=False
)
# --------------------------------------------------
# CLONE REPORT
# --------------------------------------------------
if not rep_exists:
stage("Clone Report")
rep.clone_report(
report=SOURCE_REPORT_NAME,
cloned_report=SOURCE_REPORT_NAME,
workspace=SOURCE_WORKSPACE_NAME,
target_workspace=TARGET_WORKSPACE_NAME,
target_dataset=TARGET_SEMANTICMODEL_NAME,
target_dataset_workspace=TARGET_WORKSPACE_NAME
)
rep.report_rebind(
report=SOURCE_REPORT_NAME,
dataset=TARGET_SEMANTICMODEL_NAME,
report_workspace=TARGET_WORKSPACE_NAME,
dataset_workspace=TARGET_WORKSPACE_NAME
)
# --------------------------------------------------
# FINAL REFRESH
# --------------------------------------------------
stage("Refresh Semantic Model")
time.sleep(10)
try:
labs.refresh_semantic_model(
dataset=TARGET_SEMANTICMODEL_NAME,
workspace=TARGET_WORKSPACE_NAME
)
except Exception as e:
raise
logger.info("Deployment completed successfully")
Can someone help me how can I achieve the deployment by the pipeline execution, preferably, using SP. Because my pipeline triggers via Fabric API from an external application using SP.
