Helix Connect Python SDK

Official Python SDK for Helix Connect Data Marketplace - a secure, scalable platform for exchanging datasets between producers and consumers.

🚀 Features

• Consumer API: Download and subscribe to datasets, browse the marketplace
• Producer API: Upload and manage datasets, invite partners, price on the marketplace (includes all consumer features)
• Secure: AWS SigV4 authentication + AES-256-GCM envelope encryption
• Efficient: Compress-then-encrypt pipeline with ~90% space savings
• Progress Tracking: Real-time upload/download progress callbacks
• Notifications: SQS-based dataset update notifications with long-polling
• Type-Safe: Full type hints with mypy support

📦 Installation

pip install helix-connect

Development Installation

git clone https://github.com/helix-tools/helix-connect-sdk-python.git
cd helix-connect-sdk-python
pip install -e ".[dev]"

🔧 Prerequisites

• Python 3.10 or higher
• AWS credentials (provided during customer onboarding)
• Helix Connect customer ID (UUID format)

📖 Quick Start

Consumer: Download Datasets

from helix_connect import HelixConsumer

# Initialize consumer. api_endpoint defaults to https://api-go.helix.tools
# (or the HELIX_API_ENDPOINT env var) -- only pass it to point at a
# different environment.
consumer = HelixConsumer(
    aws_access_key_id="your-access-key",
    aws_secret_access_key="your-secret-key",
    customer_id="your-customer-id",
)

# List available datasets
datasets = consumer.list_datasets()
for ds in datasets:
    print(f"{ds['name']}: {ds['description']}")

# Download a dataset
consumer.download_dataset(
    dataset_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
    output_path="./data/my_dataset.csv"
)

# Subscribe to dataset updates
consumer.subscribe_to_dataset(dataset_id="...")

# Poll for notifications (long-polling; messages are auto-acknowledged by
# default) and download whatever they point at.
notifications = consumer.poll_notifications(
    max_messages=10,
    wait_time_seconds=20,
)
for notif in notifications:
    consumer.download_dataset(
        dataset_id=notif["dataset_id"],
        output_path=f"./downloads/{notif['dataset_id']}.csv",
    )

Producer: Upload Datasets

from helix_connect import HelixProducer

# Initialize producer (inherits all consumer capabilities)
producer = HelixProducer(
    aws_access_key_id="your-access-key",
    aws_secret_access_key="your-secret-key",
    customer_id="your-customer-id"
)

# Upload a dataset with progress tracking
def progress_callback(bytes_transferred, total_bytes):
    percent = (bytes_transferred / total_bytes) * 100
    print(f"Progress: {percent:.1f}%")

producer.upload_dataset(
    file_path="./data/my_dataset.csv",
    dataset_name="my-awesome-dataset",
    description="Q4 2024 sales data",
    data_freshness="daily",
    progress_callback=progress_callback
)

# Update dataset metadata (no re-upload)
producer.update_dataset(
    dataset_id="...",
    updates={
        "description": "Updated description",
        "visibility": "public",
        "tags": ["phone-data", "usa"],
        "version": "2.0.0",
    },
)

# Replace dataset data by re-uploading a new file
producer.update_dataset_data(
    dataset_id="...",
    file_path="./data/updated_dataset.csv"
)

# List your uploaded datasets
my_datasets = producer.list_my_datasets()

Producer: Stripe Connect Payouts

from helix_connect import HelixProducer

producer = HelixProducer(
    aws_access_key_id="your-access-key",
    aws_secret_access_key="your-secret-key",
    customer_id="your-customer-id"
)

# Start (or resume) Stripe Connect Express payout onboarding.
# Returns {"url": ..., "account_id": ...} — open `url` to submit KYC + bank
# details. The SDK never opens or redirects to it for you.
onboard = producer.connect_onboard()
print(f"Open this to finish onboarding: {onboard['url']}")

# Check payout account status any time.
status = producer.get_connect_status()
if status.get("can_price_datasets"):
    print("Payouts are enabled — datasets can be priced above $0")
else:
    print(f"Still due: {status.get('requirements_due')}")

# Once onboarding is complete, get a one-time link to the Stripe Express
# dashboard (raises PermissionDeniedError if onboarding isn't complete yet).
login = producer.create_connect_login_link()
print(f"Manage your payout account: {login['url']}")

Consumer: Marketplace

from helix_connect import HelixConsumer

consumer = HelixConsumer(
    aws_access_key_id="your-access-key",
    aws_secret_access_key="your-secret-key",
    customer_id="your-customer-id",
)

# Browse the public marketplace (all filters optional)
results = consumer.browse_marketplace(search="phone", category="telecom", page=1)
for ds in results["datasets"]:
    print(f"{ds['name']}: {ds.get('marketplace')}")

# Get the public detail view for one dataset: dataset + reviews +
# related_datasets + subscription_info
details = consumer.get_dataset_details(dataset_id="producer-1-phone-numbers")
print(details["dataset"]["name"], details["subscription_info"])

