Aparavi Client SDK (Python)

A Python SDK for executing Aparavi pipelines using the Debug Adapter Protocol (DAP). This client provides a simplified interface for connecting to Aparavi DAP servers, executing pipelines, managing data transfer operations, and interacting with AI services.

Features

• DAP-based communication for reliable pipeline execution
• Simple execute-and-exit workflow for pipeline automation
• Comprehensive error handling and logging
• Automatic API key management for all DAP commands
• Object-oriented data pipe management with context manager support
• Parallel file upload capabilities with progress events (default: 64 concurrent)
• Token-based operations for data pipe commands
• Type hints with full typing support
• AI chat functionality with structured JSON responses
• Event monitoring for real-time pipeline status
• Automatic reconnection with configurable persistence
• Command-line interface for pipeline management

Table of Contents

• Installation
• Configuration
• CLI Basics
• SDK Quick Start
• API Reference
• Common Patterns
• Troubleshooting
• License

Installation

pip install aparavi-client-python

Note: The distribution files use underscores (aparavi_client_python), but the module is imported as aparavi_client:

from aparavi_client import AparaviClient

The package includes both the SDK library and a CLI tool.

Configuration

Environment Variables

You can configure the client using environment variables or a .env file:

# .env file
APARAVI_APIKEY=your-api-key-here
APARAVI_URI=https://eaas.aparavi.com

The client will automatically parse the .env file if it exists and use the values as defaults. The priority order is:

1. Constructor parameters (highest priority)
2. .env file values
3. Default values (lowest priority)

You can also pass configuration directly to the constructor:

client = AparaviClient(
    uri='https://eaas.aparavi.com',
    auth='your-api-key'
)

Environment Variable Substitution in Pipelines

The SDK automatically performs template variable substitution in pipeline configurations. Any string containing ${APARAVI_*} will be replaced with the corresponding value from your .env file.

Example .env file:

APARAVI_APIKEY=your-api-key
APARAVI_URI=https://eaas.aparavi.com
APARAVI_PROJECT_ID=project-123
APARAVI_INPUT_PATH=/data/input
APARAVI_OUTPUT_PATH=/data/output

Example pipeline configuration:

{
  "pipeline": {
    "project_id": "${APARAVI_PROJECT_ID}",
    "components": [
      {
        "id": "data-processor",
        "provider": "transform",
        "config": {
          "inputPath": "${APARAVI_INPUT_PATH}",
          "outputPath": "${APARAVI_OUTPUT_PATH}",
          "apiKey": "${APARAVI_APIKEY}"
        }
      }
    ]
  }
}

Using the pipeline:

# Variables are automatically substituted when the pipeline starts
result = await client.use(filepath='pipeline.json')

Key features:

• ✅ Only variables starting with APARAVI_ are substituted (for security)
• ✅ Unknown variables are left unchanged
• ✅ Works with nested objects and arrays
• ✅ Preserves the original pipeline configuration object

CLI Basics

The package includes an aparavi command-line tool for managing pipelines and file uploads.

CLI Commands

Start a pipeline:

aparavi start my-pipeline.json --apikey YOUR_KEY

Upload files:

