Blockchain-RPC-Exporter/src/registries.py at 223c348564f2437480e6fcddb9bf439bb663d1c3 · smartcontractkit/Blockchain-RPC-Exporter · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
"""A module that providers registries of objects."""
import sys

from configuration import Config
import collectors
from log import logger


class Endpoint():  # pylint: disable=too-few-public-methods
    """RPC Endpoint class, to store metadata."""

    def __init__(  # pylint: disable=too-many-arguments,too-many-positional-arguments
            self, url, provider, blockchain, network_name, network_type, integration_maturity,
            canonical_name, chain_selector,
            chain_id, **client_parameters):
        self.url = url
        self.chain_id = chain_id
        self.labels = [
            url, provider, blockchain, network_name, network_type, integration_maturity,
            canonical_name, str(chain_selector),
            str(chain_id)
        ]
        self.client_parameters = client_parameters


class EndpointRegistry(Config):
    """A registry of all endpoints."""

    def __init__(self):
        super().__init__()
        self._logger = logger
        self._logger_metadata = {'component': 'Registries'}

    @property
    def blockchain(self):
        """Returns blockchain as a lowercase string."""
        return self.get_property('blockchain').lower()

    @property
    def collector(self):
        """Returns type of collector used as a lowercase string."""
        return self.get_property('collector').lower()

    @property
    def get_endpoint_registry(self) -> list:
        """Iterates trough all of the endpoints and instantiates
        them as Endpoints class in a dict. Returns the populated dict."""
        endpoints_list = []
        for item in self.endpoints:
            endpoints_list.append(
                Endpoint(item['url'], item['provider'],
                         self.blockchain,
                         self.get_property('network_name'),
                         self.get_property('network_type'),
                         self.get_property('integration_maturity'),
                         self.get_property('canonical_name'),
                         self.get_property('chain_selector'),
                         self.get_property('chain_id'),
                         **self.client_parameters))
        return endpoints_list


class CollectorRegistry(EndpointRegistry):
    """A registry of all collectors."""

    @property
    def get_collector_registry(self) -> list:
        """Iterates trough all of the instantiated endpoints and loads
        proper collector type based on the collector and chain name."""
        collectors_list = []

        for item in self.get_endpoint_registry:
            collector = None
            match self.collector, self.blockchain:
                case "conflux", "conflux":
                    collector = collectors.ConfluxCollector
                case "cardano", "cardano":
                    collector = collectors.CardanoCollector
                case "bitcoin", "bitcoin":
                    collector = collectors.BitcoinCollector
                case "bitcoin", "dogecoin":
                    collector = collectors.BitcoinCollector
                case "filecoin", "filecoin":
                    collector = collectors.FilecoinCollector
                case "solana", "solana":
                    collector = collectors.SolanaCollector
                case "starknet", "starknet":
                    collector = collectors.StarknetCollector
                case "aptos", "aptos":
                    collector = collectors.AptosCollector
                case "xrpl", "xrpl":
                    collector = collectors.XRPLCollector
                case "ton", "ton":
                    collector = collectors.TonCollector
                case "evmhttp", other: # pylint: disable=unused-variable
                    collector = collectors.EvmHttpCollector
                case "evm", other:  # pylint: disable=unused-variable
                    collector = collectors.EvmCollector
            if collector is None:
                self._logger.error("Endpoint has no supported collector",
                                   collector=self.collector,
                                   blockchain=self.blockchain,
                                   **self._logger_metadata)
                sys.exit(1)
            else:
                collectors_list.append(collector(item.url,
                                                 item.labels, item.chain_id,
                                                 **self.client_parameters))
        return collectors_list