Pagination

{
  "status": 200,
  "title": "OK",
  "description": "Successful request",
  "payload": {
    "metadata": {
      "next": "https://web3api.io/api/v2/defi/lending/compoundv2/assets/ZRX?cursor=N4IgRgNg9gxg1jAFgQwJYDsCSATEAuEARhgE5SA2EgBnIFYAzGbY%2BkAGhAAcAnKAF1hQIOfCBhQAtpygBXdNgBuAJnYgJybnACmffOhkQIHDDAgzsWgIKGAKt2ToAzshh9UUdAFEFW9H0f49MgQjlocbhJaAGJQ3Oq6BIgy6ugASlrI2MiQWqrYqNxaru7oopYAygDCqo58GnwAIsh8uXiE5OSEACwkXYQA7AAcVCMcvthNLfjtnSTUXSOLHI6oAF6thKMgFj7QnFrcALJQFoHBoRz0sfGiAFaOHqqcyADmGxyxqC8YweV13I1mhs6CQAMyEJSLJYgT7fdDBTzySbA2bzKFUDjIRyhPgiAhUAAeWj62CUXUGJFo-UI2FBSiUhApZPIWSCtC0YAshHoXUoJB5JEGqggWL4NlQkVqyCkohm3V6AyFHBFtQAMlAXph5FoCdNwQBfIA"
    },
    ...
  }
}

import os
from datetime import datetime
from dataclasses import dataclass, field

from endpoint_timerange_handler import EndpointTimeRangeHandler, Endpoint
from endpoint_caller_v2 import EndpointCaller

def http_ok_next_page_url_extractor(page):
    """
    Function that extracts the next page url from the current page of data

    Parameters
    ----------
    page: dict

    Returns
    -------
    next_page_url: str
    """
    if 'payload' in page and page['payload'] is not None:
        payload = page['payload']
        if 'metadata' in payload and payload['metadata'] is not None:
            metadata = payload['metadata']
            if 'next' in metadata and metadata['next'] is not None:
                next_page_url = metadata['next']
                return next_page_url
    return ""

@dataclass(kw_only=True)
class DEXTradesHistorical(Endpoint):
    poolAddress: str
    path_template: str = field(default='/market/defi/trades/{}/historical')
    max_interval_in_seconds: int = field(default=2592000) #30 days * 24 hrs * 60 min * 60 seconds
    
    def format_path(self) -> str:
        return self.path_template.format(self.poolAddress)

"""
You can implement similar dataclasses for other endpoints as needed.

`max_interval_in_seconds` is the maximum supported range for endDate and startDate.

The maximum range for each endpoint can be found in the documentation.
"""

def get_api_responses(start_date: datetime, end_date: datetime, endpoint: DEXTradesHistorical) -> None:
    """
    Get all pages of data between `start_date` and `end_date`.
    """
    endpoint_caller = EndpointCaller(os.getenv('PRODUCTION_API_KEY'))
    endpoint_timerange_handler = EndpointTimeRangeHandler(endpoint_caller)
    for page in endpoint_timerange_handler.get_data_for_timerange(
        start_date,
        end_date,
        endpoint,
        http_ok_next_page_url_extractor
    ):
        response = page.data
        print(f"Timestamp of first entry in the page: {response['payload']['data'][0][1]}")
        # print('uncomment here below as needed')
        # print(page.data)
        # print(page.duration_seconds)

def call_dex_trades_historical() -> None:
    """
    Example configuration of an endpoint to be called.

    This example demonstrates a way to query a time range larger than the endpoint's max supported range (endDate - startDate).
    """
    start_date = "2022-01-01T00:00:00"
    end_date = "2023-01-01T00:00:00"
    start_date_as_dt = datetime.fromisoformat(start_date).replace(microsecond=0)
    end_date_as_dt = datetime.fromisoformat(end_date).replace(microsecond=0)
    poolAddress = '0xcbcdf9626bc03e24f779434178a73a0b4bad62ed' # WBTC/ETH 0.3%
    exchange = 'uniswapv3'

    dex_trades_historical = DEXTradesHistorical(poolAddress=poolAddress)
    dex_trades_historical.add_query_parameter('exchange', exchange)
    get_api_responses(start_date_as_dt, end_date_as_dt, dex_trades_historical)

if __name__ == "__main__":
    call_dex_trades_historical()

from datetime import datetime
from datetime import timedelta
from dataclasses import dataclass, field

from endpoint_caller_v2 import EndpointCaller

@dataclass(kw_only=True)
class Endpoint:
    """
    This is a parent class that should be inherited and implemented for specific endpoints.
    """
    query: dict = field(default_factory=dict)
    headers: dict = field(default_factory=dict)

    def add_query_parameter(self, parameter_name: str, parameter_value) -> None:
        if parameter_name is not None and len(parameter_name) > 0:
            self.query[parameter_name] = parameter_value
    
    def format_path(self) -> str:
        """
        Child classes must implement this function.
        """
        return ""
    
    def add_header(self, header_name: str, header_value: str) -> None:
        if header_name is not None and len(header_name) > 0:
            self.headers[header_name] = header_value

