Getting Datacloud Schema

#!/usr/bin/env python3
"""
List all Data Lake Objects and retrieve schema for one DLO using REST API.
Uses SF CLI for authentication.
"""

import subprocess
import json
import sys
import requests


def authenticate_to_org(org_alias):
    """
    Authenticate to Salesforce org using SF CLI.

    Args:
        org_alias: SF CLI org alias (e.g., 'afvibe')

    Returns:
        Tuple of (instance_url, access_token, username)
    """
    print(f"🔐 Authenticating to Salesforce org '{org_alias}'...")

    try:
        result = subprocess.run(
            ['sf', 'org', 'display', '--target-org', org_alias, '--json'],
            capture_output=True,
            text=True,
            check=True
        )

        org_data = json.loads(result.stdout)

        if org_data.get('status') != 0:
            raise Exception(f"SF CLI returned error: {org_data}")

        org_info = org_data['result']

        if org_info.get('connectedStatus') != 'Connected':
            raise Exception(f"Org '{org_alias}' is not connected. Run: sf org login web --alias {org_alias}")

        instance_url = org_info['instanceUrl']
        access_token = org_info['accessToken']
        username = org_info.get('username', 'Unknown')

        print(f"✅ Authenticated as: {username}")
        print(f"📍 Instance: {instance_url}\n")

        return instance_url, access_token, username

    except subprocess.CalledProcessError as e:
        raise Exception(f"SF CLI command failed: {e.stderr}")
    except (json.JSONDecodeError, KeyError) as e:
        raise Exception(f"Failed to parse SF CLI output: {e}")


def list_all_dlos(instance_url, access_token, api_version='v64.0'):
    """
    List all Data Lake Objects using SSOT REST API.

    Args:
        instance_url: Salesforce instance URL
        access_token: OAuth access token
        api_version: API version (default: v64.0)

    Returns:
        List of DLO dictionaries
    """
    url = f"{instance_url}/services/data/{api_version}/ssot/data-lake-objects"

    headers = {
        'Authorization': f'Bearer {access_token}',
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    }

    print("📋 Fetching all Data Lake Objects...")
    response = requests.get(url, headers=headers)

    if response.status_code != 200:
        raise Exception(f"API Error: HTTP {response.status_code}\n{response.text[:500]}")

    response_data = response.json()

    # Extract DLO list from paginated response
    if isinstance(response_data, dict) and 'dataLakeObjects' in response_data:
        dlos = response_data['dataLakeObjects']
        total_size = response_data.get('totalSize', len(dlos))
        print(f"✅ Found {len(dlos)} DLOs (Total: {total_size})\n")
    else:
        # Fallback if response format is different
        dlos = response_data if isinstance(response_data, list) else []
        print(f"✅ Found {len(dlos)} DLOs\n")

    return dlos


def get_dlo_schema(instance_url, access_token, dlo_name, api_version='v64.0'):
    """
    Get detailed schema for a specific DLO.

    Args:
        instance_url: Salesforce instance URL
        access_token: OAuth access token
        dlo_name: DLO developer name (e.g., 'Employee__dll')
        api_version: API version (default: v64.0)

    Returns:
        DLO detail dictionary with full schema
    """
    url = f"{instance_url}/services/data/{api_version}/ssot/data-lake-objects/{dlo_name}"

    headers = {
        'Authorization': f'Bearer {access_token}',
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    }

    print(f"🔍 Fetching schema for DLO: {dlo_name}...")
    response = requests.get(url, headers=headers)

    if response.status_code != 200:
        raise Exception(f"API Error: HTTP {response.status_code}\n{response.text[:500]}")

    response_data = response.json()

    # Extract DLO from paginated response
    if isinstance(response_data, dict) and 'dataLakeObjects' in response_data:
        dlos = response_data['dataLakeObjects']
        if dlos:
            return dlos[0]  # Return first (should be only) DLO
        else:
            raise Exception(f"DLO '{dlo_name}' not found")
    else:
        # Fallback if response format is different
        return response_data


def display_dlo_list(dlos):
    """Display summary of all DLOs."""
    print("=" * 80)
    print("📊 DATA LAKE OBJECTS")
    print("=" * 80)

    for idx, dlo in enumerate(dlos, 1):
        print(f"\n{idx}. {dlo.get('name', 'Unknown')}")
        print(f"   Label: {dlo.get('label', 'N/A')}")
        print(f"   Category: {dlo.get('category', 'N/A')}")
        if 'id' in dlo:
            print(f"   ID: {dlo['id']}")


