Skip to content

HTTPClient

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

Parameters:

Name Type Description Default
api str

URL to a Api node

 'https://rest.cosmos.directory/cosmoshub'
check_api bool

Check if the Api node is reachable. Returns either NotAnApiNode or NodeException if the node is not healthy.

 False
Source code in mospy/clients/HTTPClient.py 
 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
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
class HTTPClient:
    """
    Wrapper class to interact with a cosmos chain through their API endpoint

    Args:
        api (str): URL to a Api node
        check_api (bool): Check if the Api node is reachable. Returns either NotAnApiNode or NodeException if the node is not healthy.
    """

    def __init__(self, *, api: str = "https://rest.cosmos.directory/cosmoshub", check_api: bool = False):
        self._api = api.rstrip("/")

        # Check API if reachable
        if check_api:
            self.check_api()

    def check_api(self):
        """
        Checks if the API is reachable. Returns a NodeException or a NotAnApiNode Exception when the check failed.
        """
        try:
            check_response = httpx.get(self._api + "/node_info")
        except httpx.HTTPError:
            raise NodeException(
                "The passed url is not reachable. To disable the health check pass check_api=false to the HTTPClient constructor.")

        if check_response.status_code != 200:
            # Check if the returned text matches the one returned by an RPC
            if "404 page not found" in check_response.text:
                raise NotAnApiNode("Please pass an API endpoint to the HTTPClient. You probably passed an RPC.")
            else:
                raise NodeException(
                    "Health couldn't be verified. To disable the health check pass check_api=false to the HTTPClient constructor.")

    def _make_post_request(self, path, payload, timeout):
        try:
            req = httpx.post(self._api + path, json=payload, timeout=timeout)
        except httpx.TimeoutException:
            raise NodeTimeoutException(f"Node {self._api} timed out after {timeout} seconds")

        if req.status_code != 200:
            try:
                data = req.json()
                print(data)
                message = f"({data['message']}"
            except Exception:
                message = ""
            raise NodeException(f"Error while doing request to api endpoint {message}")

        data = req.json()
        return data

    def _make_get_request(self, path, timeout):
        try:
            req = httpx.get(self._api + path, timeout=timeout)
        except httpx.TimeoutException:
            raise NodeTimeoutException(f"Node {self._api} timed out after {timeout} seconds")
        if req.status_code != 200:
            try:
                data = req.json()
                message = f"({data['message']}"
            except:
                message = ""
            raise NodeException(f"Error while doing request to api endpoint {message}")

        data = req.json()
        return data


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

        Args:
            account (Account): Account
        """

        address = account.address
        url = self._api + "/cosmos/auth/v1beta1/accounts/" + address
        req = httpx.get(url=url)
        if req.status_code != 200:
            raise NodeException("Error while doing request to api endpoint")

        data = req.json()
        sequence = int(data["account"]["sequence"])
        account_number = int(data["account"]["account_number"])

        account.next_sequence = sequence
        account.account_number = account_number

    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
            timeout (int): Timeout

        Returns:
            hash: Transaction hash
            code: Result code
            log: Log (None if transaction successful)
        """
        path = "/cosmos/tx/v1beta1/txs"
        tx_bytes = transaction.get_tx_bytes_as_string()
        payload = {"tx_bytes": tx_bytes, "mode": "BROADCAST_MODE_SYNC"}

        data = self._make_post_request(path, payload, timeout)

        hash = data["tx_response"]["txhash"]
        code = data["tx_response"]["code"]
        log = None if code == 0 else data["tx_response"]["raw_log"]

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

    def estimate_gas(self,
                              *,
                              transaction: Transaction,
                              update: bool = True,
                              multiplier: float = 1.2,
                              timeout: int = 10) ->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
            timeout (int): Timeout

        Returns:
            expedted_gas: Expected gas
        """
        path = "/cosmos/tx/v1beta1/simulate"
        tx_bytes = transaction.get_tx_bytes_as_string()
        payload = {"tx_bytes": tx_bytes}

        data = self._make_post_request(path, payload, timeout)

        gas_used = int(data["gas_info"]["gas_used"])

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

        return gas_used

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

        Note:
            Takes only positional arguments.

        Args:
            tx_hash (Transaction): The transaction hash
            timeout (int): Timeout for the request before throwing a NodeException

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


        """
        path = "/cosmos/tx/v1beta1/txs/" + tx_hash

        try:
            data = self._make_get_request(path=path, timeout=timeout)
        except NodeException:
            raise TransactionNotFound(f"The transaction {tx_hash} couldn't be found")

        return data


    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.")

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
timeout int