# Pay for a priced/listed dataset via Stripe Checkout. Pass exactly one of
# dataset_id (subscribe directly) or request_id (pay for a request that was
# already approved). Returns the Checkout URL only -- the SDK never opens it.
checkout_url = consumer.create_subscription_checkout(dataset_id="producer-1-phone-numbers")
print(f"Complete payment at: {checkout_url}")

Producer: Marketplace Pricing & Earnings

from helix_connect import HelixProducer

producer = HelixProducer(
    aws_access_key_id="your-access-key",
    aws_secret_access_key="your-secret-key",
    customer_id="your-customer-id",
)

# List a dataset on the marketplace at $5.00/month (cents; 0 = free)
producer.set_dataset_marketplace(
    dataset_id="producer-1-phone-numbers",
    price_monthly_cents=500,
    listed=True,
)

# Check marketplace earnings (period is optional, e.g. "2026-07")
earnings = producer.get_earnings(period="2026-07")
print(earnings)

Both marketplace surfaces raise DatasetNotFoundError (server 404) while the marketplace_payments feature flag is off.

Producer: Partner Invites

from helix_connect import HelixProducer

producer = HelixProducer(
    aws_access_key_id="your-access-key",
    aws_secret_access_key="your-secret-key",
    customer_id="your-customer-id",
)

# Invite a consumer partner and auto-grant them access to specific datasets.
# Contract violations (bad email, empty datasets, etc.) raise ValueError
# before any network traffic; a disabled partner_invite feature flag raises
# PermissionDeniedError.
result = producer.invite_consumer(
    company_name="Acme Corp",
    business_email="partner@acme.com",
    datasets=["producer-1-phone-numbers"],
)
print(result["consumer_id"], result["status"])

# List the consumers you've invited
for relation in producer.list_consumers():
    print(f"{relation['consumer_id']}: {relation['status']}")

# Deactivate an invited consumer
producer.deactivate_consumer(consumer_id="customer-abc123")

STS Session Credentials (opt-in)

By default the SDK signs every request with the long-lived AWS key you pass in (credential_mode="static", unchanged behavior). Opt into credential_mode="sts" to use that key only as a bootstrap credential: the SDK mints short-lived (15-minute) session credentials from the broker and auto-refreshes them before they expire.

from helix_connect import HelixConsumer

consumer = HelixConsumer(
    aws_access_key_id="your-bootstrap-access-key",
    aws_secret_access_key="your-bootstrap-secret-key",
    customer_id="your-customer-id",
    credential_mode="sts",  # default is "static"
)

# consumer now signs requests with auto-refreshing 15-minute session
# credentials instead of the long-lived key above -- everything else
# (list_datasets, download_dataset, etc.) works exactly the same.
datasets = consumer.list_datasets()

# Force an immediate re-mint (test/e2e hook; no-op in static mode)
consumer.force_refresh()

credential_mode="sts" also accepts broker_endpoint (override the credential-broker URL; defaults to api_endpoint), credential_scope (optional dict merged into the mint request), and auto_refresh (default True). Available on both HelixConsumer and HelixProducer. A failed mint raises CredentialRefreshError carrying the broker's error code (e.g. subscription_expired).

🏗️ Architecture

Class Hierarchy

HelixConsumer (base class)
    ↓
HelixProducer (adds upload + producer-only capabilities)

HelixProducer inherits all HelixConsumer capabilities, so a producer can also list, download, and subscribe to datasets like a consumer.

Platform administration (customer provisioning, JWT minting, platform-wide stats) is a separate, privately-distributed helix-admin package -- not part of this SDK.

Security & Encryption

The SDK implements a compress-then-encrypt pipeline with envelope encryption:

1. Compression: Gzip compression (configurable levels 1-9)
2. Envelope Encryption:
   • Generates random 256-bit AES key
   • Encrypts data with AES-256-GCM
   • Encrypts AES key with AWS KMS
   • Packages as: [key_len][encrypted_key][iv][tag][encrypted_data]

This approach:

• ✅ Supports files of unlimited size (no KMS 4KB limit)
• ✅ Achieves ~90% space savings through compression
• ✅ Provides authenticated encryption with GCM
• ✅ Uses AWS KMS for secure key management

Network Configuration

• API Timeouts: 10s connect, 30s read (configurable)
• Download Timeouts: 10s connect, unlimited read (for large files)
• Credential Validation: Fail-fast with STS on initialization

📚 Examples

See the examples/ directory for comprehensive usage examples:

• consumer_example.py - Download, subscribe, poll notifications
• producer_example.py - Upload, update, manage datasets

🧪 Testing

# Run all tests
pytest

# Run with coverage
pytest --cov=helix_connect --cov-report=html

# Run specific test suite
pytest tests/test_encryption_compression.py -v

# Run standalone pipeline test
python tests/test_pipeline_standalone.py

Test Results

The SDK includes comprehensive tests for the encryption/compression pipeline:

✓ test_compress_data - 90.9% compression on JSON data
✓ test_envelope_encryption_decryption - AES-256-GCM envelope format
✓ test_full_pipeline_compress_then_encrypt - End-to-end verification
✓ test_wrong_order_encrypt_then_compress - Proves old order was broken
✓ 10 tests total, all passing

⚙️ Configuration

Environment Variables

# Required
export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export HELIX_CUSTOMER_ID="your-customer-id"

# Optional
export HELIX_API_ENDPOINT="https://api-go.helix.tools"

Programmatic Configuration

consumer = HelixConsumer(
    aws_access_key_id="...",
    aws_secret_access_key="...",
    customer_id="...",
    api_endpoint="https://api-go.helix.tools",
    region="us-east-1",
)

compression_level (1=fastest, 9=best compression, default: 6) is not a constructor argument -- it's set per-upload on HelixProducer.upload_dataset( ..., compression_level=6).

🔐 Security Best Practices

1. Never commit credentials to version control
2. Use environment variables or AWS Secrets Manager
3. Rotate credentials regularly
4. Use IAM roles when running on AWS infrastructure
5. Validate data integrity after downloads
6. Monitor CloudWatch logs for anomalies

🐛 Error Handling

The SDK provides specific exceptions for different error scenarios:

from helix_connect.exceptions import (
    AuthenticationError,
    PermissionDeniedError,
    DatasetNotFoundError,
    RateLimitError,
    UploadError,
    DownloadError,
    HelixError  # Base exception
)

try:
    consumer.download_dataset(dataset_id="...", output_path="...")
except AuthenticationError:
    print("Invalid AWS credentials")
except PermissionDeniedError:
    print("No access to this dataset - subscribe first")
except DatasetNotFoundError:
    print("Dataset doesn't exist")
except RateLimitError as e:
    print(f"Rate limit exceeded - retry after {e.retry_after}s")
except HelixError as e:
    print(f"General error: {e}")

📊 Performance

Compression Benchmarks

Based on real-world testing with JSON data:

Note: Encrypting first (old broken code) resulted in ~0% compression!

Network Performance

• Chunked uploads: 8MB chunks for large files
• Parallel downloads: Multi-threaded for multiple datasets
• Progress callbacks: Real-time feedback without performance impact
• Connection pooling: Reuses HTTP connections for efficiency

🛠️ Development

Build & Validate

# Build package
python -m build

# Run build script (includes validation)
./scripts/build.sh

# Lint code
flake8 helix_connect/
black helix_connect/
mypy helix_connect/

Project Structure

helix-connect-sdk-python/
├── helix_connect/          # SDK source code
│   ├── __init__.py         # Package exports
│   ├── consumer.py         # Consumer API
│   ├── producer.py         # Producer API
│   ├── credentials.py      # STS session-credential provider
│   └── exceptions.py       # Custom exceptions
├── tests/                  # Test suite
│   ├── test_encryption_compression.py
│   └── test_pipeline_standalone.py
├── examples/               # Usage examples
│   ├── consumer_example.py
│   └── producer_example.py
├── scripts/                # Build scripts
│   └── build.sh
├── pyproject.toml          # Package configuration
└── README.md               # This file

🤝 Contributing

We welcome contributions! Please:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Run tests (pytest)
4. Commit changes (git commit -m 'Add amazing feature')
5. Push to branch (git push origin feature/amazing-feature)
6. Open a Pull Request

Code Standards

• Style: Follow PEP 8 (enforced by black)
• Types: Include type hints for all functions
• Tests: Maintain >80% coverage
• Docs: Update docstrings for public APIs

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🔗 Links

• Homepage: https://helix-connect.com
• Documentation: https://docs.helix-connect.com
• GitHub: https://github.com/helix-tools/helix-connect-sdk-python
• PyPI: https://pypi.org/project/helix-connect/
• Support: contact@helix.tools

📝 Changelog

v1.0.0 (2024-10-14)

✨ Features

• Initial release with Consumer, Producer, and Admin APIs
• AES-256-GCM envelope encryption for unlimited file sizes
• Compress-then-encrypt pipeline with ~90% space savings
• Real-time progress tracking for uploads/downloads
• SQS-based dataset update notifications
• Long-polling support with auto-download
• Comprehensive test suite (10 tests, all passing)

🔧 Improvements

• Network timeouts (API: 30s, Downloads: unlimited)
• Credential validation on initialization (fail-fast)
• Proper exception handling throughout
• Type hints for all public APIs

🐛 Bug Fixes

• Fixed KMS 4KB limit with envelope encryption
• Fixed compress-then-encrypt order (was reversed)
• Removed all emojis (encoding issues)
• Fixed bare except clauses

💬 Support

For questions, issues, or feature requests:

• GitHub Issues: https://github.com/helix-tools/helix-connect-sdk-python/issues
• Email: contact@helix.tools
• Documentation: https://docs.helix-connect.com

🙏 Acknowledgments

Built with:

• boto3 - AWS SDK for Python
• cryptography - Cryptographic recipes and primitives
• requests - HTTP library

Made with ❤️ by the Helix Tools team