def display_dlo_schema(dlo_detail):
    """Display detailed schema information for a DLO."""
    print("\n" + "=" * 80)
    print(f"🔍 SCHEMA DETAILS FOR: {dlo_detail.get('name')}")
    print("=" * 80)

    print(f"\n📝 Basic Information:")
    print(f"   Name: {dlo_detail.get('name')}")
    print(f"   Label: {dlo_detail.get('label')}")
    print(f"   Category: {dlo_detail.get('category')}")
    print(f"   Description: {dlo_detail.get('description', 'N/A')}")

    if 'dataspaceInfo' in dlo_detail:
        dataspaces = dlo_detail['dataspaceInfo']
        dataspace_names = [ds.get('name', 'Unknown') for ds in dataspaces]
        print(f"   Dataspaces: {', '.join(dataspace_names)}")

    # Display field schema
    fields = dlo_detail.get('fields', [])

    if fields:
        print(f"\n🔧 Fields ({len(fields)} total):")
        print("-" * 80)

        # Show all fields with detailed info
        for field in fields:
            print(f"\n   • {field.get('name')}")
            print(f"     Label: {field.get('label', 'N/A')}")
            print(f"     Data Type: {field.get('dataType', 'Unknown')}")
            print(f"     Primary Key: {field.get('isPrimaryKey', False)}")
            print(f"     Nullable: {field.get('isNullable', True)}")

            if 'length' in field:
                print(f"     Length: {field['length']}")
            if 'precision' in field:
                print(f"     Precision: {field['precision']}")
            if 'scale' in field:
                print(f"     Scale: {field['scale']}")
    else:
        print("\n   ⚠️  No fields found in schema")

    # Show full JSON (optional, can be commented out)
    print("\n" + "=" * 80)
    print("📄 FULL SCHEMA (JSON):")
    print("=" * 80)
    print(json.dumps(dlo_detail, indent=2))


def main():
    """Main execution function."""
    if len(sys.argv) < 2:
        print("Usage: python list_dlos_and_schema.py <org_alias> [dlo_name]")
        print("\nExamples:")
        print("  python list_dlos_and_schema.py afvibe")
        print("  python list_dlos_and_schema.py afvibe Employee__dll")
        sys.exit(1)

    org_alias = sys.argv[1]
    specific_dlo = sys.argv[2] if len(sys.argv) > 2 else None

    try:
        # Step 1: Authenticate
        instance_url, access_token, username = authenticate_to_org(org_alias)

        # Step 2: List all DLOs
        dlos = list_all_dlos(instance_url, access_token)
        display_dlo_list(dlos)

        # Step 3: Get schema for a specific DLO
        if specific_dlo:
            # User specified a DLO name
            dlo_detail = get_dlo_schema(instance_url, access_token, specific_dlo)
            display_dlo_schema(dlo_detail)
        elif dlos:
            # Get schema for the first DLO
            first_dlo = dlos[0]
            dlo_name = first_dlo.get('name')
            dlo_detail = get_dlo_schema(instance_url, access_token, dlo_name)
            display_dlo_schema(dlo_detail)
        else:
            print("\n⚠️  No DLOs found in this org")

        print("\n✅ Done!")

    except Exception as e:
        print(f"\n❌ Error: {e}")
        sys.exit(1)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
