DexPaprika DEX API Python SDK — On‑chain Liquidity & Swap Data Client

from dexpaprika_sdk import DexPaprikaClient

def main():
    # Initialize client
    client = DexPaprikaClient()
    
    # Get Ethereum network details
    networks = client.networks.list()
    ethereum = next((n for n in networks if n.id == "ethereum"), None)
    print(f"Found {ethereum.display_name} with {ethereum.dexes_count} DEXes")
    
    # Get WETH token details
    weth = client.tokens.get_details(
        network_id="ethereum",
        token_address="0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"
    )
    print(f"{weth.name} price: ${weth.price_usd}")
    
    # Find WETH/USDC pools
    pools = client.tokens.get_pools(
        network_id="ethereum",
        token_address="0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",  # WETH
        limit=5, 
        order_by="volume_usd",
        sort="desc",
        address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"  # USDC
    )
    
    # Show top pools
    print('Top WETH/USDC pools:')
    for pool in pools.pools:
        print(f"{pool.dex_name}: ${pool.volume_usd:,.2f} 24h volume")

if __name__ == "__main__":
    main()

from dexpaprika_sdk import DexPaprikaClient
import requests

# Basic error handling
try:
    client = DexPaprikaClient()
    token = client.tokens.get_details("ethereum", "0xinvalidaddress")
except Exception as e:
    if "404" in str(e):
        print("Token not found")
    elif "429" in str(e):
        print("Rate limit exceeded")
    else:
        print(f"An error occurred: {e}")

# The SDK automatically retries on these status codes:
# 408 (Request Timeout), 429 (Too Many Requests),
# 500, 502, 503, 504 (Server Errors)

# Custom retry configuration
client = DexPaprikaClient(
    max_retries=4,  # Number of retry attempts (default: 4)
    backoff_times=[0.1, 0.5, 1.0, 5.0]  # Backoff times in seconds
)

# All API requests will now use these retry settings
try:
    networks = client.networks.list()
except requests.exceptions.RetryError as e:
    print(f"Failed after multiple retries: {e}")

from dexpaprika_sdk import DexPaprikaClient
import time

# The SDK includes built-in caching by default
# Demonstration of cache behavior

# Create a regular client with default cache (5 minutes TTL)
client = DexPaprikaClient()

# First call - hits the API
start_time = time.time()
networks = client.networks.list()
first_call_time = time.time() - start_time
print(f"First call (API): {len(networks)} networks, took {first_call_time:.4f}s")

# Second call - served from cache (much faster)
start_time = time.time()
networks = client.networks.list()
second_call_time = time.time() - start_time
print(f"Second call (cached): {len(networks)} networks, took {second_call_time:.4f}s")
print(f"Cache speedup: {first_call_time / second_call_time:.1f}x")

# You can skip the cache when you need fresh data
fresh_networks = client.networks._get("/networks", skip_cache=True)

# Clear the entire cache
client.clear_cache()

# Clear cache only for specific endpoints
client.clear_cache(endpoint_prefix="/networks")

# Different types of data have different cache durations:
# - Network data: 24 hours
# - Pool data: 5 minutes
# - Token data: 10 minutes
# - Statistics: 15 minutes
# - Other data: 5 minutes (default)

from dexpaprika_sdk import DexPaprikaClient
import time

def fetch_all_pools(network_id):
    client = DexPaprikaClient()
    all_pools = []
    page = 0
    limit = 50
    has_more = True
    
    while has_more:
        response = client.pools.list_by_network(
            network_id=network_id, 
            page=page, 
            limit=limit, 
            sort="desc", 
            order_by="volume_usd"
        )
        
        all_pools.extend(response.pools)
        
        # Check if there are more pages
        has_more = len(response.pools) == limit
        page += 1
        
        # Adding a small delay to avoid hitting rate limits
        time.sleep(0.1)
    
    print(f"Fetched a total of {len(all_pools)} pools on {network_id}")
    return all_pools

# Usage example
ethereum_pools = fetch_all_pools("ethereum")

from dexpaprika_sdk import DexPaprikaClient

# The SDK automatically validates parameters before making API requests
client = DexPaprikaClient()

# Invalid parameter examples will raise helpful error messages
try:
    # Invalid network ID
    client.pools.list_by_network(network_id="", limit=5)
except ValueError as e:
    print(e)  # "network_id is required"
    
try:
    # Invalid sort parameter
    client.pools.list(sort="invalid_sort")
except ValueError as e:
    print(e)  # "sort must be one of: asc, desc"
    
try:
    # Invalid limit parameter
    client.pools.list(limit=500)
except ValueError as e:
    print(e)  # "limit must be at most 100"

from dexpaprika_sdk import DexPaprikaClient
from dexpaprika_sdk.models import PoolDetails  # Type hint example

client = DexPaprikaClient()

# Get pool details
pool: PoolDetails = client.pools.get_details(
    network_id="ethereum",
    pool_address="0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640"  # USDC/WETH Uniswap v3 pool
)

# Access pool properties with type checking and auto-completion
print(f"Pool: {pool.tokens[0].symbol}/{pool.tokens[1].symbol}")
print(f"Volume (24h): ${pool.day.volume_usd:.2f}")
print(f"Transactions (24h): {pool.day.txns}")
print(f"Price: ${pool.last_price_usd:.4f}")

# Time interval data is available for multiple timeframes
print(f"1h price change: {pool.hour1.last_price_usd_change:.2f}%")
print(f"24h price change: {pool.day.last_price_usd_change:.2f}%")

# All API responses are converted to typed Pydantic models
# This provides automatic validation, serialization/deserialization,
# and IDE auto-completion support through Python type hints