Python SDK

import asyncio
from omne_sdk import OmneClient, Network

async def main():
    # Initialize client
    client = OmneClient(
        network=Network.TESTNET,
        endpoint='https://testnet-rpc.omne.foundation',
        timeout=30.0  # Optional timeout in seconds
    )
    
    # Check connection
    try:
        status = await client.get_network_status()
        print(f'Connected to: {status.network_id}')
        print(f'Latest block: {status.latest_block}')
    except Exception as error:
        print(f'Connection failed: {error}')
    finally:
        await client.close()

# Run the async function
asyncio.run(main())

from omne_sdk import Account, Wallet

# Create new account
account = Account.generate()
print(f'Address: {account.address}')
print(f'Public key: {account.public_key}')

# Import from private key
imported_account = Account.from_private_key('your-private-key-hex')

# Create wallet for multiple accounts
wallet = Wallet()
wallet.add_account(account)
wallet.add_account(imported_account)

# Get account by address
found_account = wallet.get_account('omne1...')

# Save/load wallet (encrypted)
wallet.save_to_file('wallet.json', password='secure-password')
loaded_wallet = Wallet.load_from_file('wallet.json', password='secure-password')

from omne_sdk import TransferTransaction
from omne_sdk.exceptions import TransactionError

async def send_transfer(client, account):
    try:
        # Create transfer transaction
        tx = TransferTransaction(
            from_address=account.address,
            to_address='omne1recipient-address',
            amount='1000000',  # 1 OMNE in micro-OMNE
            fee='auto',  # Auto fee estimation
            memo='Payment for services'  # Optional memo
        )
        
        # Sign transaction
        signed_tx = account.sign(tx)
        
        # Broadcast to network
        result = await client.broadcast(signed_tx)
        print(f'Transaction hash: {result.hash}')
        
        # Wait for confirmation
        receipt = await client.wait_for_transaction(result.hash)
        print(f'Confirmed in block: {receipt.block_number}')
        
        return receipt
        
    except TransactionError as e:
        print(f'Transaction failed: {e.message}')
        print(f'Error code: {e.code}')
        raise

from omne_sdk import BatchTransaction
import asyncio

async def process_payments(client, account, payments):
    batch = BatchTransaction()
    
    # Add multiple transfers to batch
    for payment in payments:
        batch.add_transfer(
            to_address=payment['to'],
            amount=payment['amount'],
            memo=payment.get('memo', '')
        )
    
    # Sign and broadcast batch
    signed_batch = account.sign(batch)
    result = await client.broadcast(signed_batch)
    
    print(f'Batch transaction hash: {result.hash}')
    
    # Wait for all transactions to confirm
    receipt = await client.wait_for_transaction(result.hash)
    
    return receipt

# Example usage
payments = [
    {'to': 'omne1recipient1', 'amount': '500000', 'memo': 'Order #1'},
    {'to': 'omne1recipient2', 'amount': '300000', 'memo': 'Order #2'},
    {'to': 'omne1recipient3', 'amount': '200000', 'memo': 'Order #3'},
]

receipt = await process_payments(client, account, payments)

# Monitor transaction status
async def monitor_transaction(client, tx_hash: str):
    tx = await client.get_transaction(tx_hash)
    
    print(f'Status: {tx.status}')
    print(f'Block: {tx.block_number}')
    print(f'Confirmations: {tx.confirmations}')
    print(f'Gas used: {tx.gas_used}')
    
    if tx.status == 'failed':
        print(f'Error: {tx.error}')

# Subscribe to account transactions
async def monitor_account(client, address: str):
    async for tx in client.subscribe_transactions(address):
        print(f'New transaction: {tx.hash}')
        print(f'Amount: {tx.amount}')
        print(f'Direction: {tx.direction}')  # 'in' or 'out'

from omne_sdk import Contract

async def deploy_contract(client, account):
    contract = Contract(
        bytecode=contract_bytecode,
        abi=contract_abi,
        constructor_args=['initial_value', 1000]
    )
    
    deployment = await contract.deploy(account)
    print(f'Contract address: {deployment.address}')
    
    return deployment

# Load contract from file
contract = Contract.from_file('contract.json')

# Connect to existing contract
contract = Contract(
    address='omne1contract-address',
    abi=contract_abi
)

# Read contract state (free)
balance = await contract.call('get_balance', ['omne1user-address'])
print(f'Balance: {balance}')

# Write contract state (requires transaction)
result = await contract.execute(
    account,
    'transfer',
    ['omne1recipient', '1000000'],
    fee='auto'
)

print(f'Transaction hash: {result.hash}')

