IDMC Snowflake Connection names with usernames

 Today I was facing challenge on getting snowflake connection names and respective users from IDMC. I was using rest API to get the connection details.

I have started first getting the user session id using Get method with URL https://dm-us.informaticacloud.com/ma/api/v2/user/login

Once get the session id, I have passed it to get the connections(list).

"https://<region>.dm-us.informaticacloud.com/saas/api/v2/connection"

 If i am pulling the directly from the rest api it was not showing user. so I had to get list of connections and then use id of that connection and get the username . 

You can filter the script based on type of connection and the parameter name that you want from the connections.

I used below python script and in case you want to use it, you need to replace username, password and urls need to updated based on region/pod.

This script uses IDMC Rest API and  python and cosmetical script changes using chatgpt.

import requests

import json



# --- CONFIGURATION ---

BASE_URL = "<PODURL>"  # Replace with your POD

USERNAME = "<username>"

PASSWORD = "<password>"





# --- STEP 1: Authenticate and get session token ---

def get_session_token():

    url = "https://dm-us.informaticacloud.com/ma/api/v2/user/login"

    headers = {'Content-Type': 'application/json'}

    payload = {

        "username": USERNAME,

        "password": PASSWORD

    }



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

    #print(response)

    response.raise_for_status()

    #print(response.text)

    login = json.loads(response.text)



    for i in login:

        if i == "icSessionId":

            sid = login[i]

            print(sid)

    return sid





# --- STEP 2: Get all connections ---

def get_connections(session_id):

    url = f"https://<region>.dm-us.informaticacloud.com/saas/api/v2/connection"

    headers = {

        'icSessionId': session_id

    }



    response = requests.get(url, headers=headers)

    #print(session_id)

    response.raise_for_status()

    #return response.json()['items']

    #print(response.text)

    return response.json()





# --- STEP 3: Print connection names and usernames ---

def print_connections(connections,session_id):

    print(f"{'Connection Name':40} | {'id'} | {'username'}" )

    print("-" * 70)

    for conn in connections:

        if conn.get("instanceDisplayName") in ("Snowflake Data Cloud"):

            name = conn.get('name', 'N/A')

            id = conn.get('id', 'N/A')  # Not all connections have usernames

            #print(f"{name:40} | {id}")

            headers = {

                'icSessionId': session_id

            }

            response = requests.get(f"https://<region>.dm-us.informaticacloud.com/saas/api/v2/connection/{id}", headers=headers)

            #print(response)

            #print(response.text)

            response_json = response.json()

            username = response_json.get("connParams", {}).get("user")

            #print(username)



            print(f"{name:40} | {username}")

        if conn.get("instanceDisplayName") in ("Snowflake Cloud Data Warehouse"):

            name = conn.get('name', 'N/A')

            id = conn.get('id', 'N/A')  # Not all connections have usernames

            #print(f"{name:40} | {id}")

            headers = {

                'icSessionId': session_id

            }

            response = requests.get(f"https://<region>.dm-us.informaticacloud.com/saas/api/v2/connection/{id}", headers=headers)

            #print(response)

            #print(response.text)

            response_json = response.json()

            username = response_json.get("connParams", {}).get("Username")

            #print(username)



            print(f"{name:40} | {username}")







# --- MAIN ---

def main():

    try:

        session_token = get_session_token()

        #session_token= sessionid

        connections = get_connections(session_token)

        print_connections(connections, session_token)

    except Exception as e:

        print(f"Error: {e}")



if __name__ == "__main__":

    main()

Do let me know if you want to modify script to get any additional details or other connection type params. 

References:https://docs.informatica.com/cloud-common-services/administrator/current-version/rest-api-reference/data-integration-rest-api/connections.html