Fetching Gas Prices Script

This script handles the process of fetching gas prices for hedger transactions from different sources, checks for errors, and ensures the gas price does not exceed a specified upper bound.

Full Script:

    def __get_gas_from_metaswap(self, priority, gas_upper_bound):
        gas_provider = f'https://gas-api.metaswap.codefi.network/networks/{self.chain_id}/suggestedGasFees'
        resp = None
        try:
            resp = requests.get(gas_provider, timeout=RequestTimeout)
            resp_json = resp.json()
            max_fee_per_gas = Decimal(resp_json[priority]['suggestedMaxFeePerGas'])
            max_priority_fee_per_gas = Decimal(resp_json[priority]['suggestedMaxPriorityFeePerGas'])
            apm.span_label(max_fee_per_gas=max_fee_per_gas, max_priority_fee_per_gas=max_priority_fee_per_gas,
                           gas_price_provider=gas_provider)
            if max_fee_per_gas > gas_upper_bound:
                raise OutOfRangeTransactionFee(f'gas price exceeded. {gas_upper_bound=} but it is {max_fee_per_gas}')
            gas_params = {
                'maxFeePerGas': Web3.to_wei(max_fee_per_gas, 'GWei'),
                'maxPriorityFeePerGas': Web3.to_wei(max_priority_fee_per_gas, 'GWei')
            }
            return gas_params
        except (RequestException, JSONDecodeError, KeyError):
            if not DevEnv:
                logging.exception(f'Failed to get gas info from metaswap {resp.status_code=}')
            raise FailedToGetGasPrice("Failed to get gas info from metaswap")


    # fixme: do it by asyncio way
    def __get_gas_from_rpc(self, priority, gas_upper_bound):
        gas_price = None
        for provider in self.providers:
            provider_url = str(provider.provider.endpoint_uri)  # noqa
            try:
                gas_price = provider.eth.gas_price
                apm.span_label(gas_price_from_provider=str(gas_price / 1e9), gas_price_provider=provider_url)
            except (ConnectionError, ReadTimeout, ValueError) as e:
                logging.error(f'Failed to get gas price from {provider_url}, {e=}')
                continue
            break
        if gas_price is None:
            raise FailedToGetGasPrice("Non of RCP could provide gas price!")
        if gas_price / 1e9 > gas_upper_bound:
            raise OutOfRangeTransactionFee(f'gas price exceeded. {gas_upper_bound=} but it is {gas_price / 1e9}')
        return dict(gasPrice=int(gas_price * TxPriorityMultiplier.get(priority, 1)))

    def _get_gas_price(self, gas_upper_bound, priority) -> dict:
        gas_params = {}
        if self.chain_id == 97:  # Test BNB Network
            gas_params['gasPrice'] = Web3.to_wei(10.1, 'GWei')
        elif DevEnv or self.chain_id in ChainIdsWithGasFromRPC:
            gas_params = self.__get_gas_from_rpc(priority, gas_upper_bound)
        else:
            try:
                gas_params = self.__get_gas_from_metaswap(priority, gas_upper_bound)
            except FailedToGetGasPrice:
                gas_params = self.__get_gas_from_rpc(priority, gas_upper_bound)
        apm.span_label(**gas_params)
        return gas_params

__get_gas_from_metaswap

Function Definition and Parameters:

def __get_gas_from_metaswap(self, priority, gas_upper_bound):

This function is designed to retrieve gas price information from the Metamask Gas API. It takes two parameters:

• priority: The priority level of the transaction, which will determine the gas fee structure to use (low, medium, high).

• gas_upper_bound: The maximum allowable gas price for the transaction.

Gas Provider URL:

gas_provider = f'https://gas-api.metaswap.codefi.network/networks/{self.chain_id}/suggestedGasFees'

Constructs the URL to fetch gas price information from the Metamask Gas API for the specific blockchain network identified by self.chain_id.

Initialize Response:

Initializes the variable resp to None. This will later be used to store the response from the API request.

Fetching and Processing Data:

try:
    resp = requests.get(gas_provider, timeout=RequestTimeout)
    resp_json = resp.json()

Attempts to send a GET request to the gas_provider URL with a specified timeout (RequestTimeout). The response from this request is stored in resp,then converts the response from the API into JSON format, which allows for easy access to the data fields within the response.

Extract Gas Fees:

max_fee_per_gas = Decimal(resp_json[priority]['suggestedMaxFeePerGas'])
max_priority_fee_per_gas = Decimal(resp_json[priority]['suggestedMaxPriorityFeePerGas'])

Extracts the suggested maximum fee per gas unit (suggestedMaxFeePerGas) and the suggested maximum priority fee per gas unit (Decimal) from the JSON response, converting these values to Decimal.

Log Span Label:

apm.span_label(max_fee_per_gas=max_fee_per_gas, max_priority_fee_per_gas=max_priority_fee_per_gas, gas_price_provider=gas_provider)

Logs the extracted gas fee information using apm.span_label for monitoring purposes. This includes the maximum fee per gas, the maximum priority fee per gas, and the URL of the gas price provider.

Check Upper Bound:

if max_fee_per_gas > gas_upper_bound:
    raise OutOfRangeTransactionFee(f'gas price exceeded. {gas_upper_bound=} but it is {max_fee_per_gas}')

