Examples

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

Getting Started with Wrapper Classes

Basic Setup

import os
import certifi
from openapi_client.amorphic_api_client import AmorphicApiClient

# Create authenticated client (recommended pattern)
def get_amorphic_client():
    return AmorphicApiClient.create_with_auth(
        host=os.getenv('API_GW_URL'),
        role_id=os.getenv('ROLE_ID'),
        ssl_ca_cert=certifi.where(),
        debug=False,
        aws_profile=None,
        env_var='PAT_TOKEN'
    )

Dataset Management Examples

Creating a Dataset

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

# Set up environment variables
ROLE_ID = os.getenv('ROLE_ID')
API_GW_URL = os.getenv('API_GW_URL')

def create_dataset_example():
    # Create authenticated client
    client = AmorphicApiClient.create_with_auth(
    host=API_GW_URL,
    role_id=ROLE_ID,
    ssl_ca_cert=certifi.where(),
    debug=False,
    aws_profile=None,
    env_var='PAT_TOKEN'

)

    # Create datasets API instance
    datasets_api = DatasetsApi(client)

    # Define dataset metadata
    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"
    )

    # Create the dataset
    try:
        result = datasets_api.create_dataset(
            role_id=ROLE_ID,
            dataset_metadata=dataset_metadata
        )
        print(f"Dataset created successfully: {result}")
        return result
    except Exception as e:
        print(f"Failed to create dataset: {e}")
        raise

if __name__ == "__main__":
    create_dataset_example()

Listing and Searching Datasets

def list_datasets_example():
    # Create authenticated client
    client = AmorphicApiClient.create_with_auth(
    host=os.getenv('API_GW_URL'),
    role_id=os.getenv('ROLE_ID'),
    ssl_ca_cert=certifi.where(),
    debug=False,
    aws_profile=None,
    env_var='PAT_TOKEN'

)

    datasets_api = DatasetsApi(client)

    try:
        # List all datasets
        datasets = datasets_api.list_datasets(role_id=os.getenv('ROLE_ID'))
        print(f"Found {len(datasets)} datasets")

        # Search datasets with filters
        search_results = datasets_api.search_datasets(
            role_id=os.getenv('ROLE_ID')
        )
        print(f"Search returned {len(search_results)} results")

        return datasets
    except Exception as e:
        print(f"Error retrieving datasets: {e}")
        raise

Bulk Operations Example

from openapi_client.models.batch_delete_datasets import BatchDeleteDatasets

def bulk_delete_example():
    client = AmorphicApiClient.create_with_auth(
    host=os.getenv('API_GW_URL'),
    role_id=os.getenv('ROLE_ID'),
    ssl_ca_cert=certifi.where(),
    debug=False,
    aws_profile=None,
    env_var='PAT_TOKEN'

)

    datasets_api = DatasetsApi(client)

    # Define datasets to delete
    dataset_ids = BatchDeleteDatasets(
        dataset_ids=["dataset-1", "dataset-2", "dataset-3"]
    )

    try:
        result = datasets_api.bulk_delete_datasets(
            role_id=os.getenv('ROLE_ID'),
            batch_delete_datasets=dataset_ids
        )
        print(f"Bulk delete completed: {result}")
        return result
    except Exception as e:
        print(f"Bulk delete failed: {e}")
        raise

Pagination with Wrapper Classes

def get_all_datasets_paginated():
    """Get all datasets using pagination with wrapper classes"""
    client = AmorphicApiClient.create_with_auth(
    host=os.getenv('API_GW_URL'),
    role_id=os.getenv('ROLE_ID'),
    ssl_ca_cert=certifi.where(),
    debug=False,
    aws_profile=None,
    env_var='PAT_TOKEN'

)

    datasets_api = DatasetsApi(client)
    all_datasets = []
    page = 1
    page_size = 50

    while True:
        try:
            # Note: Actual pagination parameters depend on API implementation
            datasets = datasets_api.search_datasets(
                role_id=os.getenv('ROLE_ID')
                # Add pagination parameters as supported by the API
            )

            if not datasets or len(datasets) == 0:
                break

            all_datasets.extend(datasets)

            if len(datasets) < page_size:
                # Last page reached
                break

            page += 1

        except Exception as e:
            print(f"Error fetching page {page}: {e}")
            break

    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():
    client = AmorphicApiClient.create_with_auth(
    host=os.getenv('API_GW_URL'),
    role_id=os.getenv('ROLE_ID'),
    ssl_ca_cert=certifi.where(),
    debug=False,
    aws_profile=None,
    env_var='PAT_TOKEN'

)

    users_api = UsersApi(client)

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

    try:
        result = users_api.create_role(
            role_id=os.getenv('ROLE_ID'),
            role_input=role_input
        )
        print(f"Role created: {result}")
        return result
    except Exception as e:
        print(f"Failed to create role: {e}")
        raise

Management Examples

System Information

def get_system_info_example():
    client = AmorphicApiClient.create_with_auth(
    host=os.getenv('API_GW_URL'),
    role_id=os.getenv('ROLE_ID'),
    ssl_ca_cert=certifi.where(),
    debug=False,
    aws_profile=None,
    env_var='PAT_TOKEN'

)

    try:
        # System information is automatically retrieved during client initialization
        # But you can also get it manually:
        system_info = client.get_system_information(role_id=os.getenv('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
    except Exception as e:
        print(f"Failed to get system information: {e}")
        raise

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