Examples

This section contains practical examples of using the Amorphic SDK with the enhanced wrapper classes.

Getting Started with Wrapper Classes

Basic Setup

The standard client setup fetches all configuration from AWS SSM Parameter Store. This pattern should be used in all operational and automated scripts.

import certifi
import boto3
from openapi_client.amorphic_api_client import AmorphicApiClient

def get_amorphic_client():
    ssm = boto3.client('ssm')
    api_gw_url = ssm.get_parameter(Name='adp/amorphic/config/apigwurl', WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString
    role_id    = ssm.get_parameter(Name='adp/amorphic/config/roleid',   WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString
    client = AmorphicApiClient.create_with_auth(
        host=api_gw_url,
        role_id=role_id,
        ssl_ca_cert=certifi.where(),
        ssm_parameter='adp/amorphic/config/pattoken'
    )
    return client, role_id

Dataset Management Examples

Creating a Dataset

import certifi
import boto3
from openapi_client.amorphic_api_client import AmorphicApiClient
from openapi_client.api.datasets_api import DatasetsApi
from openapi_client.models.dataset_metadata import DatasetMetadata

def create_dataset_example():
    ssm = boto3.client('ssm')
    api_gw_url = ssm.get_parameter(Name='adp/amorphic/config/apigwurl', WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString
    role_id    = ssm.get_parameter(Name='adp/amorphic/config/roleid',   WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString

    client = AmorphicApiClient.create_with_auth(
        host=api_gw_url,
        role_id=role_id,
        ssl_ca_cert=certifi.where(),
        ssm_parameter='adp/amorphic/config/pattoken'
    )

    datasets_api = DatasetsApi(client)

    dataset_metadata = DatasetMetadata(
        dataset_name="sales_data_q1_2024",
        dataset_description="Quarterly sales data for analysis",
        dataset_type="internal",
        domain="sales",
        keywords=["sales", "quarterly", "analytics"],
        file_type="csv",
        table_update="append",
        data_classification=["INTERNAL"],
        file_delimiter=",",
        enable_data_validation=True,
        is_data_processing_enabled=True,
        skip_file_header=False,
        skip_lz_process=False,
        enable_data_cleanup=True,
        datasource_type="api",
        TargetLocation="lf"
    )

    try:
        result = datasets_api.create_dataset(
            role_id=role_id,
            dataset_metadata=dataset_metadata
        )
        return result
    except Exception as e:
        print(f"Failed to create dataset: {e}")
        raise

if __name__ == "__main__":
    create_dataset_example()

Listing Datasets

def list_datasets_example():
    ssm = boto3.client('ssm')
    api_gw_url = ssm.get_parameter(Name='adp/amorphic/config/apigwurl', WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString
    role_id    = ssm.get_parameter(Name='adp/amorphic/config/roleid',   WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString

    client = AmorphicApiClient.create_with_auth(
        host=api_gw_url,
        role_id=role_id,
        ssl_ca_cert=certifi.where(),
        ssm_parameter='adp/amorphic/config/pattoken'
    )

    datasets_api = DatasetsApi(client)

    try:
        datasets = datasets_api.list_datasets(role_id=role_id)
        print(f"Found {len(datasets.datasets)} datasets")
        return datasets
    except Exception as e:
        print(f"Failed to list datasets: {e}")
        raise

Bulk Operations Example

from openapi_client.models.batch_delete_datasets import BatchDeleteDatasets

def bulk_delete_example():
    ssm = boto3.client('ssm')
    api_gw_url = ssm.get_parameter(Name='adp/amorphic/config/apigwurl', WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString
    role_id    = ssm.get_parameter(Name='adp/amorphic/config/roleid',   WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString

    client = AmorphicApiClient.create_with_auth(
        host=api_gw_url,
        role_id=role_id,
        ssl_ca_cert=certifi.where(),
        ssm_parameter='adp/amorphic/config/pattoken'
    )

    datasets_api = DatasetsApi(client)

    dataset_ids = BatchDeleteDatasets(
        dataset_ids=["dataset-1", "dataset-2", "dataset-3"]
    )

    try:
        result = datasets_api.bulk_delete_datasets(
            role_id=role_id,
            batch_delete_datasets=dataset_ids
        )
        return result
    except Exception as e:
        print(f"Failed to bulk delete datasets: {e}")
        raise

Pagination with Wrapper Classes

def get_all_datasets_paginated():
    """Get all datasets using pagination with wrapper classes"""
    ssm = boto3.client('ssm')
    api_gw_url = ssm.get_parameter(Name='adp/amorphic/config/apigwurl', WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString
    role_id    = ssm.get_parameter(Name='adp/amorphic/config/roleid',   WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString

    client = AmorphicApiClient.create_with_auth(
        host=api_gw_url,
        role_id=role_id,
        ssl_ca_cert=certifi.where(),
        ssm_parameter='adp/amorphic/config/pattoken'
    )

    datasets_api = DatasetsApi(client)
    all_datasets = []
    page_size = 50
    offset = 0

    while True:
        result = datasets_api.list_datasets(
            role_id=role_id,
            limit=str(page_size),
            offset=str(offset),
        )

        page = result.datasets or []
        all_datasets.extend(page)

        if len(page) < page_size:
            break

        offset += page_size

    return all_datasets

User Management Examples

Creating Roles

from openapi_client.api.users_api import UsersApi
from openapi_client.models.role_input import RoleInput

def create_role_example():
    ssm = boto3.client('ssm')
    api_gw_url = ssm.get_parameter(Name='adp/amorphic/config/apigwurl', WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString
    role_id    = ssm.get_parameter(Name='adp/amorphic/config/roleid',   WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString

    client = AmorphicApiClient.create_with_auth(
        host=api_gw_url,
        role_id=role_id,
        ssl_ca_cert=certifi.where(),
        ssm_parameter='adp/amorphic/config/pattoken'
    )

    users_api = UsersApi(client)

    role_input = RoleInput(
        role_name="data_analyst",
        description="Role for data analysts",
        permissions=["datasets:read", "insights:read"]
    )

    result = users_api.create_role(
        role_id=role_id,
        role_input=role_input
    )
    return result

Management Examples

System Information

def get_system_info_example():
    ssm = boto3.client('ssm')
    api_gw_url = ssm.get_parameter(Name='adp/amorphic/config/apigwurl', WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString
    role_id    = ssm.get_parameter(Name='adp/amorphic/config/roleid',   WithDecryption=False)['Parameter']['Value']  # String type; use WithDecryption=True for SecureString

    client = AmorphicApiClient.create_with_auth(
        host=api_gw_url,
        role_id=role_id,
        ssl_ca_cert=certifi.where(),
        ssm_parameter='adp/amorphic/config/pattoken'
    )

    system_info = client.get_system_information(role_id=role_id)
    print(f"Platform Version: {system_info.version}")
    print(f"Environment: {system_info.environment}")
    print(f"AWS Region: {system_info.aws_region}")
    return system_info

Discovery and Help Examples

Using the Help System

from openapi_client.help import HelpCommand

def explore_apis():
    # Create help instance
    help_cmd = HelpCommand()

    # List all available APIs
    print("=== Available APIs ===")
    help_cmd.list_apis()

    # Get details about datasets API
    print("\\n=== Datasets API Details ===")
    help_cmd.show_api_details('DatasetsApi')

    # Describe a specific method
    print("\\n=== Method Details ===")
    help_cmd.describe_api_method('DatasetsApi', 'create_dataset')

    # Search for dataset-related functionality
    print("\\n=== Search Results ===")
    help_cmd.search_apis('dataset')

# Command-line usage (alternative)
# python -m openapi_client.help list
# python -m openapi_client.help api DatasetsApi
# python -m openapi_client.help describe DatasetsApi create_dataset
# python -m openapi_client.help search dataset