msdev-kit

Microsoft developer toolkit for Python: Fabric/Power BI, MS Graph (Entra), and SharePoint.

Until 2026-06-17 this package was developed in a private repository. The git history was squashed when the project was opened to the public, so commits before that date are not preserved.

Supported OS

• Windows x64
• macOS Apple ARM
• Linux

Unsupported OS

• Currently, because of updates to cryptography package (>=50.0.0), Windows 32-bit and macOS Intel are not supported.

Installation

pip install msdev-kit

Or install from GitHub:

pip install git+https://github.com/Bernardo-Rufino/msdev-kit.git

For local development, use the repository instructions in Development.

Requirements: Python >= 3.10. Service principal authentication requires an Azure app registration, tenant ID, client ID, and client secret. Interactive user authentication is also supported for APIs that permit it.

Quick Start

from msdev_kit import Auth
from msdev_kit.fabric import Workspace
from msdev_kit.graph import GraphClient
from msdev_kit.sharepoint import SharePointClient

# authenticate
auth = Auth(tenant_id="...", client_id="...", client_secret="...")

# fabric: list workspaces
ws = Workspace(auth.get_token('pbi'))
workspaces = ws.list_workspaces_for_user()

# graph: look up a user
graph = GraphClient(auth)
user_id = graph.get_user_id('user@company.com')

# sharepoint: download a file
sp = SharePointClient(auth, sp_hostname='company', sp_site_path='sites/DataTeam')
sp.download_file('/Reports/monthly.xlsx', local_dir='./downloads')

Authentication

Auth owns token acquisition. Fabric and Power BI classes receive a token string, while Graph and SharePoint clients receive the Auth instance. You can use different service principals for different services by instantiating one Auth per service principal:

from msdev_kit import Auth

# service principal auth
fabric_auth = Auth(tenant_id="...", client_id="spn-a", client_secret="...")
graph_auth  = Auth(tenant_id="...", client_id="spn-b", client_secret="...")

Supported scopes

Interactive user auth

For scenarios requiring user context (e.g., RLS-enabled datasets), omit the client-secret arguments to use browser authentication by default:

auth = Auth()
token = auth.get_token('fabric')

You can also request a fresh user token from a service-principal Auth instance:

token = auth.get_token_for_user('pbi')     # opens browser for login
token = auth.get_token_for_user('fabric')

If the operating system cannot provide encrypted token-cache storage, opt in explicitly to an unencrypted cache. This can expose tokens to other users of the same machine, so use it only when secure storage is unavailable:

token = auth.get_token_for_user('fabric', allow_unencrypted_storage=True)

Credentials

For the bundled examples, copy the template and set the values:

cp .env.example .env

msdev-kit does not load .env automatically. The scripts in examples/ load .env from the repository root. Applications may pass values directly to Auth, or load environment variables with their own configuration mechanism.

Fabric & Power BI

from msdev_kit import Auth
from msdev_kit.fabric import Workspace, Dataset, Report, Dataflow, Pipeline

auth = Auth(tenant_id, client_id, client_secret)

Fabric classes take a token string — call auth.get_token('fabric') or auth.get_token('pbi') depending on the API.

• Workspace — workspaces, users, permissions
• Dataset — semantic models, DAX queries, permissions
• Report — metadata, definitions, visuals, measures
• Dataflow — Gen1, Gen2, Gen2 CI/CD management
• Pipeline — Data Pipeline management
• Other modules — Capacity, Admin, KQL, Notebook, Database

Workspace

Manage Power BI workspaces, users, and permissions.

ws = Workspace(auth.get_token('pbi'))
workspaces = ws.list_workspaces_for_user()
ws.add_user('user@company.com', workspace_id, 'Member', 'User')

add_user and update_user accept the official Power BI principal types: User, Group, and App. add_user updates an existing matching principal instead of issuing a duplicate add request.

Dataset

Manage datasets (semantic models), permissions, and execute DAX queries.

ds = Dataset(auth.get_token('pbi'))
result = ds.execute_query(workspace_id, dataset_id, "EVALUATE Sales")

Report

Retrieve report metadata, definitions, visuals, and report-level measures.

rpt = Report(auth.get_token('pbi'))
pages = rpt.list_report_pages(workspace_id, report_id)

Dataflow

Manage Power BI and Fabric dataflows, including Gen1, Gen2, and Gen2 CI/CD.

df = Dataflow(auth.get_token('pbi'))

# upgrade Gen1 to Gen2 CI/CD
result = df.upgrade_to_gen2_cicd(
    workspace_id='<workspace_id>',
    dataflow_id='<gen1_dataflow_id>',
    display_name='my_dataflow_cicd',
    source_type='gen1'
)

For a standard Gen2 source, the upgrade keeps each destination's workspace, item, table, and write method. It returns an error without creating a new item if it cannot verify destination preservation. A CI/CD source is copied from its definition without retargeting destination references. Gen1 uses its own internal storage and has no Gen2 data destination to reuse.

For service principal creation of a standard Gen2 dataflow, use a delegated Fabric user token for refresh. Fabric does not allow the service principal to run the Dataflow Gen2 CI/CD refresh job. The delegated token is used only for the refresh request and polling. Pass an SPN Power BI token separately for reading the standard source. Connection credentials are not copied. See the complete setup and runnable example before running this against an existing data destination.

spn_auth = Auth(tenant_id, client_id, client_secret)
user_auth = Auth()  # interactive user login
df = Dataflow(spn_auth.get_token('fabric'))
result = df.upgrade_to_gen2_cicd(
    workspace_id=workspace_id,
    dataflow_id=standard_gen2_dataflow_id,
    source_type='gen2',
    use_accessible_connections=True,
    refresh=True,
    refresh_access_token=user_auth.get_token('fabric'),
    pbi_access_token=spn_auth.get_token('pbi'),
)

