DexPaprika DEX API Python SDK: on-chain liquidity and swap data client

from dexpaprika_sdk import DexPaprikaClient

WETH = "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"
USDC = "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"


def main():
    client = DexPaprikaClient()

    # Networks come back as a list of Network(id, display_name)
    networks = client.networks.list()
    ethereum = next(n for n in networks if n.id == "ethereum")
    print(f"{ethereum.display_name} is one of {len(networks)} networks")

    # Price lives under summary
    weth = client.tokens.get_details(network_id="ethereum", token_address=WETH)
    print(f"{weth.name} price: ${weth.summary.price_usd:,.2f}")

    # Pools for a token are served by pool search. Rows sit under results,
    # sorted by volume_usd_24h. There is no pair filter, so match the second
    # token client-side.
    pools = client.tokens.get_pools(
        network_id="ethereum",
        token_address=WETH,
        limit=20,
        order_by="volume_usd_24h",
        sort="desc",
    )
    weth_usdc = [p for p in pools.results if any(t.id == USDC for t in p.tokens)]

    print("Top WETH/USDC pools:")
    for pool in weth_usdc[:5]:
        print(f"{pool.dex_name}: ${pool.volume_usd_24h:,.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 429 (Too Many Requests), any 5xx
# (server errors), connection errors and timeouts. On a 429 it waits
# as long as the API's Retry-After header asks.

# 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.RequestException as e:
    print(f"Request failed: {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, max_pages=10):
    client = DexPaprikaClient()
    all_pools = []
    limit = 50
    cursor = None

    # Every page is one request against your allowance, so cap the walk
    for _ in range(max_pages):
        response = client.pools.list_by_network(
            network_id=network_id,
            limit=limit,
            sort="desc",
            order_by="volume_usd_24h",
            cursor=cursor,
        )

        all_pools.extend(response.results)

        # The search endpoint is cursor-paginated, not page-numbered
        if not response.has_next_page or not response.next_cursor:
            break
        cursor = response.next_cursor

        # Keyless allows 15 requests a minute, so pause 4 seconds between pages.
        # A free key doubles the rate: https://console.dexpaprika.com
        time.sleep(4)

    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_by_network(network_id="ethereum", sort="invalid_sort")
except ValueError as e:
    print(e)  # "sort must be one of: asc, desc"
    
try:
    # Invalid limit parameter
    client.pools.list_by_network(network_id="ethereum", 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