aparavi upload files/*.csv --pipeline ./pipeline.json --apikey YOUR_KEY
# or with existing task token
aparavi upload files/*.csv --token TASK_TOKEN --apikey YOUR_KEY
# with custom thread count (default is 4)
aparavi upload files/*.csv --token TASK_TOKEN --threads 10 --apikey YOUR_KEY

Monitor pipeline status:

aparavi status --token TASK_TOKEN --apikey YOUR_KEY

Monitor pipeline events:

aparavi events DETAIL,SUMMARY --token TASK_TOKEN --apikey YOUR_KEY
# or monitor all events
aparavi events ALL --token TASK_TOKEN --apikey YOUR_KEY
# with log file
aparavi events ALL --token TASK_TOKEN --log events.log --apikey YOUR_KEY

Stop a pipeline:

aparavi stop --token TASK_TOKEN --apikey YOUR_KEY

SDK Quick Start

Basic Pipeline Execution

from aparavi_client import AparaviClient

# Create client
client = AparaviClient(
    uri='https://eaas.aparavi.com',
    auth='your-api-key'
)

# Connect to server
await client.connect()

# Start pipeline from file
result = await client.use(filepath='pipeline.json')
print(f'Pipeline started with token: {result["token"]}')

# Disconnect
await client.disconnect()

Using Context Manager (Recommended)

from aparavi_client import AparaviClient

# Context manager handles connect/disconnect automatically
async with AparaviClient(uri='https://eaas.aparavi.com', auth='your-api-key') as client:
    # Client is automatically connected
    result = await client.use(filepath='pipeline.json')
    token = result['token']
    
    # Send data
    response = await client.send(token, 'Process this text')
    # Client automatically disconnects here

Persistent Connection with Auto-Reconnect

from aparavi_client import AparaviClient

# Declare connection callbacks
async def on_connected(info: str) -> None:
    print(f'Connected: {info}')

async def on_disconnected(reason: str, has_error: bool) -> None:
    if has_error:
        print(f'Connection lost: {reason}')

# Create client with automatic reconnection enabled
client = AparaviClient(
    uri='https://eaas.aparavi.com',
    auth='your-api-key',
    persist=True,              # Enable automatic reconnection
    reconnect_delay=2.0,       # Wait 2 seconds before reconnection attempts (default: 1.0)
    on_connected=on_connected,
    on_disconnected=on_disconnected
)

await client.connect()
# If connection is lost, the client will automatically attempt to reconnect

Data Transfer with Pipes

# Using context manager (recommended)
import json

async with await client.pipe(token=myToken, mimetype='application/json') as pipe:
    for item in data_items:
        await pipe.write(json.dumps(item).encode())
    results = await pipe.close()

Bulk File Upload (Parallel)

from aparavi_client import AparaviClient

# Declare event handler
async def handle_events(event):
    if event['event'] == 'apaevt_status_upload':
        body = event['body']
        print(f"{body['filepath']}: {body['action']} - {body['bytes_sent']}/{body['file_size']} bytes")

client = AparaviClient(
    auth='your-api-key',
    on_event=handle_events
)

await client.connect()

# Simple file list (default: 64 concurrent uploads)
files = ['doc1.pdf', 'data.csv', 'report.docx']
results = await client.send_files(files, token)

# With custom concurrency control
results = await client.send_files(files, token, max_concurrent=10)

# With metadata and MIME types
files = [
    ('report.pdf', {'department': 'finance'}),
    ('data.csv', {'type': 'sales_data'}, 'text/csv')
]
results = await client.send_files(files, token)

await client.disconnect()

API Reference

AparaviClient

Constructor

AparaviClient(uri: str, auth: str, **kwargs)

Parameters:

• uri (str): Server URI (default: https://eaas.aparavi.com)
• auth (str): API key for authentication
• on_event (callable, optional): Event handler for server events
• on_connected (callable, optional): Connection established callback
• on_disconnected (callable, optional): Connection lost callback
• persist (bool, optional): Enable automatic reconnection (default: False)
• reconnect_delay (float, optional): Delay between reconnection attempts in seconds (default: 1.0)

Connection Methods

async connect() -> None
Establish connection to the Aparavi server.

async disconnect() -> None
Close connection to the Aparavi server and stop automatic reconnection.

Execution Methods

async use(**kwargs) -> Dict[str, Any]
Start an Aparavi pipeline for processing data. Automatically performs environment variable substitution on the pipeline configuration.

Parameters:

• pipeline (dict, optional): Pipeline configuration dictionary
• filepath (str, optional): Path to JSON file containing pipeline configuration
• token (str, optional): Custom token for the pipeline (auto-generated if not provided)
• threads (int, optional): Number of threads for execution (default: 1)
• args (List[str], optional): Command line arguments to pass to pipeline

Returns: Dictionary containing the task token and other metadata

async terminate(token: str) -> None
Terminate a running pipeline.

async get_task_status(token: str) -> Dict[str, Any]
Get the current status of a running pipeline.

Data Methods

async send(token: str, data: Union[str, bytes], objinfo: Dict = {}, mimetype: str = None) -> Dict[str, Any]
Send data directly to a pipeline.

Important: Use this method with pipelines that have webhook, filesys, or dropper as the source component. For chat/Q&A systems, use chat() method instead with a chat source component.

async send_files(files: List, token: str, max_concurrent: int = 64) -> List[Dict[str, Any]]
Upload multiple files in parallel with configurable concurrency.

Parameters:

• files (list): List of file paths or tuples (filepath, objinfo) or (filepath, objinfo, mimetype)
• token (str): Task token of the pipeline
• max_concurrent (int, optional): Maximum concurrent uploads (default: 64)

async pipe(token: str, objinfo: Dict = {}, mimetype: str = None, provider: str = None) -> DataPipe
Create a streaming data pipe for sending large datasets.

Chat Methods

async chat(token: str, question: Question) -> Dict[str, Any]
Ask a question to Aparavi's AI and get an intelligent response.

Important: Use this method with pipelines that have chat as the source component. For document processing/uploads, use send() or send_files() with a webhook source instead.

Example:

from aparavi_client.schema import Question

question = Question()
question.addQuestion('What are the key findings?')

response = await client.chat(token='chat-token', question=question)

Event Methods

async set_events(token: str, event_types: List[str]) -> None
Subscribe to specific types of events from the server.

Example:

await client.set_events(token, ['apaevt_status_upload', 'apaevt_status_processing'])

DataPipe

Created via client.pipe() method. Provides a stream-like interface for uploading data.

async open() -> DataPipe
Open the pipe for data transmission. Must be called before any write() operations.

async write(buffer: bytes) -> None
Write data to the pipe. Can be called multiple times to stream large datasets.

async close() -> Optional[Dict[str, Any]]
Close the pipe and get the processing results.

Question

Question builder for AI chat operations.

Constructor

Question(expectJson: bool = False)

Methods

addQuestion(text: str) -> Question
Add the main question text.

addInstruction(subtitle: str, instructions: str) -> Question
Add specific instructions to guide the AI's response.

addExample(given: str, result: Any) -> Question
Provide an example of the desired response format.

addContext(context: Union[str, Dict[str, Any]]) -> Question
Add contextual information for the AI.

addHistory(history: QuestionHistory) -> Question
Add conversation history for context.

addDocument(doc: Doc) -> Question
Add a document to the question context.

addFilter(filter: DocFilter) -> Question
Add a document filter to narrow the search scope.

Common Patterns

AI Chat with Structured JSON Responses

from aparavi_client import AparaviClient
from aparavi_client.schema import Question

# Start chat pipeline once at the beginning
client = AparaviClient(auth='your-api-key')
await client.connect()

result = await client.use(filepath='chat_pipeline.json')
token = result['token']

# Simple question
async def ask_question(my_question: str) -> str:
    question = Question()
    question.addQuestion(my_question)
    
    response = await client.chat(token=token, question=question)
    
    if 'answers' in response and len(response['answers']) > 0:
        return response['answers'][0]
    return 'No answer received'

# Structured JSON response
async def extract_data(source_document: str):
    question = Question(expectJson=True)
    question.addQuestion('Extract email addresses and phone numbers')
    question.addExample(
        'Find contacts',
        {'emails': ['john@company.com'], 'phones': ['555-1234']}
    )
    question.addContext(source_document)
    
    response = await client.chat(token=token, question=question)
    
    if 'answers' in response and len(response['answers']) > 0:
        return response['answers'][0]
    return {}

# Use the functions
answer = await ask_question('What are the main themes in these documents?')
data = await extract_data('Contact us at john@company.com or 555-1234')

Document Processing Workflow

from aparavi_client import AparaviClient

async def process_documents():
    async with AparaviClient(uri='https://eaas.aparavi.com', auth='api_key') as client:
        # Start document processing pipeline
        result = await client.use(filepath='document_analyzer.json')
        token = result['token']
        
        # Upload files
        files = ['report1.pdf', 'data.xlsx', 'notes.txt']
        results = await client.send_files(files, token)
        
        return results

Event Monitoring

from typing import Dict, Any

async def handle_events(event: Dict[str, Any]) -> None:
    event_type = event['event']
    body = event['body']
    
    if event_type == 'apaevt_status_upload':
        if body['action'] == 'write':
            progress = (body['bytes_sent'] / body['file_size']) * 100
            print(f'Upload progress: {progress:.1f}%')

# Create client with event handler
client = AparaviClient(
    uri='https://eaas.aparavi.com',
    auth='api_key',
    on_event=handle_events
)

await client.connect()

# Start pipeline and subscribe to events
result = await client.use(filepath='pipeline.json')
await client.set_events(result['token'], ['apaevt_status_upload', 'apaevt_status_processing'])

# Monitor status
while True:
    status = await client.get_task_status(result['token'])
    if status['state'] in ['completed', 'failed']:
        break
    await asyncio.sleep(1)

Troubleshooting

Error Handling

from aparavi_client import AparaviClient, AparaviException

try:
    client = AparaviClient(uri='https://eaas.aparavi.com', auth='your-api-key')
    await client.connect()
    
    result = await client.use(filepath='pipeline.json')
    print(f'Pipeline started: {result["token"]}')
    
except AparaviException as e:
    print(f'Aparavi Error: {e}')
except ConnectionError as e:
    print(f'Connection Error: {e}')
except Exception as e:
    print(f'Error: {e}')
finally:
    if client:
        await client.disconnect()

Common Issues

Connection Failed:

• Ensure you're using the latest version: pip install --upgrade aparavi-client-python
• Check your network connection and firewall settings

403 Forbidden Error:

• Your API key is invalid, expired, or lacks permissions
• Get a new API key from your Aparavi account settings

Pipeline Not Found:

• Verify the pipeline file path and JSON format
• Use absolute paths if relative paths don't work

Upload Errors:

• Ensure files are accessible and not too large
• Check file permissions

Authentication Errors:

• Verify your API key is correct
• Ensure the key has the necessary permissions

Requirements

• Python 3.8 or higher
• Valid API key for authentication

License

MIT

Support

• Documentation: https://docs.aparavi.com