Python Microsoft Fabric API - Connection

A Fabric connection is created once in the Dashboard and used in services as self.microsoft.fabric[name]. This page is about the connection itself - what goes into the form, what it does with the credentials and the two methods that are not about one kind of item - invoke and ping.

Dashboard fields

Under Cloud → Microsoft Fabric:

What the app registration needs

An admin does three things once:

1. Enables the tenant setting "Service principals can use Fabric APIs" under Tenant settings → Developer settings in the Fabric admin portal
2. Adds the app registration to each workspace the connection should reach, as a Contributor for writing or a Viewer for reading only
3. Enables the Livy endpoint tenant setting on the same page, for query and the other Spark methods

Tokens

The connection signs in with the client credentials grant - it posts the client ID and secret to https://login.microsoftonline.com/<tenant ID>/oauth2/v2.0/token and gets a token back. It gets two tokens, because the Fabric API and OneLake storage are separate services:

Each token is obtained the first time it is needed, kept, and replaced a minute before it would expire. When Fabric rejects a token anyway - for instance, the secret was rotated - the connection gets a new one and repeats the request once. Services never see a token.

invoke

conn.invoke(method, path, params=None, data=None)

Calls any endpoint of the Fabric API with the connection's token, for the parts of the API that have no method of their own.

Returns the parsed JSON response, or nothing when the endpoint returned no body. A status code outside the success range raises an exception with the code and the response text.

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class Invoke(Service):

    def handle(self):

        workspace_id = '11111111-2222-3333-4444-555555555555'

        conn = self.microsoft.fabric['My Fabric']

        # Every item of one type in a workspace, through the API's own filter.
        path = f'/workspaces/{workspace_id}/items'
        params = {'type': 'Eventstream'}

        response = conn.invoke('GET', path, params=params)

        names = []
        for item in response['value']:
            names.append(item['displayName'])

        self.response.payload = {'eventstreams': names}

{"eventstreams": ["events"]}

conn.get(path, params), conn.post(path, data, params), conn.patch(path, data, params) and conn.delete(path, params) are the same call with the method filled in. conn.invoke_raw takes the same parameters as invoke and returns the whole response object instead, for the headers.

ping

conn.ping()

Confirms the connection works by obtaining a token and listing the workspaces the app registration can see. The Ping button in the Dashboard calls it, and a service can too. Returns nothing, raises an exception when the token or the listing fails, and logs how many workspaces it found.

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class Ping(Service):

    def handle(self):

        conn = self.microsoft.fabric['My Fabric']
        conn.ping()

        self.response.payload = {'connection': conn.name}

See also

Learn more