Skip to content

HTTPClient

Wrapper class to interact with a cosmos chain through their grpc endpoint

Parameters:

Name Type Description Default
host str

URL to a Cosmos api node

 'cosmoshub.strange.love'
port int

Port to connect to

 9090
ssl bool

Whether a ssl encrypted endpoint should be used

 False
protobuf str

Define which protobuf files to use. Cosmos, Evmos and Osmosis are built in and otherwise pass the raw package name (cosmospy-protobuf)

 'cosmos'
Source code in mospy/clients/GRPCClient.py 
 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
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
class GRPCClient:
    """
    Wrapper class to interact with a cosmos chain through their grpc endpoint

    Args:
        host (str): URL to a Cosmos api node
        port (int): Port to connect to
        ssl (bool): Whether a ssl encrypted endpoint should be used
        protobuf (str): Define which protobuf files to use. Cosmos, Evmos and Osmosis are built in and otherwise pass the raw package name (cosmospy-protobuf)

    """

    def __init__(
        self,
        *,
        host: str = "cosmoshub.strange.love",
        port: int = 9090,
        ssl: bool = False,
        protobuf="cosmos",
    ):

        _protobuf_packages = {
            "cosmos": "cosmospy_protobuf",
            "osmosis": "osmosis_protobuf",
            "evmos": "evmos_protobuf",
            "sentinel": "sentinel_protobuf",
        }
        _protobuf_package = (_protobuf_packages[protobuf.lower()]
                             if protobuf.lower() in _protobuf_packages.keys()
                             else protobuf)
        try:
            self._cosmos_tx_service_pb2 = importlib.import_module(
                _protobuf_package +
                ".cosmos.tx.v1beta1.service_pb2")
            self._cosmos_auth_query_pb2 = importlib.import_module(
                _protobuf_package + ".cosmos.auth.v1beta1.query_pb2")
            self._cosmos_auth_query_pb2_grpc = importlib.import_module(
                _protobuf_package + ".cosmos.auth.v1beta1.query_pb2_grpc")
            self._cosmos_tx_service_pb2_grpc = importlib.import_module(
                _protobuf_package + ".cosmos.tx.v1beta1.service_pb2_grpc")

            self._BroadcastTxRequest = self._cosmos_tx_service_pb2.BroadcastTxRequest
            self._SimulateTxRequest = self._cosmos_tx_service_pb2.SimulateRequest

        except AttributeError:
            raise ImportError(
                "It seems that you are importing conflicting protobuf files. Have sou set the protobuf attribute to specify your coin? Check out the documentation for more information."
            )
        except:
            raise ImportError(
                f"Couldn't import from {_protobuf_package}. Is the package installed? "
            )

        self._host = host
        self._port = port
        self._ssl = ssl

    def _connect(self):
        if self._ssl:
            con = grpc.secure_channel(
                f"{self._host}:{self._port}",
                credentials=grpc.ssl_channel_credentials())
        else:
            con = grpc.insecure_channel(f"{self._host}:{self._port}")
        return con

    def load_account_data(self, account: Account):
        """
        Load the ``next_sequence`` and ``account_number`` into the account object.

        Args:
            account (Account): Account
        """
        con = self._connect()
        address = account.address

        query_stub = self._cosmos_auth_query_pb2_grpc.QueryStub(con)
        account_request = self._cosmos_auth_query_pb2.QueryAccountRequest(address=address)

        req = query_stub.Account(account_request)
        data = dict(MessageToDict(req.account))

        sequence = 0 if not "sequence" in data else int(data["sequence"])
        account_number = int(data["accountNumber"])

        account.next_sequence = sequence
        account.account_number = account_number
        con.close()

    def broadcast_transaction(self,
                              *,
                              transaction: Transaction,
                              timeout: int = 10) -> [str, int, str]:
        """
        Sign and broadcast a transaction.

        Note:
            Takes only positional arguments

        Args:
            transaction (Transaction): The transaction object

        Returns:
            hash: Transaction hash
            code: Result code
            log: Log (None if transaction successful)
        """
        con = self._connect()
        tx_bytes = transaction.get_tx_bytes()

        tx_request = self._BroadcastTxRequest(
            tx_bytes=tx_bytes,
            mode=2  # BROADCAST_MODE_SYNC
        )

        tx_stub = self._cosmos_tx_service_pb2_grpc.ServiceStub(con)
        tx_data = tx_stub.BroadcastTx(tx_request)

        hash = tx_data.tx_response.txhash
        code = tx_data.tx_response.code
        log = None if code == 0 else tx_data.tx_response.raw_log

        return {"hash": hash, "code": code, "log": log}

    def get_tx(self, *, tx_hash: str):
        """
        Query a transaction by passing the hash

        Note:
            Takes only positional arguments.

        Args:
            tx_hash (Transaction): The transaction hash

        Returns:
            transaction (dict): Transaction dict as returned by the chain


        """
        con = self._connect()
        tx_stub = self._cosmos_tx_service_pb2_grpc.ServiceStub(con)
        try:
            return MessageToDict(tx_stub.GetTx(self._cosmos_tx_service_pb2.GetTxRequest(hash=tx_hash)))
        except grpc.RpcError:
            raise TransactionNotFound(f"The transaction {tx_hash} couldn't be found on chain.")

    def wait_for_tx(self, *, tx_hash: str, timeout: float = 60, poll_period: float = 10):
        """
        Waits for a transaction hash to hit the chain.

        Note:
            Takes only positional arguments

        Args:
            tx_hash (Transaction): The transaction hash
            timeout (bool): Time to wait before throwing a TransactionTimeout. Defaults to 60
            poll_period (float): Time to wait between each check. Defaults to 10

        Returns:
            transaction (dict): Transaction dict as returned by the chain
        """
        start = time.time()
        while time.time() < (start + timeout):
            try:
                return self.get_tx(tx_hash=tx_hash)
            except TransactionNotFound:
                time.sleep(poll_period)

        raise TransactionTimeout(f"The transaction {tx_hash} couldn't be found on chain within {timeout} seconds.")
    def estimate_gas(self,
                              *,
                              transaction: Transaction,
                              update: bool = True,
                              multiplier: float = 1.2
                     ) ->int:
        """
        Simulate a transaction to get the estimated gas usage.

        Note:
            Takes only positional arguments

        Args:
            transaction (Transaction): The transaction object
            update (bool): Update the transaction with the estimated gas amount
            multiplier (float): Multiplier for the estimated gas when updating the transaction. Defaults to 1.2

        Returns:
            expedted_gas: Expected gas
        """
        con = self._connect()
        tx_bytes = transaction.get_tx_bytes()

        simulate_request = self._SimulateTxRequest(
            tx_bytes=tx_bytes,
        )

        tx_stub = self._cosmos_tx_service_pb2_grpc.ServiceStub(con)
        tx_data = tx_stub.Simulate(simulate_request)

        gas_used = tx_data.gas_info.gas_used

        if update:
            transaction.set_gas(int(gas_used * multiplier))

        return gas_used

