Create terraformer

fastlane_bot/bot.py

@@ -1361,6 +1361,16 @@ def _ensure_connection(self, tenderly_fork: str):
         self.db.cfg.w3 = Web3(Web3.HTTPProvider(tenderly_uri))
         self.ConfigObj.w3 = Web3(Web3.HTTPProvider(tenderly_uri))
 
+    def get_tokens_in_exchange(
+            self,
+            exchange_name: str,
+    ) -> List[str]:
+        """
+        Gets all tokens that exist in pools on the specified exchange.
+        :param exchange_name: the exchange name
+        """
+        return self.db.get_tokens_from_exchange(exchange_name=exchange_name)
+
     def run(
         self,
         *,

fastlane_bot/config/config.py

@@ -7,6 +7,7 @@
 
 import os
 from dataclasses import dataclass, field, InitVar, asdict
+
 # from .base import ConfigBase
 from . import network as network_, db as db_, logger as logger_, provider as provider_
 from .cloaker import CloakerL
@@ -18,18 +19,23 @@
 load_dotenv()
 TENDERLY_FORK_ID = os.environ.get("TENDERLY_FORK_ID")
 if TENDERLY_FORK_ID is None:
-    TENDERLY_FORK_ID = ''
+    TENDERLY_FORK_ID = ""
 WEB3_ALCHEMY_PROJECT_ID = os.environ.get("WEB3_ALCHEMY_PROJECT_ID")
-PROVIDER_URL = f'https://rpc.tenderly.co/fork/{TENDERLY_FORK_ID}' if TENDERLY_FORK_ID != '' else f"https://eth-mainnet.alchemyapi.io/v2/{WEB3_ALCHEMY_PROJECT_ID}"
-NETWORK_ID = 'mainnet' if TENDERLY_FORK_ID == '' else 'tenderly'
-NETWORK_NAME = "Ethereum Mainnet" if TENDERLY_FORK_ID == '' else 'Tenderly (Alchemy)'
+PROVIDER_URL = (
+    f"https://rpc.tenderly.co/fork/{TENDERLY_FORK_ID}"
+    if TENDERLY_FORK_ID != ""
+    else f"https://eth-mainnet.alchemyapi.io/v2/{WEB3_ALCHEMY_PROJECT_ID}"
+)
+NETWORK_ID = "mainnet" if TENDERLY_FORK_ID == "" else "tenderly"
+NETWORK_NAME = "Ethereum Mainnet" if TENDERLY_FORK_ID == "" else "Tenderly (Alchemy)"
 
 
 @dataclass
-class Config():
+class Config:
     """
     Fastlane bot configuration object
     """
+
     __VERSION__ = __VERSION__
     __DATE__ = __DATE__
 
@@ -52,7 +58,17 @@ class Config():
     LL_WARN = S.LOGLEVEL_WARNING
     LL_ERR = S.LOGLEVEL_ERROR
 
-    SUPPORTED_EXCHANGES = ['carbon_v1', 'bancor_v2', 'bancor_v3', 'uniswap_v2', 'uniswap_v3', 'sushiswap_v2', 'bancor_pol', 'pancakeswap_v2', 'pancakeswap_v3']
+    SUPPORTED_EXCHANGES = [
+        "carbon_v1",
+        "bancor_v2",
+        "bancor_v3",
+        "uniswap_v2",
+        "uniswap_v3",
+        "sushiswap_v2",
+        "bancor_pol",
+        "pancakeswap_v2",
+        "pancakeswap_v3",
+    ]
     connection = EthereumNetwork(
         network_id=NETWORK_ID,
         network_name=NETWORK_NAME,
@@ -62,11 +78,14 @@ class Config():
     connection.connect_network()
     w3 = connection.web3
 
+    UNI_V2_FORKS = ["uniswap_v2", "sushiswap_v2", "pancakeswap_v2"]
+    UNI_V3_FORKS = ["uniswap_v3", "pancakeswap_v3"]
+
     @classmethod
     def new(cls, *, config=None, loglevel=None, logging_path=None, **kwargs):
         """
         Alternative constructor: create and return new Config object
-        
+
         :config:    CONFIG_MAINNET(default), CONFIG_TENDERLY, CONFIG_UNITTEST
         :loglevel:  LOGLEVEL_DEBUG, LOGLEVEL_INFO (default), LOGLEVEL_WARNING, LOGLEVEL_ERROR
         """
@@ -88,7 +107,9 @@ def new(cls, *, config=None, loglevel=None, logging_path=None, **kwargs):
         elif config == cls.CONFIG_UNITTEST:
             C_db = db_.ConfigDB.new(db=S.DATABASE_UNITTEST, POSTGRES_DB="unittest")
             C_nw = network_.ConfigNetwork.new(network=S.NETWORK_MAINNET)
-            C_pr = provider_.ConfigProvider.new(network=C_nw, provider=S.PROVIDER_DEFAULT)
+            C_pr = provider_.ConfigProvider.new(
+                network=C_nw, provider=S.PROVIDER_DEFAULT
+            )
             return cls(db=C_db, logger=C_log, network=C_nw, provider=C_pr, **kwargs)
         raise ValueError(f"Invalid config: {config}")
 
@@ -113,22 +134,28 @@ def get_attribute_from_config(self, name: str):
         for obj in [self.network, self.db, self.provider, self.logger]:
             if hasattr(obj, name):
                 return getattr(obj, name)
-        raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
+        raise AttributeError(
+            f"'{self.__class__.__name__}' object has no attribute '{name}'"
+        )
 
     def __getattr__(self, name: str):
         """
         If of type attribute, return it.
         """
         if self.is_config_item(name):
             return self.get_attribute_from_config(name)
-        raise AttributeError(f"'{self.__class__.__name__}' object has no attribute '{name}'")
+        raise AttributeError(
+            f"'{self.__class__.__name__}' object has no attribute '{name}'"
+        )
 
     def __post_init__(self):
         """
         Post-initialization initialization.
         """
         if self.network is None:
-            self.network = network_.ConfigNetwork.new(network_.ConfigNetwork.NETWORK_ETHEREUM)
+            self.network = network_.ConfigNetwork.new(
+                network_.ConfigNetwork.NETWORK_ETHEREUM
+            )
         assert issubclass(type(self.network), network_.ConfigNetwork)
 
         if self.db is None:
@@ -147,14 +174,16 @@ def __post_init__(self):
             self.provider = provider_.ConfigProvider.new(self.network)
         assert issubclass(type(self.provider), provider_.ConfigProvider)
 
-        assert self.network is self.provider.network, f"Network mismatch: {self.network} != {self.provider.network}"
+        assert (
+            self.network is self.provider.network
+        ), f"Network mismatch: {self.network} != {self.provider.network}"
 
     VISIBLE_FIELDS = "network, db, logger, provider, w3, ZERO_ADDRESS"
 
     def cloaked(self, incl=None, excl=None):
         """
         returns a cloaked version of the object
-        
+
         :incl:  fields to _include_ in the cloaked version (plus those in VISIBLE_FIELDS)
         :excl:  fields to _exclude_ from the cloaked version
         """