# Listen to contract events
async def listen_to_events(contract, account):
    # Listen to all Transfer events
    async for event in contract.subscribe_events('Transfer'):
        print(f'Transfer event:')
        print(f'From: {event.from_address}')
        print(f'To: {event.to_address}')
        print(f'Amount: {event.amount}')

# Listen with filters
async def listen_to_my_transfers(contract, account):
    async for event in contract.subscribe_events(
        'Transfer',
        filters={'to': account.address}
    ):
        print(f'Received transfer: {event.amount} OMNE')

from omne_sdk.models import (
    Transaction, 
    TransactionReceipt, 
    Block, 
    Account as AccountModel,
    Balance
)
from pydantic import BaseModel
from typing import Optional

# All SDK models are Pydantic BaseModel subclasses
class CustomTransaction(BaseModel):
    hash: str
    from_address: str
    to_address: str
    amount: str
    fee: str
    memo: Optional[str] = None
    
    class Config:
        # Enable validation
        validate_assignment = True
        # Use enum values
        use_enum_values = True

# Example usage
tx_data = {
    'hash': '0x...',
    'from_address': 'omne1...',
    'to_address': 'omne1...',
    'amount': '1000000',
    'fee': '1000'
}

tx = CustomTransaction(**tx_data)
print(tx.json(indent=2))  # Pretty JSON output

# Query blockchain with typed results
async def query_account_info(client, address: str):
    # Get account balance
    balance: Balance = await client.get_balance(address)
    print(f'Balance: {balance.amount} {balance.denom}')
    
    # Get account details
    account: AccountModel = await client.get_account(address)
    print(f'Sequence: {account.sequence}')
    print(f'Account number: {account.account_number}')
    
    # Get transaction history with pagination
    from omne_sdk.models import TransactionHistory
    
    history: TransactionHistory = await client.get_transaction_history(
        address,
        limit=20,
        offset=0
    )
    
    print(f'Total transactions: {history.total}')
    for tx in history.transactions:
        print(f'{tx.hash}: {tx.amount} OMNE')
    
    return history

import asyncio
from omne_sdk import OmneClient

async def concurrent_queries(client, addresses):
    # Run multiple balance queries concurrently
    tasks = [
        client.get_balance(address) 
        for address in addresses
    ]
    
    balances = await asyncio.gather(*tasks)
    
    for address, balance in zip(addresses, balances):
        print(f'{address}: {balance.amount} OMNE')
    
    return balances

# Context manager for automatic cleanup
async def main():
    async with OmneClient(
        network=Network.TESTNET,
        endpoint='https://testnet-rpc.omne.foundation'
    ) as client:
        addresses = ['omne1...', 'omne2...', 'omne3...']
        balances = await concurrent_queries(client, addresses)

async def stream_new_blocks(client):
    async for block in client.subscribe_new_blocks():
        print(f'New block: {block.height}')
        print(f'Transactions: {len(block.transactions)}')
        print(f'Timestamp: {block.timestamp}')
        
        # Process transactions in block
        for tx in block.transactions:
            if tx.amount > 1000000:  # Large transactions
                print(f'Large transaction: {tx.hash} - {tx.amount} OMNE')

async def monitor_multiple_accounts(client, addresses):
    # Create async generators for each address
    streams = [
        client.subscribe_transactions(address) 
        for address in addresses
    ]
    
    # Merge all streams
    async def merged_stream():
        tasks = [stream.__anext__() for stream in streams]
        while tasks:
            done, pending = await asyncio.wait(
                tasks, 
                return_when=asyncio.FIRST_COMPLETED
            )
            
            for task in done:
                tx = await task
                yield tx
                # Restart the completed stream
                stream_idx = tasks.index(task)
                tasks[stream_idx] = streams[stream_idx].__anext__()
    
    async for tx in merged_stream():
        print(f'Transaction on {tx.address}: {tx.hash}')

from omne_sdk.exceptions import (
    OmneException,
    NetworkException,
    TransactionException,
    ValidationException,
    AuthenticationException
)

async def robust_transfer(client, account, to_address, amount):
    try:
        result = await client.transfer(
            from_address=account.address,
            to_address=to_address,
            amount=amount
        )
        return result
        
    except ValidationException as e:
        print(f'Validation failed: {e.message}')
        print(f'Field: {e.field}')
        print(f'Value: {e.value}')
        
    except NetworkException as e:
        print(f'Network error: {e.message}')
        print(f'Status code: {e.status_code}')
        print(f'Response: {e.response}')
        
    except TransactionException as e:
        print(f'Transaction failed: {e.message}')
        print(f'Code: {e.code}')
        print(f'Gas used: {e.gas_used}')
        
    except AuthenticationException as e:
        print(f'Authentication failed: {e.message}')
        
    except OmneException as e:
        print(f'General Omne error: {e.message}')
        
    except Exception as e:
        print(f'Unexpected error: {e}')