Timeout

 10

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/HTTPClient.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
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
        timeout (int): Timeout

    Returns:
        hash: Transaction hash
        code: Result code
        log: Log (None if transaction successful)
    """
    path = "/cosmos/tx/v1beta1/txs"
    tx_bytes = transaction.get_tx_bytes_as_string()
    payload = {"tx_bytes": tx_bytes, "mode": "BROADCAST_MODE_SYNC"}

    data = self._make_post_request(path, payload, timeout)

    hash = data["tx_response"]["txhash"]
    code = data["tx_response"]["code"]
    log = None if code == 0 else data["tx_response"]["raw_log"]

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

check_api()

Checks if the API is reachable. Returns a NodeException or a NotAnApiNode Exception when the check failed.

Source code in mospy/clients/HTTPClient.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def check_api(self):
    """
    Checks if the API is reachable. Returns a NodeException or a NotAnApiNode Exception when the check failed.
    """
    try:
        check_response = httpx.get(self._api + "/node_info")
    except httpx.HTTPError:
        raise NodeException(
            "The passed url is not reachable. To disable the health check pass check_api=false to the HTTPClient constructor.")

    if check_response.status_code != 200:
        # Check if the returned text matches the one returned by an RPC
        if "404 page not found" in check_response.text:
            raise NotAnApiNode("Please pass an API endpoint to the HTTPClient. You probably passed an RPC.")
        else:
            raise NodeException(
                "Health couldn't be verified. To disable the health check pass check_api=false to the HTTPClient constructor.")

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

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
timeout int

Timeout

 10

Returns:

Name Type Description
expedted_gas int

Expected gas
Source code in mospy/clients/HTTPClient.py 
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
def estimate_gas(self,
                          *,
                          transaction: Transaction,
                          update: bool = True,
                          multiplier: float = 1.2,
                          timeout: int = 10) ->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
        timeout (int): Timeout

    Returns:
        expedted_gas: Expected gas
    """
    path = "/cosmos/tx/v1beta1/simulate"
    tx_bytes = transaction.get_tx_bytes_as_string()
    payload = {"tx_bytes": tx_bytes}

    data = self._make_post_request(path, payload, timeout)

    gas_used = int(data["gas_info"]["gas_used"])

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

    return gas_used

get_tx(*, tx_hash, timeout=5)

Query a transaction by passing the hash

Note

Takes only positional arguments.

Parameters:

Name Type Description Default
tx_hash Transaction

The transaction hash

 required
timeout int

Timeout for the request before throwing a NodeException

 5

Returns:

Name Type Description
transaction dict

Transaction dict as returned by the chain
Source code in mospy/clients/HTTPClient.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def get_tx(self, *, tx_hash: str, timeout: int = 5):
    """
    Query a transaction by passing the hash

    Note:
        Takes only positional arguments.

    Args:
        tx_hash (Transaction): The transaction hash
        timeout (int): Timeout for the request before throwing a NodeException

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


    """
    path = "/cosmos/tx/v1beta1/txs/" + tx_hash

    try:
        data = self._make_get_request(path=path, timeout=timeout)
    except NodeException:
        raise TransactionNotFound(f"The transaction {tx_hash} couldn't be found")

    return data

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/HTTPClient.py 
 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
    """

    address = account.address
    url = self._api + "/cosmos/auth/v1beta1/accounts/" + address
    req = httpx.get(url=url)
    if req.status_code != 200:
        raise NodeException("Error while doing request to api endpoint")

    data = req.json()
    sequence = int(data["account"]["sequence"])
    account_number = int(data["account"]["account_number"])

    account.next_sequence = sequence
    account.account_number = account_number

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/HTTPClient.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
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.")