Authentication not granted for service principal token in MS Fabric API using Python

I am trying to connect to OneLake API in Microsoft Fabric using Python in VScode.

So far I have

1. Registered an app in Azure with these API permissions

2. Then created a secret for my service principal
3. Then I try to get the token with this function, using azure.identity:

from azure.identity import ClientSecretCredential, AuthenticationRequiredError

def get_access_token(app_id, client_secret, directory_id):
    try:
        # Create the ClientSecretCredential using the provided credentials
        credential = ClientSecretCredential(
            client_id=app_id,
            client_secret=client_secret,
            tenant_id=directory_id
            #scope="https://storage.azure.com/.default"
        )

        # Use the credential to get the access token
        token = credential.get_token("https://storage.azure.com/.default").token

        return token, credential

    except AuthenticationRequiredError as e:
        print("Authentication failed. Please check your credentials.")
        raise e

    except Exception as e:
        print("An error occurred while getting the access token:")
        print(str(e))
        raise e
    
access_token, credential = get_access_token(app_id, client_secret, directory_id)

It seems I get the token fine and all. But there is something wrong with the permissions or scope or access. Because when I run this function to check for connection i get status code 400

def check_connection_with_onelake(access_token):
    base_url = "https://onelake.dfs.fabric.microsoft.com/9c3ffd43-b537-4ca2-b9ba-0c59d0094033/Files/sample?resource=file" 
    token_headers = {
        "Authorization": "Bearer " + access_token
    }

    try:
        response = requests.put(base_url, headers=token_headers)

        if response.status_code == 200:
            print("Connection with OneLake is successful.")
        else:
            print("Failed to connect with OneLake. Status code:", response.status_code)

    except requests.exceptions.RequestException as e:
        print("An error occurred while checking the connection:", str(e))

# Assuming 'access_token' is already defined and contains a valid access token
check_connection_with_onelake(access_token)

4. I also added the app's service principal to users in the Fabric workspace as an admin

Where am I missing access and how do I grant the correct access?

references: https://learn.microsoft.com/en-us/fabric/onelake/onelake-access-api https://amitchandak.medium.com/on-premise-python-code-to-local-sql-server-data-to-microsoft-fabric-lakehouse-using-token-d15b8795e349

I registered one Azure AD application and granted API permissions as below:

In my Fabric Workspace, I added above service principal as Admin same as you:

When I ran your python code in my environment, I too got status code 400. To know the exact error, I printed content of response like this:

import requests
from azure.identity import ClientSecretCredential, AuthenticationRequiredError

def get_access_token(app_id, client_secret, directory_id):
    try:
        # Create the ClientSecretCredential using the provided credentials
        credential = ClientSecretCredential(
            client_id=app_id,
            client_secret=client_secret,
            tenant_id=directory_id
            #scope="https://storage.azure.com/.default"
        )

        # Use the credential to get the access token
        token = credential.get_token("https://storage.azure.com/.default").token

        return token, credential

    except AuthenticationRequiredError as e:
        print("Authentication failed. Please check your credentials.")
        raise e

    except Exception as e:
        print("An error occurred while getting the access token:")
        print(str(e))
        raise e
    
access_token, credential = get_access_token("appId", "secret", "tenantId")


def check_connection_with_onelake(access_token):
    base_url = "https://onelake.dfs.fabric.microsoft.com/0f0050d1-5601-47d0-9eb6-xxxxxxx/Files/sample?resource=file" 
    token_headers = {
        "Authorization": "Bearer " + access_token
    }

    try:
        response = requests.put(base_url, headers=token_headers)

        if response.status_code == 200:
            print("Connection with OneLake is successful.")
        else:
            print("Failed to connect with OneLake. Status code:", response.status_code)
            print(response.content)

    except requests.exceptions.RequestException as e:
        print("An error occurred while checking the connection:", str(e))

# Assuming 'access_token' is already defined and contains a valid access token
check_connection_with_onelake(access_token)

Response:

The error occurred as you missed adding LakehouseId after workspace ID in your base URL and you need to use GET request to check connection by removing resource=file.

You can find LakehouseId in address bar after /lakehouses/ when you open it in browser like this:

When I ran below modified code by changing the base_url and request type, I got response successfully like below:

import requests
from azure.identity import ClientSecretCredential, AuthenticationRequiredError

def get_access_token(app_id, client_secret, directory_id):
    try:
        # Create the ClientSecretCredential using the provided credentials
        credential = ClientSecretCredential(
            client_id=app_id,
            client_secret=client_secret,
            tenant_id=directory_id
            #scope="https://storage.azure.com/.default"
        )

        # Use the credential to get the access token
        token = credential.get_token("https://storage.azure.com/.default").token

        return token, credential

    except AuthenticationRequiredError as e:
        print("Authentication failed. Please check your credentials.")
        raise e

    except Exception as e:
        print("An error occurred while getting the access token:")
        print(str(e))
        raise e
    
access_token, credential = get_access_token("appId", "secret", "tenantId")


def check_connection_with_onelake(access_token):
    base_url = "https://onelake.dfs.fabric.microsoft.com/0f0050d1-5601-47d0-9eb6-xxxxxxx/37a6ca4e-27b6-437b-800c-xxxxxxxxxx/Files/sample" 
    token_headers = {
        "Authorization": "Bearer " + access_token
    }

    try:
        response = requests.get(base_url, headers=token_headers)

        if response.status_code == 200:
            print("Connection with OneLake is successful.")
        else:
            print("Failed to connect with OneLake. Status code:", response.status_code)
            print(response.content)

    except requests.exceptions.RequestException as e:
        print("An error occurred while checking the connection:", str(e))

# Assuming 'access_token' is already defined and contains a valid access token
check_connection_with_onelake(access_token)

Response: