Issues with Fabric Connection REST API and On-Prem Connections

Hi everyone,

I’ve been working with the Fabric Connection REST API (POST https://api.fabric.microsoft.com/v1/connections) and have run into challenges when trying to use it with an on-premises connection. So far, I haven’t been able to get it working, and the documentation doesn’t provide much clarity on whether this scenario is fully supported or if there are specific steps required to make it work.

Has anyone successfully used the API with an on-prem connection? If so, I’d love to hear about your experience and any insights you can share. Additionally, I think the documentation for this API could be more detailed, especially around on-prem support, to help avoid confusion and unnecessary troubleshooting.

Any help or guidance would be greatly appreciated!

Thanks!

import requests
import json
from notebookutils import mssparkutils
# Function to create a connection and return the connection ID
def create_connection(display_name, server_name, database_name, username, password, gateway_id):
# Define headers
headers = {
"Authorization": "Bearer " + mssparkutils.credentials.getToken('pbi'),
"Content-Type": "application/json"
}

url = "https://api.fabric.microsoft.com/v1/connections"

payload_onprem_sql = {
"connectivityType": "OnPremisesGateway",
"gatewayId": gateway_id,
"displayName": display_name,
"connectionDetails": {
"type": "SQL",
"creationMethod": "SQL",
#"path": f"{server_name};{database_name}",
"parameters": [
{ "dataType": "Text",
"name": "server",
"value": server_name
},
{ "dataType": "Text",
"name": "database",
"value": database_name
}
]
},
"privacyLevel": "Organizational",
"credentialDetails": {
"singleSignOnType": "None",
"connectionEncryption": "Any",
"skipTestConnection": False,
"credentials": {
"credentialType": "Basic",
"username": username,
"password": password
},

}
}

print(f"\nPayload for creating connection (JSON format):\n{json.dumps(payload_onprem_sql, indent=4)}\n")

response = requests.post(url, headers=headers, json=payload_onprem_sql)

if response.status_code == 201:
print(f'Connection "{display_name}" was successfully created.')
return response.json().get('id')
else:
print(f"Failed to create connection. Status code: {response.status_code}")
print("Response:", response.text)
return None

print("\nTesting OnPrem SQL Connection:")
create_connection(
display_name="sqlconnonprem",
server_name=r"",
database_name="",
username="",
password="",
gateway_id="\"
)