Send messages without messages queues

ct-app/README.md

@@ -139,7 +139,6 @@ Parameter | Recommanded value (staging) | Description
 `GCP_FILE_PREFIX` | `expected_reward` | File prefix for GCP distribution list storage
 `GCP_FOLDER` | `staging` | Folder on GCP where to store distribution list
 `PEER_MIN_VERSION` | `2.0.0` | Minimum node version to be eligible
-
 `RABBITMQ_HOST` | (check Bitwarden) | 
 `RABBITMQ_PASSWORD` | (check Bitwarden) | 
 `RABBITMQ_PROJECT_NAME` | `ct-app` | Name of the RabbitMQ project
@@ -155,6 +154,8 @@ Parameter | Recommanded value (staging) | Description
 `NODE_ADDRESS_X` (multiple, min. 2) | (check Bitwarden) |
 `NODE_KEY_X` | (check Bitwarden) | 
 
+Then there's a bunch of optional flags to enable features of the app (the name of the feature should be explicit enough).
+The following enables `Core` features:
 
 Then there's a bunch of optional flags to enable features of the app
 Flag | Recommanded value (staging)

ct-app/core/__main__.py

@@ -23,7 +23,6 @@ def main(configfile: str = None):
     params.from_env("SUBGRAPH_", "PG", "RABBITMQ_")
     params.overrides("OVERRIDE_")
 
-
     Utils.stringArrayToGCP(
         params.gcp.bucket,
         Utils.generateFilename("", "startup", "csv"),

ct-app/core/components/baseclass.py

@@ -9,6 +9,7 @@ class Base:
     """
     Base class for logging and printing messages with different colors.
     """
+
     handler = logging.StreamHandler()
     handler.setFormatter(formatter)
 
@@ -19,7 +20,10 @@ class Base:
 
     @property
     def print_prefix(self) -> str:
-        return ""
+        cls = self.__class__
+        raise NotImplementedError(
+            f"print_prefix not implemented for class '{cls.__name__}'"
+        )
 
     @classmethod
     def class_prefix(cls) -> str:

ct-app/core/components/decorators.py

@@ -31,10 +31,10 @@ async def wrapper(self, *args, **kwargs):
         if not hasattr(self.params, "flags"):
             self.error("No flags available")
             return
-        
+
         if not hasattr(self.params.flags, self.class_prefix()):
             raise AttributeError(f"Feature `{func.__name__}` not in config file")
-        
+
         class_flags = getattr(self.params.flags, self.class_prefix())
 
         params_raw = dir(class_flags)
@@ -71,7 +71,7 @@ async def wrapper(self, *args, **kwargs):
             if func_name_clean not in params_clean:
                 self.error(f"Feature `{func.__name__}` not regonized")
                 return
-            
+
             index = params_clean.index(func_name_clean)
             delay = getattr(class_flags, params_raw[index])
 
@@ -85,11 +85,10 @@ async def wrapper(self, *args, **kwargs):
 
                 if delay == 0:
                     break
-                
+
                 if delay is not None:
                     await asyncio.sleep(delay)
 
-
         return wrapper
 
     return decorator

ct-app/core/components/environment_utils.py

@@ -0,0 +1,25 @@
+from os import environ
+from typing import Any
+
+from .baseclass import Base
+
+
+class EnvironmentUtils(Base):
+    @property
+    def print_prefix(self) -> str:
+        return "EnvUtils"
+
+    @classmethod
+    def envvar(cls, var_name: str, default: Any = None, type: type = str):
+        if var_name in environ:
+            return type(environ[var_name])
+        else:
+            return default
+
+    @classmethod
+    def envvarWithPrefix(cls, prefix: str, type=str) -> dict[str, Any]:
+        var_dict = {
+            key: type(v) for key, v in environ.items() if key.startswith(prefix)
+        }
+
+        return dict(sorted(var_dict.items()))

ct-app/core/components/graphql_providers.py

@@ -58,7 +58,9 @@ async def _test_query(self, key: str, **kwargs) -> bool:
 
         # call `self._execute(self._sku_query, vars)` with a timeout
         try:
-            response = await asyncio.wait_for(self._execute(self._sku_query, vars), timeout=30)
+            response = await asyncio.wait_for(
+                self._execute(self._sku_query, vars), timeout=30
+            )
         except asyncio.TimeoutError:
             self.error("Query timeout occurred")
             return False
@@ -81,10 +83,12 @@ async def _get(self, key: str, **kwargs) -> dict:
             vars.update(kwargs)
 
             try:
-                response = await asyncio.wait_for(self._execute(self._sku_query, vars), timeout=30)
+                response = await asyncio.wait_for(
+                    self._execute(self._sku_query, vars), timeout=30
+                )
             except asyncio.TimeoutError:
                 self.error("Timeout error while fetching data from subgraph.")
-                break            
+                break
             if response is None:
                 break
 
@@ -136,9 +140,10 @@ async def test(self, **kwargs):
 
         if result is None:
             return False
-        
+
         return result
 
+
 class SafesProvider(GraphQLProvider):
     def __init__(self, url: str):
         super().__init__(url)
@@ -175,14 +180,13 @@ def __init__(self, url: str):
     def print_prefix(self) -> str:
         return "transaction-provider"
 
+
 class RewardsProvider(GraphQLProvider):
     def __init__(self, url: str):
         super().__init__(url)
         self._default_key = "accounts"
-        self._sku_query = self._load_query(
-            "core/subgraph_queries/rewards.graphql"
-        )
+        self._sku_query = self._load_query("core/subgraph_queries/rewards.graphql")
 
     @property
     def print_prefix(self) -> str:
-        return "rewards-provider"
+        return "rewards-provider"

ct-app/core/components/hoprd_api.py

@@ -10,13 +10,7 @@
     SendMessageBodyRequest,
     TagQueryRequest,
 )
