Python Microsoft Fabric - Running a pipeline and refreshing a report
After the nightly load, run the pipeline, wait for it, refresh the report's data and tell the team.
Once the nightly load has put the day's rows into the lakehouse, two more things have to happen before a report is current. A pipeline - a sequence of steps the data team built in Fabric - reshapes the raw rows into the tables the report reads, and then the report's data has to be refreshed, because a Power BI report shows what it loaded last, not what the tables hold now. This page chains both in one service that runs right after the nightly load.
The pipeline before the run
The pipeline's run history shows last night's run and nothing yet for tonight:
The service
A pipeline run and a report refresh are both jobs in Fabric, started with run_job and checked with get_job, so one helper waits for either. It cancels a job that has not ended within a set time.
# -*- coding: utf-8 -*-
# stdlib
from time import sleep
# Zato
from zato.common.typing_ import any_
from zato.server.service import Service
_running = 'NotStarted', 'InProgress'
_check_interval = 20
_pipeline_timeout = 1800
_refresh_timeout = 600
_timed_out = 'TimedOut'
class NightlyPipelineAndReport(Service):
name = 'nightly.pipeline-and-report'
def wait_for_job(self, conn:'any_', workspace_id:'str', item_id:'str', job_id:'str', timeout:'int') -> 'str':
""" Checks on a job until it ends or the timeout passes, then returns its final status.
"""
waited = 0
while waited < timeout:
job = conn.get_job(workspace_id, item_id, job_id)
status = job['status']
# The job has ended, one way or another ..
if status not in _running:
out = status
break
sleep(_check_interval)
waited += _check_interval
# .. out of time, stop the job.
else:
conn.cancel_job(workspace_id, item_id, job_id)
out = _timed_out
return out
def handle(self):
workspace_id = '11111111-2222-3333-4444-555555555555'
pipeline_id = 'bbbbbbbb-cccc-dddd-eeee-ffffffffffff'
dataset_id = 'cccccccc-dddd-eeee-ffff-000000000000'
conn = self.microsoft.fabric['My Fabric']
# Run the pipeline and wait for it, up to 30 minutes ..
pipeline_job_id = conn.run_job(workspace_id, pipeline_id, 'Pipeline')
pipeline_status = self.wait_for_job(conn, workspace_id, pipeline_id, pipeline_job_id, _pipeline_timeout)
if pipeline_status != 'Completed':
text = f'Nightly pipeline ended with {pipeline_status}'
self.microsoft.teams.send('Ops Teams', to='Data team/Nightly', text=text)
self.response.payload = {'pipeline': pipeline_status}
return
# .. then refresh the report's data and wait for that too, up to 10 minutes ..
refresh_job_id = conn.run_job(workspace_id, dataset_id, 'DefaultSemanticModelRefresh')
refresh_status = self.wait_for_job(conn, workspace_id, dataset_id, refresh_job_id, _refresh_timeout)
# .. and send the outcome to the team.
text = f'Nightly pipeline Completed, report refresh {refresh_status}'
self.microsoft.teams.send('Ops Teams', to='Data team/Nightly', text=text)
self.response.payload = {'pipeline': pipeline_status, 'refresh': refresh_status}
The message to the team goes through a Microsoft Teams connection created in the Dashboard beforehand.
Schedule the service under Scheduler → Config with the cron-style type and 30 2 * * *, half an hour after the nightly load, so the load has finished before the pipeline starts.
What Fabric shows afterwards
The pipeline's run history has tonight's run, submitted by the app registration:
And the dataset's refresh history shows the refresh that followed:
The report now shows what the tables hold after tonight's load.
See also
| Page | What it covers |
|---|---|
| Loading data into tables | The nightly load this service runs after |
| Picking up results when a notebook finishes | Waiting for a job from a scheduled service instead of inside one |
| Jobs | run_job, get_job and cancel_job, job types and statuses |