nexla-sdk 0.1.0

A Python SDK for the Nexla API

Nexla Python SDK

A Python SDK for interacting with the Nexla API.

Installation

pip install nexla-sdk

Authentication

The Nexla SDK requires a Service Key for authentication. You can create a service key from the Nexla UI:

1. Go to your Nexla UI instance (e.g., https://dataops.nexla.io)
2. Navigate to the Authentication screen in the Settings section
3. Click the Create Service Key button
4. Store the service key securely - it should be treated as highly sensitive since it is equivalent to your account password

Quick Start

from nexla_sdk import NexlaClient

# Initialize the client with your service key
client = NexlaClient(service_key="your_nexla_service_key")

# List flows - returns a FlowList object with typed items
flows = client.flows.list()
print(f"Found {flows.total} flows")

# Access Flow objects with proper typing
for flow in flows.items:
    print(f"Flow name: {flow.name}, ID: {flow.id}")
    print(f"Active: {flow.config.is_active}")

# Get a specific flow - returns a Flow object
flow = client.flows.get("flow_id")
print(f"Flow details: {flow.name}, type: {flow.flow_type}")

# Create a data source
source_data = {
    "name": "My New Source",
    "description": "Created via SDK",
    "config": {
        "connector_type": "file",
        # Additional configuration...
    }
}

# Returns a Source object with proper typing
new_source = client.sources.create(source_data)
print(f"Created source: {new_source.id}, name: {new_source.name}")

Features

The SDK provides access to the following Nexla API features:

• Flows management
• Sources management
• Destinations (Data Sinks) management
• Nexsets (Data Sets) management
• Credentials management
• Data Maps (Lookups) management
• Transforms management
• Webhooks integration
• Organization management
• User management
• Teams management
• Projects management

Type-Safe Models

The Nexla SDK uses Pydantic models to provide type safety and validation for API responses. All API methods return properly typed model objects instead of raw dictionaries:

# Get a flow - returns a Flow object
flow = client.flows.get("flow_id")

# Access properties with proper typing
print(flow.name)
print(flow.config.is_active)
print(flow.status.status if flow.status else "No status")

# List sources - returns a SourceList object
sources = client.sources.list()

# Access typed items
for source in sources.items:
    print(f"Source: {source.name}")
    print(f"Connector type: {source.config.connector_type}")

Examples

Working with Flows

# List all flows - returns a FlowList
flows = client.flows.list()

# Get details of a specific flow - returns a Flow
flow = client.flows.get("flow_id")

# Activate a flow - returns the updated Flow
activated_flow = client.flows.activate("flow_id")
print(f"Flow activated: {activated_flow.config.is_active}")

# Pause a flow - returns the updated Flow
paused_flow = client.flows.pause("flow_id")
print(f"Flow active: {paused_flow.config.is_active}")

# Create a copy of a flow - returns the new Flow
new_flow = client.flows.copy("flow_id", new_name="Copy of my flow")
print(f"New flow created: {new_flow.id}, name: {new_flow.name}")

Working with Data Sources

# List all data sources - returns a SourceList
sources = client.sources.list()

# Get details of a specific data source - returns a Source
source = client.sources.get("source_id")

# Create a new data source - returns a Source
source_config = {
    "name": "My API Source",
    "description": "Created via SDK",
    "config": {
        "connector_type": "rest_api",
        # Other configuration properties...
    }
}
new_source = client.sources.create(source_config)
print(f"New source ID: {new_source.id}")

# Activate a data source - returns the updated Source
activated_source = client.sources.activate("source_id")

Working with Credentials

# List all credentials - returns a CredentialList
credentials = client.credentials.list()

# Create a new credential - returns a Credential
cred_config = {
    "name": "AWS S3 Credential",
    "credential_type": "aws_s3",
    "credential_details": {
        "credential_type": "aws_s3",
        "properties": {
            "access_key": "your_access_key",
            "secret_key": "your_secret_key",
            "region": "us-west-2"
        }
    }
}
new_credential = client.credentials.create(cred_config)
print(f"New credential ID: {new_credential.id}")

# Test a credential - returns a ProbeResult
probe_result = client.credentials.probe("credential_id")
print(f"Probe success: {probe_result.success}")

# Get a directory tree for a credential - returns a DirectoryTree
tree = client.credentials.probe_tree("credential_id", path="/some/path")
for item in tree.items:
    print(f"{item.name} ({item.type}): {item.path}")

Error Handling

The SDK provides specific error classes for different error types:

from nexla_sdk import NexlaClient
from nexla_sdk.exceptions import NexlaAPIError, NexlaAuthError, NexlaValidationError

client = NexlaClient(service_key="your_nexla_service_key")

try:
    flows = client.flows.list()
except NexlaAuthError:
    print("Authentication failed. Please check your service key.")
except NexlaAPIError as e:
    print(f"API error: {str(e)}, Status code: {e.status_code}")
except NexlaValidationError as e:
    print(f"Validation error: {str(e)}")

License

This project is licensed under the terms of the MIT license.

Development

Running Integration Tests

The SDK includes a comprehensive suite of integration tests that verify the functionality against a real Nexla API instance. These tests create temporary test resources, perform operations on them, and then clean up.

To run the integration tests:

1. Set up your environment variables:

   export NEXLA_TEST_API_URL="https://your-nexla-api-url"
   export NEXLA_TEST_SERVICE_KEY="your-nexla-service-key"
   export NEXLA_TEST_API_VERSION="v1"  # Optional, defaults to v1
   export NEXLA_TEST_LOG_LEVEL="INFO"  # Optional, defaults to INFO
   

   Alternatively, create a .env file in the project root or tests directory with these variables.

2. Run the tests using the provided script:

   ./run_integration_tests.py
   

   Or use pytest directly:

   pytest -v -m integration tests/api/
   

The integration tests cover:

• Flows API (create, get, update, delete, list, activate/pause, tags)
• Sources API (create, get, update, delete, list, activate/pause)
• Nexsets API (Data Sets) (create, get, update, delete, list, schema, samples)
• Projects API (create, get, update, delete, list, add/remove flows)
• Users API (get current user, preferences, list users, metrics)

Note: These tests require API credentials with appropriate permissions to create and manage resources.