-from hoprd_sdk.api import (
-    AccountApi,
-    ChannelsApi,
-    MessagesApi,
-    NetworkApi,
-    NodeApi,
-)
+from hoprd_sdk.api import AccountApi, ChannelsApi, MessagesApi, NetworkApi, NodeApi
 from hoprd_sdk.rest import ApiException
 from requests import Response
 from urllib3.exceptions import MaxRetryError
@@ -40,7 +34,7 @@ def _refresh_token_hook(self):
         self.configuration.refresh_api_key_hook = _refresh_token_hook
 
     @property
-    def print_prefix(self) -> str:
+    def print_prefix(cls) -> str:
         return "api"
 
     async def __call_api(
@@ -120,7 +114,7 @@ async def balances(self, type: Union[str, list[str]] = "all"):
         is_ok, response = await self.__call_api(AccountApi, "balances")
 
         if not is_ok:
-            return None
+            return {}
 
         return_dict = {}
 
@@ -142,9 +136,7 @@ async def open_channel(self, peer_address: str, amount: str):
         """
         body = OpenChannelBodyRequest(amount, peer_address)
 
-        is_ok, response = await self.__call_api(
-            ChannelsApi, "open_channel", body=body
-        )
+        is_ok, response = await self.__call_api(ChannelsApi, "open_channel", body=body)
 
         return response.channel_id if is_ok else None
 
@@ -212,7 +204,6 @@ async def outgoing_channels(self, only_id: bool = False):
             full_topology=False,
             including_closed=False,
         )
-
         if is_ok:
             if not hasattr(response, "outgoing"):
                 self.warning("Response does not contain 'outgoing'")
@@ -241,14 +232,15 @@ async def all_channels(self, include_closed: bool):
             full_topology="true",
             including_closed="true" if include_closed else "false",
         )
+
         return response if is_ok else []
 
     async def peers(
         self,
         params: Union[list, str] = "peer_id",
         status: str = "connected",
         quality: float = 0.5,
-    ):
+    ) -> list[dict]:
         """
         Returns a list of peers.
         :param: param: list or str = "peer_id"
@@ -335,7 +327,7 @@ async def messages_pop(self, tag: int = MESSAGE_TAG) -> bool:
         """
         body = TagQueryRequest(tag=tag)
         _, response = await self.__call_api(MessagesApi, "pop", body=body)
-    
+
         return response
 
     async def messages_pop_all(self, tag: int = MESSAGE_TAG) -> list:
@@ -350,7 +342,7 @@ async def messages_pop_all(self, tag: int = MESSAGE_TAG) -> list:
 
     async def node_info(self):
         _, response = await self.__call_api(NodeApi, "info")
-        
+
         return response
 
     async def ticket_price(self) -> int:

ct-app/core/components/parameters.py

@@ -1,11 +1,12 @@
 from .baseclass import Base
-from .utils import Utils
+from .environment_utils import EnvironmentUtils
 
 
 class Parameters(Base):
     """
     Class that represents a set of parameters that can be accessed and modified. The parameters are stored in a dictionary and can be accessed and modified using the dot notation. The parameters can be loaded from environment variables with a specified prefix.
     """
+
     def __init__(self):
         super().__init__()
 
@@ -21,7 +22,7 @@ def parse(self, data: dict):
                 setattr(self, key, value)
 
     def overrides(self, prefix: str):
-        for key, value in Utils.envvarWithPrefix(prefix).items():
+        for key, value in EnvironmentUtils.envvarWithPrefix(prefix).items():
             path = key.replace(prefix, "").lower().split("_")
 
             parent = self
@@ -41,7 +42,6 @@ def overrides(self, prefix: str):
                 else:
                     raise KeyError(f"Key {key} not found in parameters")
 
-
     def from_env(self, *prefixes: list[str]):
         for prefix in prefixes:
             subparams_name = prefix.lower()
@@ -55,29 +55,22 @@ def from_env(self, *prefixes: list[str]):
             else:
                 subparams = type(self)()
 
-            for key, value in Utils.envvarWithPrefix(prefix).items():
-                k = key.replace(prefix, "").lower()
-
-                # convert snake case to camel case
-                k = k.replace("_", " ").title().replace(" ", "")
-                k = k[0].lower() + k[1:]
-
-                try:
-                    value = float(value)
-                except ValueError:
-                    pass
-
-                try:
-                    integer = int(value)
-                    if integer == value:
-                        value = integer
-                except ValueError:
-                    pass
-
-                setattr(subparams, k, value)
+            self._parse_env_vars(prefix, subparams)
 
             setattr(self, subparams_name, subparams)
 
+    def _parse_env_vars(self, prefix, subparams):
+        for key, value in EnvironmentUtils.envvarWithPrefix(prefix).items():
+            k = self._format_key(key, prefix)
+            value = self._convert(value)
+            setattr(subparams, k, value)
+
+    def _format_key(self, key, prefix):
+        k = key.replace(prefix, "").lower()
+        k = k.replace("_", " ").title().replace(" ", "")
+        k = k[0].lower() + k[1:]
+        return k
+
     def _convert(self, value: str):
         try:
             value = float(value)