Turbo Python SDK

A Python SDK for interacting with the ArDrive Turbo Upload service, supporting both Ethereum and Arweave signers for permanent data storage on Arweave.

Get Started

Installation

pip install turbo-sdk

Quick Start

Ethereum Usage

from turbo_sdk import Turbo, EthereumSigner

# Create Ethereum signer
signer = EthereumSigner("0x1234567890abcdef...")  # Your private key

# Create Turbo client
turbo = Turbo(signer, network="mainnet")

# Upload data
result = turbo.upload(b"Hello, Turbo!", tags=[
    {"name": "Content-Type", "value": "text/plain"},
    {"name": "App-Name", "value": "MyApp"}
])

print(f"✅ Uploaded! TX ID: {result.id}")
print(f"🌐 View at: https://arweave.net/{result.id}")

Arweave Usage

import json
from turbo_sdk import Turbo, ArweaveSigner

# Load Arweave wallet (JWK format)
with open("test-wallet.json") as f:
    jwk = json.load(f)

# Create Arweave signer
signer = ArweaveSigner(jwk)

# Create Turbo client
turbo = Turbo(signer, network="mainnet")

# Upload data
result = turbo.upload(b"Hello from Arweave!", tags=[
    {"name": "Content-Type", "value": "text/plain"}
])

print(f"✅ Uploaded! URI: ar://{result.id}")

APIs

Core Classes

Turbo(signer, network="mainnet", upload_url=None, payment_url=None)

Main client for interacting with Turbo services.

Parameters:

• signer: Either EthereumSigner or ArweaveSigner instance
• network: "mainnet" or "testnet" (default: "mainnet")
• upload_url: Optional custom upload service URL (overrides network default)
• payment_url: Optional custom payment service URL (overrides network default)

# Using default URLs (mainnet)
turbo = Turbo(signer)

# Using testnet
turbo = Turbo(signer, network="testnet")

# Using custom URLs
turbo = Turbo(signer, upload_url="https://my-upload-service.example.com")

Methods:

upload(data=None, tags=None, on_progress=None, chunking=None, data_size=None, stream_factory=None) -> TurboUploadResponse

Upload data to the Turbo datachain. Supports both small files (single request) and large files (chunked multipart upload).

Parameters:

• data: Data to upload (bytes or file-like BinaryIO object)
• tags: Optional list of metadata tags
• on_progress: Optional callback (processed_bytes, total_bytes) -> None
• chunking: Optional ChunkingParams for upload configuration
• data_size: Required when data is a file-like object or when using stream_factory
• stream_factory: Optional callable that returns a fresh BinaryIO stream each time it's called. Use this for non-seekable streams or when you want to avoid loading the entire file into memory.

# Simple upload
result = turbo.upload(
    data=b"Your data here",
    tags=[
        {"name": "Content-Type", "value": "application/json"},
        {"name": "App-Name", "value": "MyApp"}
    ]
)

Returns: TurboUploadResponse

@dataclass
class TurboUploadResponse:
    id: str                        # Transaction ID
    owner: str                     # Owner address
    data_caches: List[str]         # Cache endpoints
    fast_finality_indexes: List[str] # Fast finality indexes
    winc: str                      # Winston credits cost

Large File Uploads with Progress

For files >= 5 MiB, the SDK automatically uses chunked multipart uploads. Use stream_factory to avoid loading the entire file into memory. A factory is needed because the stream is consumed twice — once for signing and once for uploading — so the SDK calls it each time to get a fresh stream.

import os

def on_progress(processed: int, total: int):
    pct = (processed / total) * 100
    print(f"Upload progress: {pct:.1f}%")

file_path = "large-video.mp4"

result = turbo.upload(
    stream_factory=lambda: open(file_path, "rb"),
    data_size=os.path.getsize(file_path),
    tags=[{"name": "Content-Type", "value": "video/mp4"}],
    on_progress=on_progress,
)

Chunking Configuration

Use ChunkingParams to customize chunked upload behavior:

from turbo_sdk import ChunkingParams

result = turbo.upload(
    data=large_data,
    chunking=ChunkingParams(
        chunk_size=10 * 1024 * 1024,  # 10 MiB chunks (default: 5 MiB)
        max_chunk_concurrency=3,             # Parallel chunk uploads (default: 1)
        chunking_mode="auto",                # "auto", "force", or "disabled"
    ),
    on_progress=lambda p, t: print(f"{p}/{t} bytes"),
)

