Python Examples

Custom implementation examples for Axol API using Python and standard libraries.

For production use, see the Python SDK which provides a complete client with error handling, retries, and type hints.

Setup

pip install requests websocket-client web3

Basic Client

Quick example using requests:

import requests

# Simple wrapper
client = requests.Session()
client.headers.update({'X-API-Key': 'your_key'})
base_url = 'https://api.axol.io/api/v1'

# Make requests
gas = client.get(f'{base_url}/gas/ethereum/current').json()
print(gas)

Gas Price Monitor

Monitor gas prices and alert when they drop below a threshold:

import requests
import time
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def monitor_gas(api_key: str, chain: str = 'ethereum', threshold: int = 50):
    """Alert when gas drops below threshold."""
    headers = {'X-API-Key': api_key}
    url = f'https://api.axol.io/api/v1/gas/{chain}/current'
    retry_delay = 30

    while True:
        try:
            response = requests.get(url, headers=headers, timeout=10)
            response.raise_for_status()

            gas = response.json()
            current = gas['standard']

            logger.info(f"[{datetime.now()}] Gas: {current} gwei")

            if current < threshold:
                logger.info(f"LOW GAS ALERT: {current} gwei")
                return current

        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:
                retry_after = int(e.response.headers.get('Retry-After', 60))
                logger.warning(f"Rate limited. Retrying in {retry_after}s")
                time.sleep(retry_after)
                continue
            elif e.response.status_code == 401:
                logger.error("Invalid API key")
                raise
            else:
                logger.error(f"HTTP error: {e}")
        except requests.exceptions.Timeout:
            logger.warning("Request timed out, retrying...")
        except requests.exceptions.RequestException as e:
            logger.error(f"Request failed: {e}")

        time.sleep(retry_delay)

# Usage
monitor_gas('your_api_key', chain='ethereum', threshold=50)

WebSocket Streaming

import json
import websocket

class AxolWebSocket:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.ws = None
    
    def connect(self):
        self.ws = websocket.WebSocketApp(
            f'wss://api.axol.io/api/v1/ws/connect?token={self.api_key}',
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        self.ws.run_forever()
    
    def on_open(self, ws):
        print("Connected to WebSocket")
        # Subscribe to channels
        ws.send(json.dumps({
            'type': 'subscribe',
            'channels': ['gas_prices_ethereum', 'blocks_ethereum']
        }))
    
    def on_message(self, ws, message):
        data = json.loads(message)
        print(f"Received: {data}")
    
    def on_error(self, ws, error):
        print(f"Error: {error}")
    
    def on_close(self, ws, close_status_code, close_msg):
        print("WebSocket closed")

Transaction Analysis

Analyze transactions using web3.py for gas cost calculations:

import requests
import logging
from web3 import Web3
from typing import Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class TransactionAnalysisError(Exception):
    """Custom exception for transaction analysis errors."""
    pass

def analyze_transaction(
    api_key: str,
    tx_hash: str,
    chain: str = 'ethereum',
    timeout: int = 10
) -> dict:
    """Analyze a transaction with gas cost calculation."""
    headers = {'X-API-Key': api_key}
    url = f'https://api.axol.io/api/v1/blockchain/{chain}/tx/{tx_hash}'

    try:
        response = requests.get(url, headers=headers, timeout=timeout)
        response.raise_for_status()
        tx = response.json()

        # Handle API error responses
        if not tx.get('success', True):
            raise TransactionAnalysisError(
                f"API error: {tx.get('error', 'Unknown error')}"
            )

        data = tx.get('data', tx)

        # Calculate gas cost using web3.py
        w3 = Web3()
        gas_used = data.get('gasUsed', 0)
        gas_price = data.get('gasPrice', 0)
        gas_cost_wei = gas_used * gas_price
        gas_cost_eth = w3.from_wei(gas_cost_wei, 'ether')

        return {
            'hash': tx_hash,
            'status': 'success' if data.get('status') else 'failed',
            'gas_cost_eth': float(gas_cost_eth),
            'confirmations': data.get('confirmations', 0),
            'block': data.get('blockNumber')
        }

    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 404:
            raise TransactionAnalysisError(f"Transaction not found: {tx_hash}")
        elif e.response.status_code == 429:
            raise TransactionAnalysisError("Rate limited - try again later")
        raise TransactionAnalysisError(f"HTTP error: {e}")
    except requests.exceptions.Timeout:
        raise TransactionAnalysisError(f"Request timed out for {tx_hash}")
    except requests.exceptions.RequestException as e:
        raise TransactionAnalysisError(f"Request failed: {e}")

def batch_analyze(
    api_key: str,
    tx_hashes: list,
    chain: str = 'ethereum'
) -> tuple[list, list]:
    """Batch analyze multiple transactions. Returns (results, errors)."""
    results = []
    errors = []

    for tx_hash in tx_hashes:
        try:
            result = analyze_transaction(api_key, tx_hash, chain)
            results.append(result)
            logger.info(f"Analyzed {tx_hash[:10]}...")
        except TransactionAnalysisError as e:
            errors.append({'hash': tx_hash, 'error': str(e)})
            logger.warning(f"Failed to analyze {tx_hash[:10]}...: {e}")

    logger.info(f"Completed: {len(results)} success, {len(errors)} failed")
    return results, errors

# Usage
results, errors = batch_analyze(
    'your_api_key',
    ['0x123...', '0x456...', '0x789...']
)