netsuite-async 0.2.4

High-performance async Python client for NetSuite REST API with concurrency controls and intelligent pagination

netsuite-async

A high-performance, async-first Python client for NetSuite's REST API with built-in concurrency controls, intelligent pagination, and comprehensive error handling.

Features

• Async-first design - Built from the ground up for asyncio
• High-performance concurrent operations - Configurable concurrency limits with semaphore-based rate limiting
• Smart pagination - Automatic handling of NetSuite's pagination with both sequential and concurrent strategies
• OAuth 1.0 authentication - Secure authentication with HMAC-SHA256 signatures
• Comprehensive error handling - Detailed exception hierarchy for different error types
• Type-safe - Full type hints and dataclass models
• Structured parameters - Type-safe parameter classes for NetSuite API options
• Flexible record mapping - Dynamic record access with fuzzy matching for typos

Installation

pip install netsuite-async

Quick Start

1. Set up environment variables

export NETSUITE_ACCOUNT_ID="your_account_id"
export NETSUITE_CONSUMER_KEY="your_consumer_key"
export NETSUITE_CONSUMER_SECRET="your_consumer_secret"
export NETSUITE_TOKEN_KEY="your_token_key"
export NETSUITE_TOKEN_SECRET="your_token_secret"

2. Basic usage

import asyncio
from netsuite_async import AsyncNetsuiteRestClient, OAuth1AsyncAuthProvider, OAuth1Credentials
from netsuite_async import GetParams, UpdateParams

async def main():
    # Option 1: Use default OAuth1 auth provider (reads from environment)
    async with AsyncNetsuiteRestClient() as netsuite:
        # Use built-in record types (no registration needed)
        customer = await netsuite.customers.get("123")
        print(f"Customer: {customer.raw['companyname']}")
        
        # Use structured parameters for better type safety
        get_params = GetParams(fields=["companyname", "email"], expand=True)
        customer = await netsuite.customers.get("123", params=get_params)
        
        # List records with pagination
        customers = await netsuite.customers.list_summaries()
        print(f"Found {len(customers)} customers")
        
        # Other built-in types work the same way
        projects = await netsuite.projects.list_summaries()
        employees = await netsuite.employees.get("456")
        
        # Concurrent bulk operations
        customer_ids = ["123", "456", "789"]
        results = await netsuite.customers.get_many(customer_ids, max_concurrency=5)
        
        for customer_id, result in results.items():
            if result.success:
                print(f"Success {customer_id}: {result.data.raw['companyname']}")
            else:
                print(f"Error {customer_id}: {result.error}")
    
    # Option 2: Custom auth provider with explicit credentials
    credentials = OAuth1Credentials(
        consumer_key="your_key",
        consumer_secret="your_secret",
        token_key="your_token",
        token_secret="your_token_secret",
        realm="your_account_id"
    )
    auth_provider = OAuth1AsyncAuthProvider(credentials)
    
    async with AsyncNetsuiteRestClient(auth_provider) as netsuite:
        customers = await netsuite.customers.list_summaries()

asyncio.run(main())

Advanced Usage

Concurrent Pagination

For large datasets, use concurrent pagination for significantly faster retrieval:

# Sequential pagination (default)
customers = await netsuite.customers.list_summaries()

# Concurrent pagination (faster for large datasets)
customers = await netsuite.customers.list_summaries_concurrent(
    max_concurrency=10
)

# Get all records as full records (sequential)
full_customers = await netsuite.customers.all_full()

# Get all records as full records (concurrent)
full_results = await netsuite.customers.all_full_concurrent(max_concurrency=10)

Structured Parameters

Use type-safe parameter classes for better IDE support and validation:

from netsuite_async import GetParams, UpdateParams, CreateParams

# Type-safe GET parameters
get_params = GetParams(
    fields=["companyname", "email", "phone"],  # List of specific fields
    expand=True,                               # Expand sublists and subrecords
    simple_enum_format=True                   # Use simple enum format
)
customer = await netsuite.customers.get("123", params=get_params)

# Or use comma-separated string
get_params = GetParams(fields="companyname,email,phone", expand=True)
customer = await netsuite.customers.get("123", params=get_params)

# Type-safe UPDATE parameters
update_params = UpdateParams(
    replace=["addressbook"],          # Replace addressbook sublist
    replace_selected_fields=True      # Replace only specified fields
)
update_data = {"phone": "555-0124"}
await netsuite.customers.update("123", update_data, params=update_params)

# Type-safe CREATE parameters
create_params = CreateParams(
    replace=["addressbook", "contactlist"]  # Replace multiple sublists as list
)
customer_data = {"companyname": "Acme Corp", "email": "[e-mail]"}
customer_id = await netsuite.customers.create(customer_data, params=create_params)

