From 01acb3cc7bfbbaf43250af4603b1e8da2b11a45b Mon Sep 17 00:00:00 2001 From: KoleBarnes Date: Mon, 20 Dec 2021 08:14:50 -0800 Subject: [PATCH] Added dynamically generating list that is used to select the available Networks. Signed-off-by: KoleBarnes --- fetch-validator-status/networks.py | 12 ++++++++++++ fetch-validator-status/rest_api.py | 12 +++++++----- 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/fetch-validator-status/networks.py b/fetch-validator-status/networks.py index 6ee9e55..d079d74 100644 --- a/fetch-validator-status/networks.py +++ b/fetch-validator-status/networks.py @@ -3,12 +3,17 @@ import urllib.request import sys import re +from enum import Enum from collections import namedtuple from util import log from singleton import Singleton Network = namedtuple('Network', ['id', 'name', 'genesis_url', 'genesis_path']) +class NetworkEnum(Enum): + def _generate_next_value_(name, start, count, last_values): + return name + class Networks(object, metaclass=Singleton): def __init__(self): self._networks = self.__load_network_list() @@ -45,6 +50,13 @@ def __download_genesis_file(genesis_url: str, destination_path: str): log("Fetching genesis file ...") urllib.request.urlretrieve(genesis_url, destination_path) + @staticmethod + def get_NetworkEnum() -> NetworkEnum: + """Dynamically generates a NetworkEnum that can be used to select the available Networks. + """ + networks = Networks() + return NetworkEnum('Network', list(networks.ids)) + def resolve(self, network_id: str = None, genesis_url: str = None, genesis_path: str = None): network_name = None genesis_path_base = f"{self.__get_script_dir()}/" diff --git a/fetch-validator-status/rest_api.py b/fetch-validator-status/rest_api.py index cdd76af..22a11ff 100644 --- a/fetch-validator-status/rest_api.py +++ b/fetch-validator-status/rest_api.py @@ -9,7 +9,7 @@ create_did ) from pool import PoolCollection -from networks import Networks +from networks import Networks, NetworkEnum from fetch_status import FetchStatus, NodeNotFound from plugin_collection import PluginCollection @@ -30,6 +30,8 @@ pool_collection = None node_info = None +Network: NetworkEnum = Networks.get_NetworkEnum() + def set_plugin_parameters(status: bool = False, alerts: bool = False): # Store args and monitor_plugins for lazy loading. global default_args, pool_collection, node_info @@ -72,17 +74,17 @@ async def networks(): return data @app.get("/networks/{network}") -async def network(network: str = Path(..., example="sbn", description="The network code."), +async def network(network: Network = Path(Network.sbn, example="sbn", description="The network code."), status: bool = Query(False, description="Filter results to status only."), alerts: bool = Query(False, description="Filter results to alerts only."), seed: Optional[str] = Header(None, description="Your network monitor seed.")): monitor_plugins = set_plugin_parameters(status, alerts) ident = create_did(seed) - result = await node_info.fetch(network_id=network, monitor_plugins=monitor_plugins, ident=ident) + result = await node_info.fetch(network_id=network.value, monitor_plugins=monitor_plugins, ident=ident) return result @app.get("/networks/{network}/{node}") -async def node(network: str = Path(..., example="sbn", description="The network code."), +async def node(network: Network = Path(Network.sbn, example="sbn", description="The network code."), node: str = Path(..., example="FoundationBuilder", description="The node name."), status: bool = Query(False, description="Filter results to status only."), alerts: bool = Query(False, description="Filter results to alerts only."), @@ -90,7 +92,7 @@ async def node(network: str = Path(..., example="sbn", description="The network monitor_plugins = set_plugin_parameters(status, alerts) ident = create_did(seed) try: - result = await node_info.fetch(network_id=network, monitor_plugins=monitor_plugins, nodes=node, ident=ident) + result = await node_info.fetch(network_id=network.value, monitor_plugins=monitor_plugins, nodes=node, ident=ident) except NodeNotFound as error: print(error) raise HTTPException(status_code=400, detail=str(error))