Metadata-Version: 2.3
Name: netsuite-async
Version: 0.1.3
Summary: High-performance async Python client for NetSuite REST API with concurrency controls and intelligent pagination
Keywords: netsuite,async,rest,api,client,erp,oauth
Author: Michael Drummond
Author-email: Michael Drummond <97915761+MichaelD3289@users.noreply.github.com>
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Internet :: WWW/HTTP :: HTTP Servers
Classifier: Topic :: Office/Business :: Financial :: Accounting
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Dist: authlib>=1.6.5
Requires-Dist: httpx>=0.28.1
Requires-Dist: pytest>=7.0 ; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21.0 ; extra == 'dev'
Requires-Dist: pytest-cov>=4.0 ; extra == 'dev'
Requires-Dist: black>=23.0 ; extra == 'dev'
Requires-Dist: ruff>=0.1.0 ; extra == 'dev'
Requires-Dist: mypy>=1.0 ; extra == 'dev'
Requires-Python: >=3.12
Project-URL: Documentation, https://github.com/MichaelD3289/netsuite-async#readme
Project-URL: Homepage, https://github.com/MichaelD3289/netsuite-async
Project-URL: Issues, https://github.com/MichaelD3289/netsuite-async/issues
Project-URL: Repository, https://github.com/MichaelD3289/netsuite-async
Provides-Extra: dev
Description-Content-Type: text/markdown

# netsuite-async

[![Python 3.12+](https://img.shields.io/badge/python-3.12+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

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
- **Flexible record mapping** - Dynamic record access with fuzzy matching for typos

## Installation

```bash
pip install netsuite-async
```

## Quick Start

### 1. Set up environment variables

```bash
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

```python
import asyncio
from netsuite_async import AsyncNetsuiteRestClient, OAuth1AsyncAuthProvider, OAuth1Credentials

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']}")
        
        # 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:

```python
# 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)
```

### Query Filtering

```python
# 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:

```python
# 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:

```python
# 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

```python
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

```python
# Create a new customer
customer_data = {
    "companyname": "Acme Corp",
    "email": "contact@acme.com",
    "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.

```python
# 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:

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

**Methods:**
- `get(internal_id: str) -> FullRecord` - Fetch a single record
- `get_many(ids: List[str], max_concurrency: int = 10) -> Dict[str, FetchResult]` - Fetch multiple records concurrently
- `list_summaries(q: Optional[str] = None) -> List[SummaryRecord]` - List all records (with pagination)
- `list_summaries_concurrent(max_concurrency: int = 10) -> List[SummaryRecord]` - Concurrent pagination
- `all_full(q: Optional[str] = None) -> List[FullRecord]` - Fetch all records as full records (sequential)
- `all_full_concurrent(q: Optional[str] = None, max_concurrency: int = 10) -> Dict[str, FetchResult]` - Fetch all records as full records (concurrent)
- `iter_summary_pages(limit: int = 1000, offset: int = 0) -> AsyncGenerator` - Iterate through pages
- `create(data: dict) -> str` - Create a new record
- `update(internal_id: str, data: dict) -> 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:**
```python
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)`

### Authentication

#### `OAuth1AsyncAuthProvider`

Auth provider for OAuth1-authenticated HTTP clients:

```python
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:

```python
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:
   ```python
   # 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**:
   ```python
   # 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**:
   ```python
   # 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:
   ```python
   # 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](LICENSE) file for details.

## Contributing

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

## Support

- [NetSuite REST API Documentation](https://docs.oracle.com/en/cloud/saas/netsuite/ns-online-help/chapter_1540391670.html)
- [Report Issues](https://github.com/your-username/netsuite-async/issues)
- [Discussions](https://github.com/your-username/netsuite-async/discussions)