Refresh is opt-in because it writes to the configured destination. A failed refresh returns its job details and the new item's ID for investigation.

Inventory workspace data destinations

from msdev_kit import Auth
from msdev_kit.fabric import Dataflow

workspace_id = "<workspace-id>"
auth = Auth(tenant_id="<tenant-id>", client_id="<client-id>", client_secret="<client-secret>")
dataflow = Dataflow(auth.get_token("pbi"))

result = dataflow.get_workspace_data_destinations(workspace_id, max_workers=4)

The method inspects every dataflow so result["content"] also records empty inspections and failures. Its workbook contains only table rows with a real data destination. Table name is exported as table_name, not table_table. A Fabric source without an API generation value is normalized to 2.1. The progress line is updated in place. If a 429 response occurs, the method prints the completed count, waits, then resumes the progress line. See examples/dataflow_destinations.py for a DataFrame normalizer and a runnable placeholder.

Pipeline

Manage Fabric Data Pipelines.

pipe = Pipeline(auth.get_token('fabric'))
activities = pipe.get_pipeline_activities(workspace_id, 'My Pipeline')
all_activities = pipe.get_pipeline_activities(
    workspace_id, 'My Pipeline', recursive=True
)

Recursive results are flat and include workspace_id on each activity. They include activities inside control-flow containers and follow both InvokePipeline and ExecutePipeline references. Each pipeline is scanned once, so cycles and repeated invocations do not duplicate downstream activities. An inaccessible child returns an error rather than a partial result.

To find pipelines for multiple dataflows, pass IDs, display names, or a mixture of both. The method lists the pipelines once and scans each pipeline definition once, concurrently up to max_workers:

matches = pipe.find_pipelines_by_dataflow(
    workspace_id,
    ['sales-dataflow-id', 'Finance Dataflow'],
    max_workers=5,
)

For a list input, each matching pipeline includes a dataflows list. Every entry identifies the matched dataflow and its matching RefreshDataflow activities. Single-dataflow calls keep the existing activities list output.

from msdev_kit import Auth
from msdev_kit.fabric import Dataflow, Pipeline

auth = Auth(tenant_id, client_id, client_secret)
dataflow = Dataflow(auth.get_token('pbi'))
pipeline = Pipeline(auth.get_token('fabric'))

workspace_id = '<workspace_id>'
old_dataflow_id = '<dataflow_id>'

# 1. replace dataflow destination (creates new CI/CD, deletes original)
result = dataflow.change_data_destination(
    workspace_id=workspace_id,
    dataflow_id=old_dataflow_id,
    destination_type='Warehouse',
    destination_workspace_id=workspace_id,
    destination_item_id='<warehouse_id>',
    mode='replace'
)
new_dataflow_id = result['content']['id']

# 2. find pipelines referencing the old dataflow ID
matches = pipeline.find_pipelines_by_dataflow(workspace_id, old_dataflow_id)

# 3. update each pipeline to use the new ID
for m in matches['content']:
    pipeline.replace_dataflow_id_in_pipeline(
        workspace_id, m['pipeline_id'], old_dataflow_id, new_dataflow_id
    )

Other modules

MS Graph (Entra)

Manage Entra ID (Azure AD) users and groups via the MS Graph API.

from msdev_kit import Auth
from msdev_kit.graph import GraphClient

auth = Auth(tenant_id, client_id, client_secret)
graph = GraphClient(auth)

# look up a user and add them to a group
user_id = graph.get_user_id('user@company.com')
group_id = graph.get_group_id('Data Team')
graph.add_group_member(group_id, user_id)

# list all members of a group
members = graph.list_group_members(group_id)

SharePoint

Manage SharePoint files and folders via MS Graph API (no ACS/Office365 dependency).

from msdev_kit import Auth
from msdev_kit.sharepoint import SharePointClient

auth = Auth(tenant_id, client_id, client_secret)
sp = SharePointClient(auth, sp_hostname='company', sp_site_path='sites/DataTeam')

# download a file
sp.download_file('/Reports/monthly.xlsx', local_dir='./downloads')

# upload a file (from path or bytes)
sp.upload_file('/Reports/updated.xlsx', source='./local/updated.xlsx')
sp.upload_file('/Reports/data.csv', source=csv_bytes, content_type='text/csv')

# create nested folders
sp.create_folder('/Reports/2026/Q1')

Hostname and site path inputs are normalized automatically:

Limitations

• Power BI and Fabric endpoints apply operation-specific throttling. Do not assume one global quota. get_workspace_data_destinations intentionally paces definition lookups at 200 requests per minute and backs off after HTTP 429.
• Not all users can be updated via the API. See Microsoft docs: Dataset permissions.
• Dataset query limits (executeQueries API):
  • Max 100,000 rows or 1,000,000 values (rows x columns) per query, whichever is hit first.
  • Max 15 MB of data per query.
  • 120 query requests per minute per user.
  • Only DAX queries are supported (no MDX, INFO functions, or DMV).
  • Datasets hosted in Azure Analysis Services or with a live connection to on-premises AAS are not supported.
  • Service Principals are not supported for datasets with RLS or SSO enabled.

Examples

Runnable, placeholder-based scripts live in examples/. Start with the examples guide, then run a read-only example from the repository root:

cp .env.example .env
python -m examples.workspaces
python -m examples.dataflow_destinations

Write examples never perform a mutation when run directly. Edit the placeholder values and call their explicit helper only after reviewing the target IDs.

Contributing

Read Contributing before opening a pull request. It links the local setup, validation commands, issue and pull request process, branch conventions, and release boundary.