Checks if the max_fee_per_gas exceeds the specified gas_upper_bound. If it does, raises an OutOfRangeTransactionFee exception with a descriptive message.

Prepare Gas Parameters:

gas_params = {
    'maxFeePerGas': Web3.to_wei(max_fee_per_gas, 'GWei'),
    'maxPriorityFeePerGas': Web3.to_wei(max_priority_fee_per_gas, 'GWei')
}
return gas_params

If the gas fee is within the acceptable range, this prepares a dictionary gas_params containing the gas fee values converted to Wei (a smaller unit of Ether) using Web3.to_wei, then returns the gas_params dictionary.

Exception Handling:

except (RequestException, JSONDecodeError, KeyError):
    if not DevEnv:
        logging.exception(f'Failed to get gas info from metaswap {resp.status_code=}')
    raise FailedToGetGasPrice("Failed to get gas info from metaswap")

Handles potential exceptions that might occur during the process:

• RequestException: General exception for request issues.

• JSONDecodeError: Issue decoding the JSON response.

• KeyError: If the expected keys are not present in the JSON response.

If an exception occurs, logs the error (unless in development environment DevEnv) and raises a FailedToGetGasPrice exception with an appropriate message.

__get_gas_from_rpc

Function Definition and Parameters:

def __get_gas_from_rpc(self, priority, gas_upper_bound):

This function is a private method to get gas prices from an RPC provider. It takes priority and gas_upper_bound as parameters.

Initialize Gas Price:

gas_price = None

Initializes the gas price variable to None.

Loop Through Providers:

for provider in self.providers:
    provider_url = str(provider.provider.endpoint_uri)  # noqa
    try:
        gas_price = provider.eth.gas_price
        apm.span_label(gas_price_from_provider=str(gas_price / 1e9), gas_price_provider=provider_url)
    except (ConnectionError, ReadTimeout, ValueError) as e:
        logging.error(f'Failed to get gas price from {provider_url}, {e=}')
        continue
    break

Iterates through available RPC providers to fetch the gas price. If it fails for a provider, logs the error and continues to the next one. Also logs the gas fees and provider URL for tracing purposes with apm.span_label

Check Gas Price:

if gas_price is None:
    raise FailedToGetGasPrice("None of RCP could provide gas price!")
if gas_price / 1e9 > gas_upper_bound:
    raise OutOfRangeTransactionFee(f'gas price exceeded. {gas_upper_bound=} but it is {gas_price / 1e9}')

If a gas price was retrieved, the function checks whether this price divided by 1e9 (to convert from wei to GWei) exceeds the gas_upper_bound. If it does, it raises an exception for exceeding the specified gas price limit.

Return Gas Parameters:

return dict(gasPrice=int(gas_price * TxPriorityMultiplier.get(priority, 1)))

Finally, the function returns a dictionary with a single key-value pair. The key is gasPrice, and the value is the fetched gas_price adjusted by a priority multiplier. This multiplier adjusts the gas price based on the transaction priority, allowing for some dynamic control over how much to pay per transaction.

__get_gas_price

Function Definition and Parameters:

def _get_gas_price(self, gas_upper_bound, priority) -> dict:

This function fetches the gas price based on the specified gas_upper_bound and priority.

Initialize Gas Parameters:

gas_params = {}

Initializes an empty dictionary gas_params to store the gas price parameters.

Conditional Check for Test BNB Network

if self.chain_id == 97:  # Test BNB Network
    gas_params['gasPrice'] = Web3.to_wei(10.1, 'GWei')

Checks if the chain_id is 97, which corresponds to the Test BNB Network. If true, it sets the gasPrice to 10.1 GWei (converted to Wei) directly in the gas_params dictionary.

Conditional Check for Development Environment or Specific Chains

   elif DevEnv or self.chain_id in ChainIdsWithGasFromRPC:
        gas_params = self.__get_gas_from_rpc(priority, gas_upper_bound)

If the code is running in a development environment (DevEnv) or if the chain_id is in the ChainIdsWithGasFromRPC list, it fetches the gas price from RPC providers by calling the __get_gas_from_rpc method with the specified priority and gas_upper_bound.

Attempt to Get Gas Price from Metamask Swap API

   else:
        try:
            gas_params = self.__get_gas_from_metaswap(priority, gas_upper_bound)
        except FailedToGetGasPrice:
            gas_params = self.__get_gas_from_rpc(priority, gas_upper_bound)

If the previous conditions are not met, it attempts to fetch the gas price from MetaSwap using the __get_gas_from_metaswap method. If this attempt fails and raises a FailedToGetGasPrice exception, it falls back to fetching the gas price from RPC providers using the __get_gas_from_rpc method.

Logging the Gas Parameters

   apm.span_label(**gas_params)

Logs the gas price parameters using for monitoring purposes. The **gas_params syntax unpacks the dictionary so that each key-value pair in gas_params is passed as a separate keyword argument.

Returning the Gas Parameters

   apm.span_label(**gas_params)

Returns the gas_params dictionary to the caller.

Summary:

The _get_gas_price function is designed to determine and return the appropriate gas price parameters for a transaction. It first checks if the network is the Test BNB Network, in which case it directly sets the gas price. If the environment is a development environment or the chain ID is in a specific list, it fetches the gas price from RPC providers. Otherwise, it tries to fetch the gas price from MetaSwap and falls back to RPC providers if that fails.