import asyncio
from typing import Callable, Any
from omne_sdk.exceptions import NetworkException

async def retry_with_backoff(
    func: Callable,
    max_retries: int = 3,
    base_delay: float = 1.0,
    backoff_factor: float = 2.0,
    *args,
    **kwargs
) -> Any:
    """Retry function with exponential backoff."""
    last_exception = None
    
    for attempt in range(max_retries + 1):
        try:
            return await func(*args, **kwargs)
        except NetworkException as e:
            last_exception = e
            if attempt < max_retries:
                delay = base_delay * (backoff_factor ** attempt)
                print(f'Attempt {attempt + 1} failed, retrying in {delay}s...')
                await asyncio.sleep(delay)
            else:
                raise
        except Exception as e:
            # Don't retry for non-network errors
            raise
    
    raise last_exception

# Usage example
async def reliable_balance_check(client, address):
    return await retry_with_backoff(
        client.get_balance,
        max_retries=3,
        address=address
    )

import pytest
from omne_sdk.testing import MockOmneClient, MockAccount
from omne_sdk.models import Balance, Transaction

@pytest.fixture
async def mock_client():
    client = MockOmneClient()
    
    # Set up mock responses
    client.mock_balance('omne1test-address', Balance(
        amount='1000000',
        denom='uomne'
    ))
    
    client.mock_transaction_result(Transaction(
        hash='0x123...',
        status='success',
        block_number=12345
    ))
    
    return client

@pytest.mark.asyncio
async def test_transfer(mock_client):
    account = MockAccount.generate()
    
    result = await mock_client.transfer(
        from_address=account.address,
        to_address='omne1recipient',
        amount='500000'
    )
    
    assert result.hash == '0x123...'
    assert result.status == 'success'

@pytest.mark.asyncio 
async def test_balance_check(mock_client):
    balance = await mock_client.get_balance('omne1test-address')
    
    assert balance.amount == '1000000'
    assert balance.denom == 'uomne'

import pytest
import asyncio
from omne_sdk import OmneClient, Account, Network

class TestOmneIntegration:
    @pytest.fixture
    async def client(self):
        client = OmneClient(
            network=Network.TESTNET,
            endpoint='https://testnet-rpc.omne.foundation'
        )
        yield client
        await client.close()
    
    @pytest.fixture
    def test_account(self):
        # Use a test account with known private key
        return Account.from_private_key('test-private-key')
    
    @pytest.mark.asyncio
    async def test_network_connection(self, client):
        status = await client.get_network_status()
        assert status.network_id.startswith('omne-testnet')
        assert status.latest_block > 0
    
    @pytest.mark.asyncio
    async def test_account_balance(self, client, test_account):
        balance = await client.get_balance(test_account.address)
        assert balance.denom == 'uomne'
        assert int(balance.amount) >= 0
    
    @pytest.mark.asyncio
    async def test_small_transfer(self, client, test_account):
        # Small test transfer
        result = await client.transfer(
            from_address=test_account.address,
            to_address='omne1testrecipient',
            amount='1000'  # 0.001 OMNE
        )
        
        assert result.hash
        receipt = await client.wait_for_transaction(result.hash)
        assert receipt.status == 'success'

from omne_sdk import OmneClient, Network
from omne_sdk.config import ClientConfig

# Full configuration example
config = ClientConfig(
    # Network settings
    network=Network.MAINNET,
    endpoint='https://rpc.omne.foundation',
    
    # Request settings
    timeout=30.0,
    max_retries=3,
    retry_delay=1.0,
    
    # WebSocket settings
    ws_endpoint='wss://ws.omne.foundation',
    ws_reconnect=True,
    ws_reconnect_delay=5.0,
    
    # Fee settings
    default_fee='1000',
    fee_multiplier=1.2,
    
    # Gas settings
    default_gas=200000,
    gas_multiplier=1.5,
    
    # HTTP headers
    headers={
        'User-Agent': 'MyApp/1.0.0',
        'X-Custom-Header': 'value'
    }
)

client = OmneClient(config=config)

import os
from omne_sdk import OmneClient
from omne_sdk.config import load_config_from_env

# Load configuration from environment variables
# OMNE_NETWORK=testnet
# OMNE_ENDPOINT=https://testnet-rpc.omne.foundation
# OMNE_TIMEOUT=30
# OMNE_PRIVATE_KEY=your-private-key

config = load_config_from_env()
client = OmneClient(config=config)

# Or use environment variables directly
client = OmneClient(
    network=os.getenv('OMNE_NETWORK', 'testnet'),
    endpoint=os.getenv('OMNE_ENDPOINT'),
    timeout=float(os.getenv('OMNE_TIMEOUT', '30'))
)