# You can still use plain dictionaries for backward compatibility
plain_params = {"fields": "companyname,email", "expandSubResources": "true"}
customer = await netsuite.customers.get("123", params=plain_params)

Query Filtering

# Filter records using NetSuite's query syntax
active_customers = await netsuite.customers.list_summaries(
    q='isinactive IS false'
)

# Complex queries
recent_orders = await netsuite.sales_orders.list_summaries(
    q='trandate AFTER "2024-01-01" AND status IS "Pending Fulfillment"'
)

Built-in Record Types

The following record types are available out of the box with full IDE autocompletion:

# Built-in record types (no registration needed)
await netsuite.customers.list_summaries()    # customer records
await netsuite.contacts.get("123")           # contact records  
await netsuite.projects.list_summaries()     # job records
await netsuite.partners.get("456")           # partner records
await netsuite.employees.list_summaries()    # employee records

Custom Record Types

For other record types, register them first:

# Register custom record types
RecordCatalog.register("items", "inventoryitem")
RecordCatalog.register("equipment_config", "customrecord_equipment_config")

# Then use them
items = await netsuite.items.list_summaries()
configs = await netsuite.equipment_config.get("456")

Error Handling

from netsuite_async import (
    NetsuiteAuthError,
    NetsuiteRateLimitError,
    NetsuiteValidationError,
    NetsuiteServerError
)

try:
    customer = await netsuite.customers.get("invalid_id")
except NetsuiteAuthError:
    print("Authentication failed - check credentials")
except NetsuiteRateLimitError:
    print("Rate limited - implement backoff strategy")
except NetsuiteValidationError as e:
    print(f"Validation error: {e}")
except NetsuiteServerError:
    print("NetSuite server error - retry later")

Creating and Updating Records

# Create a new customer
customer_data = {
    "companyname": "Acme Corp",
    "email": "[e-mail]",
    "phone": "555-0123"
}
customer_id = await netsuite.customers.create(customer_data)

# Update existing customer
update_data = {"phone": "555-0124"}
await netsuite.customers.update(customer_id, update_data)

API Reference

Core Classes

AsyncNetsuiteRestClient

The main client for interacting with NetSuite's REST API. Supports context manager usage.

# Default usage (uses OAuth1AsyncAuthProvider with environment variables)
client = AsyncNetsuiteRestClient()

# Custom auth provider
auth_provider = OAuth1AsyncAuthProvider(credentials)
client = AsyncNetsuiteRestClient(
    auth_provider=auth_provider,
    account_id="your_account_id",  # optional, reads from NETSUITE_ACCOUNT_ID if not provided
    base_url="https://custom.suitetalk.api.netsuite.com/services/rest/record/v1/"  # optional
)

# Context manager usage (automatically manages HTTP client lifecycle)
async with AsyncNetsuiteRestClient() as client:
    customers = await client.customers.list_summaries()

RecordAccessor

Provides CRUD operations for NetSuite records. Accessed via dynamic attributes on the client:

# These create RecordAccessor instances
customers = client.customers
items = client.items
custom_records = client.my_custom_record

Methods:

• get(internal_id: str, params: Optional[ParamsLike] = None) -> FullRecord - Fetch a single record
• get_many(ids: List[str], params: Optional[ParamsLike] = None, max_concurrency: int = 10) -> Dict[str, FetchResult] - Fetch multiple records concurrently
• list_summaries(q: Optional[str] = None, params: Optional[ParamsLike] = None) -> List[SummaryRecord] - List all records (with pagination)
• list_summaries_concurrent(max_concurrency: int = 10, params: Optional[ParamsLike] = None) -> List[SummaryRecord] - Concurrent pagination
• all_full(q: Optional[str] = None, params: Optional[ParamsLike] = None) -> List[FullRecord] - Fetch all records as full records (sequential)
• all_full_concurrent(q: Optional[str] = None, params: Optional[ParamsLike] = None, max_concurrency: int = 10) -> Dict[str, FetchResult] - Fetch all records as full records (concurrent)
• iter_summary_pages(limit: int = 1000, offset: int = 0, params: Optional[ParamsLike] = None) -> AsyncGenerator - Iterate through pages
• create(data: dict, params: Optional[ParamsLike] = None) -> str - Create a new record
• update(internal_id: str, data: dict, params: Optional[ParamsLike] = None) -> str - Update an existing record

RecordCatalog

Maps friendly names to NetSuite record type IDs. Ships with common record types pre-registered:

Built-in mappings:

• customers → customer
• contacts → contact
• projects → job
• partners → partner
• employees → employee

Register additional mappings:

RecordCatalog.register("items", "inventoryitem")
RecordCatalog.register("sales_orders", "salesorder")
RecordCatalog.register("custom_record", "customrecord_my_record")

# Get all registered names
names = RecordCatalog.registered_names()

Data Models

SummaryRecord

Lightweight record representation from list operations:

• id: str - NetSuite internal ID
• type: str - Record type name
• links: List[Link] - HATEOAS links
• raw: Dict[str, Any] - Raw NetSuite data

FullRecord

Complete record data from get operations:

• id: str - NetSuite internal ID
• type: str - Record type name
• links: List[Link] - HATEOAS links
• raw: Dict[str, Any] - Raw NetSuite data

FetchResult

Union type for bulk operation results:

• FetchResultSuccess(data: FullRecord, success: True)
• FetchResultError(error: str, success: False)

Parameter Classes

GetParams

Parameters for GET requests to retrieve NetSuite records:

• fields: Union[List[str], str] - The names of the fields and sublists on the record. Only the selected fields and sublists will be returned in the response. Can be a string (comma-separated) or list of field names.
• expand: bool - Set to True to automatically expand all sublists, sublist lines, and subrecords on this record.
• simple_enum_format: bool - Set to True to return enumeration values in a format that only shows the internal ID value.

UpdateParams

Parameters for PATCH requests to update NetSuite records:

• replace: Union[List[str], str] - The names of sublists on this record. All sublist lines will be replaced with lines specified in the request. The names are delimited by comma.
• replace_selected_fields: bool - If set to True, all fields that should be deleted in the update request, including body fields, must be included in the 'replace' query parameter.

CreateParams

Parameters for POST requests to create NetSuite records:

• replace: Union[List[str], str] - The names of sublists on this record. All sublist lines will be replaced with lines specified in the request. The names are delimited by comma.

ParamsLike

Type alias for parameter inputs: Union[BaseParams, Mapping[str, Any]]

Accepts either structured parameter classes (GetParams, UpdateParams, CreateParams) or plain dictionaries for backward compatibility.

Authentication

OAuth1AsyncAuthProvider

Auth provider for OAuth1-authenticated HTTP clients:

from netsuite_async import OAuth1AsyncAuthProvider, OAuth1Credentials

# Using environment variables (default)
auth_provider = OAuth1AsyncAuthProvider()

# Using explicit credentials
credentials = OAuth1Credentials(
    consumer_key="key",
    consumer_secret="secret", 
    token_key="token",
    token_secret="token_secret",
    realm="123456"  # optional, uses NETSUITE_ACCOUNT_ID from env if not provided
)
auth_provider = OAuth1AsyncAuthProvider(credentials=credentials)

# With custom options
auth_provider = OAuth1AsyncAuthProvider(
    timeout=60.0,
    retries=3,
    headers={"Custom-Header": "value"}
)

async_oauth1_client()

Convenience function that creates an authenticated HTTP client directly:

from netsuite_async import async_oauth1_client

# Creates and returns an AsyncOAuth1Client
client = await async_oauth1_client()

Environment Variables

Variable	Description	Required
NETSUITE_ACCOUNT_ID	Your NetSuite account ID	Yes
NETSUITE_CONSUMER_KEY	OAuth consumer key	Yes
NETSUITE_CONSUMER_SECRET	OAuth consumer secret	Yes
NETSUITE_TOKEN_KEY	OAuth token key	Yes
NETSUITE_TOKEN_SECRET	OAuth token secret	Yes
NETSUITE_REALM	OAuth realm (defaults to account ID)	No

Performance Tips

1. Use concurrent operations for bulk data retrieval:

   # Fast - concurrent requests
   results = await client.customers.get_many(ids, max_concurrency=10)
   
   # Slow - sequential requests  
   results = [await client.customers.get(id) for id in ids]
   

2. Choose the right pagination strategy:

   # Fast for large datasets
   records = await client.customers.list_summaries_concurrent()
   
   # Memory efficient for streaming
   async for page in client.customers.iter_summary_pages():
       process_page(page)
   

3. Use all_full methods for complete record data:

   # Sequential - memory efficient
   full_records = await client.customers.all_full()
   
   # Concurrent - faster for large datasets
   results = await client.customers.all_full_concurrent(max_concurrency=10)
   

4. Tune concurrency limits based on your NetSuite plan's rate limits:

   # Conservative for shared environments
   results = await client.customers.get_many(ids, max_concurrency=5)
   
   # Aggressive for dedicated environments
   results = await client.customers.get_many(ids, max_concurrency=20)
   

Requirements

• Python 3.12+
• httpx >= 0.28.1
• authlib >= 1.6.5

License

MIT License - see LICENSE file for details.

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

Support

• NetSuite REST API Documentation
• Report Issues
• Discussions