# Fetching Gas Prices Helper 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:** ```python 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:** ```python 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:** ```python 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:** ```python 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:** ```python 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:** ```python 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:** ```python 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:** ```python 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:** ```python 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: ```python gas_price = None ``` Initializes the gas price variable to None. ### Loop Through Providers: ```python 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: ```python 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: ```python 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: ```python 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: ```python gas_params = {} ``` Initializes an empty dictionary `gas_params` to store the gas price parameters. ### Conditional Check for Test BNB Network ```python 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** ```python 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** ```python 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** ```python 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 ```python 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.