class EndpointTimeRangeHandler:
    def __init__(self, endpoint_caller: EndpointCaller) -> None:
        self.endpoint_caller = endpoint_caller

    def get_data(self,
                 start_date: datetime, 
                 end_date: datetime, 
                 endpoint: Endpoint,
                 http_ok_next_page_url_extractor):

        endpoint.add_query_parameter('startDate', str(start_date).replace(" ", "T"))
        endpoint.add_query_parameter('endDate', str(end_date).replace(" ", "T"))
        
        yield from self.endpoint_caller.call_endpoint_and_get_all_pages(
            endpoint.format_path(), 
            endpoint.query, 
            endpoint.headers, 
            http_ok_next_page_url_extractor
        )

    def get_data_for_timerange(self,
                               start_date: datetime, 
                               end_date: datetime, 
                               endpoint: Endpoint,
                               http_ok_next_page_url_extractor):
        """
        Given an arbitrarily large timerange denoted by `end_date` and `start_date`, this
        function will call the endpoint continuously by breaking up the requested time range into chunks.
        
        The chunks do not exceed the single request maximum range (endDate - startDate) for the specific endpoint.
        
        Parameters
        ----------
        start_date: datetime
        end_date: datetime
        endpoint: Endpoint
            An implementation of Endpoint for a specific class
        http_ok_next_page_url_extractor: function
            The function that extracts the next page url from a given API response

        Returns
        -------
        page: AmberdataResponse
            A single page is yielded for each call to this generator function
        """

        duration = end_date - start_date
        intervals = duration.total_seconds()/endpoint.max_interval_in_seconds
        hours = duration.total_seconds()/3600
        print(f"Getting {hours} hours of data (# of intervals: {intervals})")

        start_date_copy = start_date
        timerange_stack = []
        while start_date_copy + timedelta(seconds=endpoint.max_interval_in_seconds) <= end_date:
            intermediate_end_date = start_date_copy + timedelta(seconds=endpoint.max_interval_in_seconds)
            timerange_stack.append((start_date_copy, intermediate_end_date))
            start_date_copy = intermediate_end_date
        timerange_stack.append((start_date_copy, end_date))
        timerange_stack.reverse()

        while len(timerange_stack) > 0:
            timerange = timerange_stack.pop()
            print(f"Retrieving data from {str(timerange[0])} to {str(timerange[1])}")
            yield from self.get_data(timerange[0], timerange[1], endpoint, http_ok_next_page_url_extractor)
    

import time
import requests
import json

PRODUCTION_BASE_URL = "https://web3api.io/api/v2"
PRODUCTION_BASE_URL_FORMATTABLE = "https://web3api.io/api/v2{}"

class AmberdataResponse:
    def __init__(self, data, status, duration_seconds, request_url):
        self.data = data
        self.status = status
        self.duration_seconds = duration_seconds
        self.request_url = request_url
        self.attempts = 0

    def increment_attempt(self, by=None):
        if by is not None:
            self.attempts += by
        else:
            self.attempts += 1

"""
Type Hints
"""
AmberdataResponseStack = list[AmberdataResponse]

class EndpointCaller:
    def __init__(self, amberdata_api_key):
        self.x_api_key = amberdata_api_key

    def call_endpoint_and_get_data_as_json(self, path: str, query: dict, headers: dict):
        headers_with_api_key = {
            **headers,
            'x-api-key': self.x_api_key
        }

        if path.startswith(PRODUCTION_BASE_URL):
            """
            When the next page URL is pre-formed and can be used as-is.
            """
            full_api_url = path
        else:    
            full_api_url = PRODUCTION_BASE_URL_FORMATTABLE.format(path)
        
        start_time = time.time()
        request = None
        try: 
            request = requests.get(full_api_url, params=query, headers=headers_with_api_key)
            print("Making HTTP call for: {}".format(request.url))
            end_time = time.time()
            duration = end_time - start_time
            if request.status_code == 200:
                json_data = request.json()
                return AmberdataResponse(json_data, 200, duration, request.url)
            else:
                return AmberdataResponse(request.text, request.status_code, duration, request.url)
        except Exception as exc:
            print(exc)
            as_5xx_error_json = {
                'status': 500, #default to 500 error
                'message': 'Failed to complete HTTP request.'
            }

            end_time = time.time()
            duration = end_time - start_time
            return AmberdataResponse(json.dumps(as_5xx_error_json), 500, duration, None)

    def call_endpoint_and_get_all_pages(self, path: str, query: dict, headers: dict, http_ok_next_page_url_extractor):
        """
        Iterative, non-recursive way to get all the pages given an initial URL.

        Avoids Python's max recursion depth (~1000).

        This is a generator function and should be used accordingly.

        Parameters
        ----------
        path: str
            The endpoint path with the path parameters inserted i.e if the path is `/market/defi/trades/{pool}/historical/` then provide `/market/defi/trades/0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640/historical`
        query: dict
            The query parameters in a dict i.e `{'startDate': '2023-05-20', 'endDate': '2023-05-21'}`
        headers: dict
            The request headers in a dict, you do not have to pass in the API key because the `EndpointCaller` class is initialized with it
        http_ok_next_page_url_extractor: function
            The function that extracts the next page url from a given API response

        Returns
        -------
        api_response: AmberdataResponse
            Individual page from the calling the endpoint

        """
        stack: AmberdataResponseStack = []
        current_page_response = self.call_endpoint_and_get_data_as_json(path, query, headers)
        current_page_response.increment_attempt()
        stack.append(current_page_response)
        while len(stack) > 0:
            response = stack.pop()
            if response.status == 200:
                next_page_url = http_ok_next_page_url_extractor(response.data)
                if len(next_page_url) > 0:
                    next_page_response = self.call_endpoint_and_get_data_as_json(next_page_url, None, headers)
                    next_page_response.increment_attempt()
                    stack.append(next_page_response)
                yield response
            else:
                if response.attempts < 3 and response.request_url is not None:
                    retried_page_response = self.call_endpoint_and_get_data_as_json(response.request_url, None, headers)
                    retried_page_response.increment_attempt(by=response.attempts + 1)
                    stack.append(retried_page_response)
                else:
                    yield response

        assert len(stack) == 0, "Stack is not empty, more pages need to be retrieved. # of remaining pages is {}".format(len(stack))
        return AmberdataResponse("DONE", -1, 0, "")