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
import os
import requests
from typing import Dict, Any
class AxolClient:
def __init__(self, api_key: str = None):
self.api_key = api_key or os.getenv('AXOL_API_KEY')
self.base_url = 'https://api.axol.io/v1'
self.session = requests.Session()
self.session.headers.update({
'X-API-Key': self.api_key,
'Content-Type': 'application/json'
})
def get(self, endpoint: str) -> Dict[str, Any]:
response = self.session.get(f'{self.base_url}{endpoint}')
response.raise_for_status()
return response.json()
def post(self, endpoint: str, data: Dict) -> Dict[str, Any]:
response = self.session.post(
f'{self.base_url}{endpoint}',
json=data
)
response.raise_for_status()
return response.json()
Gas Price Monitor
import time
from datetime import datetime
class GasMonitor:
def __init__(self, client: AxolClient):
self.client = client
def get_current_gas(self, chain: str = 'ethereum'):
return self.client.get(f'/gas/{chain}/current')
def monitor_gas(self, chain: str = 'ethereum', threshold: int = 50):
"""Alert when gas drops below threshold"""
while True:
gas = self.get_current_gas(chain)
current = gas['standard']
print(f"[{datetime.now()}] Gas: {current} gwei")
if current < threshold:
print(f"LOW GAS ALERT: {current} gwei")
return current
time.sleep(30)
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://ws.axol.io?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
from web3 import Web3
from decimal import Decimal
class TransactionAnalyzer:
def __init__(self, client: AxolClient):
self.client = client
self.w3 = Web3()
def analyze_tx(self, tx_hash: str, chain: str = 'ethereum'):
"""Analyze a transaction"""
tx = self.client.get(f'/blockchain/{chain}/tx/{tx_hash}')
# Calculate gas cost
gas_used = tx['gasUsed']
gas_price = tx['gasPrice']
gas_cost_wei = gas_used * gas_price
gas_cost_eth = self.w3.from_wei(gas_cost_wei, 'ether')
return {
'hash': tx_hash,
'status': 'success' if tx['status'] else 'failed',
'gas_cost_eth': float(gas_cost_eth),
'confirmations': tx['confirmations'],
'block': tx['blockNumber']
}
def batch_analyze(self, tx_hashes: list, chain: str = 'ethereum'):
"""Analyze multiple transactions"""
results = []
for tx_hash in tx_hashes:
try:
result = self.analyze_tx(tx_hash, chain)
results.append(result)
except Exception as e:
print(f"Error analyzing {tx_hash}: {e}")
return results