broadcast_transaction(*, transaction, timeout=10)

Sign and broadcast a transaction.

Note

Takes only positional arguments

Parameters:

Name Type Description Default
transaction Transaction

The transaction object

 required

Returns:

Name Type Description
hash [str, int, str]

Transaction hash
code [str, int, str]

Result code
log [str, int, str]

Log (None if transaction successful)
Source code in mospy/clients/GRPCClient.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def broadcast_transaction(self,
                          *,
                          transaction: Transaction,
                          timeout: int = 10) -> [str, int, str]:
    """
    Sign and broadcast a transaction.

    Note:
        Takes only positional arguments

    Args:
        transaction (Transaction): The transaction object

    Returns:
        hash: Transaction hash
        code: Result code
        log: Log (None if transaction successful)
    """
    con = self._connect()
    tx_bytes = transaction.get_tx_bytes()

    tx_request = self._BroadcastTxRequest(
        tx_bytes=tx_bytes,
        mode=2  # BROADCAST_MODE_SYNC
    )

    tx_stub = self._cosmos_tx_service_pb2_grpc.ServiceStub(con)
    tx_data = tx_stub.BroadcastTx(tx_request)

    hash = tx_data.tx_response.txhash
    code = tx_data.tx_response.code
    log = None if code == 0 else tx_data.tx_response.raw_log

    return {"hash": hash, "code": code, "log": log}

estimate_gas(*, transaction, update=True, multiplier=1.2)

Simulate a transaction to get the estimated gas usage.

Note

Takes only positional arguments

Parameters:

Name Type Description Default
transaction Transaction

The transaction object

 required
update bool