List all Data Model Objects and retrieve schema for one DMO using REST API.
Uses SF CLI for authentication.
"""

import subprocess
import json
import sys
import requests


def authenticate_to_org(org_alias):
    """
    Authenticate to Salesforce org using SF CLI.

    Args:
        org_alias: SF CLI org alias (e.g., 'afvibe')

    Returns:
        Tuple of (instance_url, access_token, username)
    """
    print(f"🔐 Authenticating to Salesforce org '{org_alias}'...")

    try:
        result = subprocess.run(
            ['sf', 'org', 'display', '--target-org', org_alias, '--json'],
            capture_output=True,
            text=True,
            check=True
        )

        org_data = json.loads(result.stdout)

        if org_data.get('status') != 0:
            raise Exception(f"SF CLI returned error: {org_data}")

        org_info = org_data['result']

        if org_info.get('connectedStatus') != 'Connected':
            raise Exception(f"Org '{org_alias}' is not connected. Run: sf org login web --alias {org_alias}")

        instance_url = org_info['instanceUrl']
        access_token = org_info['accessToken']
        username = org_info.get('username', 'Unknown')

        print(f"✅ Authenticated as: {username}")
        print(f"📍 Instance: {instance_url}\n")

        return instance_url, access_token, username

    except subprocess.CalledProcessError as e:
        raise Exception(f"SF CLI command failed: {e.stderr}")
    except (json.JSONDecodeError, KeyError) as e:
        raise Exception(f"Failed to parse SF CLI output: {e}")


def list_all_dmos(instance_url, access_token, api_version='v64.0'):
    """
    List all Data Model Objects using SSOT REST API.

    Args:
        instance_url: Salesforce instance URL
        access_token: OAuth access token
        api_version: API version (default: v64.0)

    Returns:
        List of DMO dictionaries
    """
    url = f"{instance_url}/services/data/{api_version}/ssot/data-model-objects"

    headers = {
        'Authorization': f'Bearer {access_token}',
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    }

    print("📋 Fetching all Data Model Objects...")
    response = requests.get(url, headers=headers)

    if response.status_code != 200:
        raise Exception(f"API Error: HTTP {response.status_code}\n{response.text[:500]}")

    response_data = response.json()

    # Extract DMO list from paginated response
    if isinstance(response_data, dict) and 'dataModelObject' in response_data:
        dmos = response_data['dataModelObject']
        total_size = response_data.get('totalSize', len(dmos))
        print(f"✅ Found {len(dmos)} DMOs (Total: {total_size})\n")
    else:
        # Fallback if response format is different
        dmos = response_data if isinstance(response_data, list) else []
        print(f"✅ Found {len(dmos)} DMOs\n")

    return dmos


def get_dmo_schema(instance_url, access_token, dmo_name, api_version='v64.0'):
    """
    Get detailed schema for a specific DMO.

    Args:
        instance_url: Salesforce instance URL
        access_token: OAuth access token
        dmo_name: DMO developer name (e.g., 'Individual__dlm')
        api_version: API version (default: v64.0)

    Returns:
        DMO detail dictionary with full schema
    """
    url = f"{instance_url}/services/data/{api_version}/ssot/data-model-objects/{dmo_name}"

    headers = {
        'Authorization': f'Bearer {access_token}',
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    }

    print(f"🔍 Fetching schema for DMO: {dmo_name}...")
    response = requests.get(url, headers=headers)

    if response.status_code != 200:
        raise Exception(f"API Error: HTTP {response.status_code}\n{response.text[:500]}")

    response_data = response.json()

    # Single DMO endpoint returns the object directly (not wrapped in an array)
    return response_data


def display_dmo_list(dmos):
    """Display summary of all DMOs."""
    print("=" * 80)
    print("📊 DATA MODEL OBJECTS")
    print("=" * 80)

    for idx, dmo in enumerate(dmos, 1):
        print(f"\n{idx}. {dmo.get('name', 'Unknown')}")
        print(f"   Label: {dmo.get('label', 'N/A')}")
        print(f"   Category: {dmo.get('category', 'N/A')}")
        print(f"   Creation Type: {dmo.get('creationType', 'N/A')}")
        print(f"   Data Space: {dmo.get('dataSpaceName', 'N/A')}")


def display_dmo_schema(dmo_detail):
    """Display detailed schema information for a DMO."""
    print("\n" + "=" * 80)
    print(f"🔍 SCHEMA DETAILS FOR: {dmo_detail.get('name')}")
    print("=" * 80)

    print(f"\n📝 Basic Information:")
    print(f"   Name: {dmo_detail.get('name')}")
    print(f"   Label: {dmo_detail.get('label')}")
    print(f"   Category: {dmo_detail.get('category')}")
    print(f"   Creation Type: {dmo_detail.get('creationType', 'N/A')}")
    print(f"   Data Space: {dmo_detail.get('dataSpaceName', 'N/A')}")

    # Display field schema
    fields = dmo_detail.get('fields', [])

    if fields:
        print(f"\n🔧 Fields ({len(fields)} total):")
        print("-" * 80)

        # Show all fields with detailed info
        for field in fields:
            print(f"\n   • {field.get('name')}")
            print(f"     Label: {field.get('label', 'N/A')}")
            print(f"     Data Type: {field.get('type', 'Unknown')}")
            print(f"     Primary Key: {field.get('isPrimaryKey', False)}")
            print(f"     Creation Type: {field.get('creationType', 'N/A')}")
            print(f"     Usage Tag: {field.get('usageTag', 'N/A')}")

            if 'length' in field:
                print(f"     Length: {field['length']}")
            if 'precision' in field:
                print(f"     Precision: {field['precision']}")
            if 'scale' in field:
                print(f"     Scale: {field['scale']}")
    else:
        print("\n   ⚠️  No fields found in schema")

    # Show full JSON
    print("\n" + "=" * 80)
    print("📄 FULL SCHEMA (JSON):")
    print("=" * 80)
    print(json.dumps(dmo_detail, indent=2))


def main():
    """Main execution function."""
    if len(sys.argv) < 2:
        print("Usage: python get_dmo_schema.py <org_alias> [dmo_name]")
        print("\nExamples:")
        print("  python get_dmo_schema.py afvibe")
        print("  python get_dmo_schema.py afvibe Individual__dlm")
        sys.exit(1)

    org_alias = sys.argv[1]
    specific_dmo = sys.argv[2] if len(sys.argv) > 2 else None

    try:
        # Step 1: Authenticate
        instance_url, access_token, username = authenticate_to_org(org_alias)

        # Step 2: List all DMOs
        dmos = list_all_dmos(instance_url, access_token)
        display_dmo_list(dmos)

        # Step 3: Get schema for a specific DMO
        if specific_dmo:
            # User specified a DMO name
            dmo_detail = get_dmo_schema(instance_url, access_token, specific_dmo)
            display_dmo_schema(dmo_detail)
        elif dmos:
            # Get schema for the first DMO
            first_dmo = dmos[0]
            dmo_name = first_dmo.get('name')
            dmo_detail = get_dmo_schema(instance_url, access_token, dmo_name)
            display_dmo_schema(dmo_detail)
        else:
            print("\n⚠️  No DMOs found in this org")

        print("\n✅ Done!")

    except Exception as e:
        print(f"\n❌ Error: {e}")
        sys.exit(1)


if __name__ == '__main__':
    main()