ChunkingParams options:

• chunk_size: Chunk size in bytes (5-500 MiB, default: 5 MiB)
• max_chunk_concurrency: Number of parallel chunk uploads (default: 1)
• chunking_mode:
  • "auto" (default): Use chunked upload for files >= 5 MiB
  • "force": Always use chunked upload
  • "disabled": Always use single request upload

get_balance(address=None) -> TurboBalanceResponse

Get winston credit balance. Uses signed request for authenticated balance check when no address specified.

# Check your own balance (signed request)
balance = turbo.get_balance()
print(f"Balance: {balance.winc} winc")

# Check another address (no signature needed)
other_balance = turbo.get_balance("0x742d35Cc6635C0532925a3b8C17af2e95C5Aca4A")
print(f"Other balance: {other_balance.winc} winc")

Returns: TurboBalanceResponse

@dataclass
class TurboBalanceResponse:
    winc: str                      # Available winston credits
    controlled_winc: str           # Controlled amount
    effective_balance: str         # Effective balance including shared credits

get_upload_price(byte_count) -> int

Get the cost to upload data of a specific size.

cost = turbo.get_upload_price(1024)  # Cost for 1KB
print(f"Upload cost: {cost} winc")

Signers

EthereumSigner(private_key)

Ethereum signer using ECDSA signatures.

Parameters:

• private_key (str): Hex private key with or without 0x prefix

signer = EthereumSigner("0x1234567890abcdef...")

ArweaveSigner(jwk)

Arweave signer using RSA-PSS signatures.

Parameters:

• jwk (dict): Arweave wallet in JWK format

signer = ArweaveSigner({
    "kty": "RSA",
    "n": "...",
    "e": "AQAB",
    "d": "...",
    # ... other JWK fields
})

Signer Methods

Both signers provide:

get_wallet_address() -> str

Get the wallet address for the signer.

address = signer.get_wallet_address()
print(f"Wallet address: {address}")

create_signed_headers() -> dict

Create signed headers for authenticated API requests.

headers = signer.create_signed_headers()

Exceptions

The SDK provides specific exceptions for error handling:

from turbo_sdk import UnderfundedError, ChunkedUploadError

try:
    result = turbo.upload(large_data)
except UnderfundedError:
    print("Insufficient balance - please top up your account")
except ChunkedUploadError as e:
    print(f"Upload failed: {e}")

Exception types:

• ChunkedUploadError: Base exception for chunked upload failures
• UnderfundedError: Account has insufficient balance (HTTP 402)
• UploadValidationError: Upload validation failed (INVALID status)
• UploadFinalizationError: Finalization timed out or failed

Developers

Setup

1. Create a virtual environment:

python -m venv venv
source venv/bin/activate

2. Install dependencies:

pip install -e ".[dev]"

3. Run tests:

pytest

With coverage

pytest --cov=turbo_sdk

4. Lint and format:

black turbo_sdk tests
flake8 turbo_sdk tests

5. Run performance benchmarks (requires funded wallet):

export TURBO_TEST_WALLET=/path/to/wallet.json
export TURBO_UPLOAD_URL=https://upload.ardrive.dev  # optional, defaults to testnet
pytest -m performance -v -s

The test suite includes comprehensive unit tests for all components. Performance tests measure real upload throughput against the Turbo service.

Publishing

Releases are published to PyPI via the GitHub Actions workflow at .github/workflows/release.yml. It runs on release events or can be triggered manually via workflow_dispatch.

There is no automated versioning. Before publishing, update the version field in pyproject.toml to reflect the new release:

[project]
version = "0.0.5"

Steps to release:

1. Merge feature branches into alpha.
2. Review the commits and update the version field in pyproject.toml accordingly.
3. Push to the alpha branch.
4. Manually run the release workflow at .github/workflows/release.yml via workflow_dispatch.

The workflow runs tests across Python 3.8-3.12, builds the package, and publishes to PyPI using trusted OIDC publishing.

To publish locally instead:

pip install build twine
python -m build
twine check dist/*
twine upload dist/*

Acknowledgments

This package leverages implementations from the Irys Python SDK for ANS-104 DataItem format and cryptographic operations. Special thanks to the Irys team for their work on permanent data storage standards.

License

MIT License - see LICENSE for details.