Update the transaction with the estimated gas amount

 True
multiplier float

Multiplier for the estimated gas when updating the transaction. Defaults to 1.2

 1.2

Returns:

Name Type Description
expedted_gas int

Expected gas
Source code in mospy/clients/GRPCClient.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def estimate_gas(self,
                          *,
                          transaction: Transaction,
                          update: bool = True,
                          multiplier: float = 1.2
                 ) ->int:
    """
    Simulate a transaction to get the estimated gas usage.

    Note:
        Takes only positional arguments

    Args:
        transaction (Transaction): The transaction object
        update (bool): Update the transaction with the estimated gas amount
        multiplier (float): Multiplier for the estimated gas when updating the transaction. Defaults to 1.2

    Returns:
        expedted_gas: Expected gas
    """
    con = self._connect()
    tx_bytes = transaction.get_tx_bytes()

    simulate_request = self._SimulateTxRequest(
        tx_bytes=tx_bytes,
    )

    tx_stub = self._cosmos_tx_service_pb2_grpc.ServiceStub(con)
    tx_data = tx_stub.Simulate(simulate_request)

    gas_used = tx_data.gas_info.gas_used

    if update:
        transaction.set_gas(int(gas_used * multiplier))

    return gas_used

get_tx(*, tx_hash)

Query a transaction by passing the hash

Note

Takes only positional arguments.

Parameters:

Name Type Description Default
tx_hash Transaction

The transaction hash

 required

Returns:

Name Type Description
transaction dict

Transaction dict as returned by the chain
Source code in mospy/clients/GRPCClient.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def get_tx(self, *, tx_hash: str):
    """
    Query a transaction by passing the hash

    Note:
        Takes only positional arguments.

    Args:
        tx_hash (Transaction): The transaction hash

    Returns:
        transaction (dict): Transaction dict as returned by the chain


    """
    con = self._connect()
    tx_stub = self._cosmos_tx_service_pb2_grpc.ServiceStub(con)
    try:
        return MessageToDict(tx_stub.GetTx(self._cosmos_tx_service_pb2.GetTxRequest(hash=tx_hash)))
    except grpc.RpcError:
        raise TransactionNotFound(f"The transaction {tx_hash} couldn't be found on chain.")

load_account_data(account)

Load the next_sequence and account_number into the account object.

Parameters:

Name Type Description Default
account Account

Account

 required
Source code in mospy/clients/GRPCClient.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def load_account_data(self, account: Account):
    """
    Load the ``next_sequence`` and ``account_number`` into the account object.

    Args:
        account (Account): Account
    """
    con = self._connect()
    address = account.address

    query_stub = self._cosmos_auth_query_pb2_grpc.QueryStub(con)
    account_request = self._cosmos_auth_query_pb2.QueryAccountRequest(address=address)

    req = query_stub.Account(account_request)
    data = dict(MessageToDict(req.account))

    sequence = 0 if not "sequence" in data else int(data["sequence"])
    account_number = int(data["accountNumber"])

    account.next_sequence = sequence
    account.account_number = account_number
    con.close()

wait_for_tx(*, tx_hash, timeout=60, poll_period=10)

Waits for a transaction hash to hit the chain.

Note

Takes only positional arguments

Parameters:

Name Type Description Default
tx_hash Transaction

The transaction hash

 required
timeout bool

Time to wait before throwing a TransactionTimeout. Defaults to 60

 60
poll_period float

Time to wait between each check. Defaults to 10

 10

Returns:

Name Type Description
transaction dict

Transaction dict as returned by the chain
Source code in mospy/clients/GRPCClient.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def wait_for_tx(self, *, tx_hash: str, timeout: float = 60, poll_period: float = 10):
    """
    Waits for a transaction hash to hit the chain.

    Note:
        Takes only positional arguments

    Args:
        tx_hash (Transaction): The transaction hash
        timeout (bool): Time to wait before throwing a TransactionTimeout. Defaults to 60
        poll_period (float): Time to wait between each check. Defaults to 10

    Returns:
        transaction (dict): Transaction dict as returned by the chain
    """
    start = time.time()
    while time.time() < (start + timeout):
        try:
            return self.get_tx(tx_hash=tx_hash)
        except TransactionNotFound:
            time.sleep(poll_period)

    raise TransactionTimeout(f"The transaction {tx_hash} couldn't be found on chain within {timeout} seconds.")