tradehub.transactions.Transactions.submit_transaction_on_chain() - Code Metrics - Mai-Te-Pora/tradehub-python - Measure and Improve Code Quality continuously with Scrutinizer

Transactions.submit_transaction_on_chain() A
last analyzed 2021-03-26 05:54 UTC

↳ Parent: tradehub.transactions

Complexity

Conditions

Size

Total Lines	16
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	5
nop	4
dl	0
loc	16
rs	10
c	0
b	0
f	0

"""
Description:

    Transaction Class used as part of the Authenticated Client to standardize how on-chain messages are constructed.
    This client is the basis any on-chain transaction and handles message construction, standards, signing, and broadcasting.

Usage::

    from tradehub.transactions import Transactions
"""
import jsons

from tradehub.public_account_client import PublicClient as TradehubPublicClient
import tradehub.types as types
from tradehub.wallet import Wallet


class Transactions(TradehubPublicClient):
    """
    This class constructs, standardizes, signs, and broadcasts messages to the Tradehub network.

    Execution of this function is as follows::

        Transactions(wallet=Wallet(),
                     trusted_ips=None,
                     trusted_uris=None,
                     network="mainnet")
    """

    def __init__(self, wallet: Wallet, trusted_ips: list = None, trusted_uris: list = None, network: str = "testnet"):
        """
        :param wallet: Wallet class for Tradehub interaction.
        :type wallet: Wallet
        :param trusted_ips: Known and trusted IPs to connect to for your API requests.
        :type trusted_ips: list
        :param trusted_uris: Known and trusted URIs to connect to for your API requests.
        :type trusted_uris: list
        :param network: Network to submit the transaction.
        :type network: str
        """
        TradehubPublicClient.__init__(self, network=network, trusted_ips=trusted_ips, trusted_uris=trusted_uris)
        self.wallet = wallet
        self.update_account_details()
        self.network_variables = {
            "testnet": {"chain_id": "switcheochain", },
            "mainnet": {"chain_id": "switcheo-tradehub-1", },
        }
        self.chain_id = self.network_variables[network]["chain_id"]
        self.fees = self.get_transactions_fees()
        self.mode = "block"        # Need to automate
        self.gas = "100000000000"  # Need to automate
        self.tokens = self.get_token_details()

    def update_account_details(self) -> None:
        """
        Triggers an update to fetch current sequence number.

        :return: None
        """
        self.account_blockchain_dict = self.get_account_details()
        self.account_nbr = self.account_blockchain_dict["result"]["value"]["account_number"]
        self.account_sequence_nbr = self.account_blockchain_dict["result"]["value"]["sequence"]

    def get_account_details(self):
        """
        Retrieve Wallet account details required for submitting transactions on Tradehub.

        Execution of this function is as follows::

            get_account_details()

        The expected return result for this function is as follows::

            {
                'height': '1950991',
                'result': {
                    'type': 'cosmos-sdk/Account',
                    'value': {
                        'address': 'tswth1urfldmcspc9nk4w8vcfakxdf2rc5ee4fr2dn76',
                        'coins': [{
                            'denom': 'btc',
                            'amount': '99033160'
                        }, {
                        ...
                        }],
                        'public_key': {
                            'type': 'tendermint/PubKeySecp256k1',
                            'value': 'Ao8h4bhZr1/m8ZEwPaOizNCcIMDX/yqwQh7LXhI77FLW'
                        },
                        'account_number': '56',
                        'sequence': '3195'
                    }
                }
            }

        :return: Dictionary for the current state of the wallet, including sequence number and balances.
        """
        return self.get_account(swth_address=self.wallet.address)

    def get_transaction_fee_type(self, transaction_type: str):
        return types.fee_types[transaction_type]

    def submit_transaction_on_chain(self, messages: list, transaction_type: str, fee: dict):
        """
        This is the function that every function in higher classes should call.
        Every function in the authenticated_client calls this function.
        It standardizes the transaction construction and the signing and broadcasting.

        Execution of this function is as follows::

            submit_transaction_on_chain(messages=[message], transaction_type="UPDATE_PROFILE_MSG_TYPE", fee=None)

        :return: Dictionary for the transaction submitted on-chain.
        """
        messages_list, transactions_list, fee_dict = self.set_transaction_standards(messages=messages,
                                                                                    transaction_type=transaction_type,
                                                                                    fee=fee)
        return self.sign_and_broadcast(messages=messages_list, transaction_types=transactions_list, fee=fee_dict)

    def set_transaction_standards(self, messages: list, transaction_type: str, fee: dict):
        """
        This function sets standards for the messages in the transaction submitted to the chain.
        Fees, Integer precision as string, and Transaction messages are required.

        Execution of this function is as follows::

            submit_transaction_standards(messages=[message], transaction_type="UPDATE_PROFILE_MSG_TYPE", fee=None)

        :return: Three return variables, 1 for each input but run through standardizations.
        """
        messages_list = self.set_message_standards(messages=messages)
        tradehub_transaction_type = types.transaction_types[transaction_type]
        transactions_list = [tradehub_transaction_type] * len(messages_list)
        fee_dict = self.set_fees(transaction_cnt=len(messages), transaction_type=tradehub_transaction_type, fee=fee)
        return messages_list, transactions_list, fee_dict

    def set_message_standards(self, messages: list):
        """
        Standardize messages, adding an originator or takingo ut keys that don't exist.

        Execution of this function is as follows::

            set_message_standards(messages=[message])

        :return: Three return variables, 1 for each input but run through standardizations.
        """
        messages_list = []
        for message in messages:
            if hasattr(message, 'originator') and message.originator in [None, ""]:
                message.originator = self.wallet.address
            message_json = jsons.dump(message)
            message_dict = {}
            for key, value in message_json.items():
                if message_json[key] is not None:
                    message_dict[key] = value
            messages_list.append(message_dict)
        return messages_list

    def set_fees(self, transaction_cnt: int, transaction_type: str, fee: dict):
        """
        Standardize, find, and calculate fees for the number of transactions.

        Execution of this function is as follows::

            set_fees(transaction_cnt=1, transaction_type="UPDATE_PROFILE_MSG_TYPE", fee=None)

        :return: Dictionary for the fee's that will be needed for this transaction.
        """
        if fee:
            fee_dict = fee
        else:
            fee_type = self.get_transaction_fee_type(transaction_type)
            fee_amount = str(int(self.fees[fee_type]) * transaction_cnt)
            fee_dict = {
                "amount": [{"amount": fee_amount, "denom": "swth"}],
                "gas": self.gas,
            }
        return fee_dict

    def sign_and_broadcast(self, messages: list, transaction_types: list, fee: dict):   # Eventually need to add memo to this.
        """
        Now that messages are standardized we need to sign the transaction and broadcast it.
        The actual signature part will be handled by the Wallet Client to avoid leaking keys between classes.

        Execution of this function is as follows::

            sign_and_broadcast(messages=messages_list, transaction_types=transactions_list, fee=fee_dict)

        :return: Dictionary for the transaction submitted on-chain.
        """
        transactions = self.sign_transaction(messages=messages, transaction_types=transaction_types, fee=fee)
        broadcast_response = self.broadcast_transactions(transactions=transactions)
        if 'code' not in broadcast_response:
            self.account_sequence_nbr = str(int(self.account_sequence_nbr) + 1)
        elif broadcast_response['code'] == '3':
            self.update_account_details()
        elif broadcast_response['code'] == '19':
            self.account_sequence_nbr = str(int(self.account_sequence_nbr) + 1)
        return broadcast_response

    def sign_transaction(self,
                         messages: list,
                         transaction_types: list,
                         memo: str = None,
                         fee: dict = None):
        """
        A Signature has the following sequence.  https://docs.switcheo.org/#/?id=authentication
        (1) Construct a list of dictionaries combining message and message types together. <- construct_concrete_messages
        (2) Sign the list of messages generated in step (1). <- sign_message
        (3) Take the signature from step (2) and create a signature JSON message. <- construct_signatures
        (4) Take the signature JSON from step (3) and wrape it in a transaction JSON message. <- construct_transaction
        (5) Take the transaction JSON from step (4) and create the final layer of the transaction JSON message. <- construct_complete_transaction

        Execution of this function is as follows::

            sign_transaction(messages=messages_list, transaction_types=transactions_list, fee=fee_dict)

        :return: Dictionary for a complete transaction message.
        """
        concrete_messages = self.construct_concrete_messages(messages=messages, transaction_types=transaction_types)
        signature = self.sign_message(messages=concrete_messages, memo=memo, fee=fee)
        signatures = self.construct_signatures(signature=signature)
        transaction = self.construct_transaction(message=concrete_messages, signatures=[signatures], fees=fee)
        return self.construct_complete_transaction(transaction=transaction)

    def construct_concrete_messages(self, messages: list, transaction_types: list):  # both of these are lists of strings
        """
        The first step to building a transaction is merging the messages with the transaction types for each.
        Once constructed then each message can be sent off for signature.

        Execution of this function is as follows::

            construct_concrete_messages(messages=messages_list, transaction_types=transactions_list)

        :return: List of dictionaries of messages with transaction types.
        """
        if len(messages) != len(transaction_types):
            raise ValueError('Message length is not equal to transaction types length.')
        if len(messages) > 100:
            raise ValueError('Cannot broadcast more than 100 messages in 1 transaction')

        concrete_messages = []   # ConcreteMsg[] from JS code -> {type: string, value: object}

        for (message, transaction_type) in zip(messages, transaction_types):
            concrete_messages.append({
                "type": transaction_type,
                "value": message,
            })

        return concrete_messages

    def sign_message(self,
                     messages: list,
                     memo: str = None,
                     fee: dict = None):
        """
        Sign the messages constructed in the previous function.

        Execution of this function is as follows::

            sign_message(messages=messages_list, memo=None, fee=None)

        :return: String as a result of the signed messages.
        """
        if self.account_sequence_nbr is None or self.account_nbr is None or self.account_nbr == '0':  # no sequence override, get latest from blockchain
            self.update_account_details()
            if self.account_nbr == '0' or self.account_nbr is None:
                raise ValueError('Account number still 0 after refetching. This suggests your account is not initialized with funds.')

        fee_dict = self.set_fees(transaction_cnt=len(messages), transaction_type=messages[0]["type"], fee=fee)

        constructed_signing_message = {
            "account_number": self.account_nbr,
            "chain_id": self.chain_id,
            "fee": fee_dict,
            "memo": memo if memo else '',
            "msgs": messages,
            "sequence": self.account_sequence_nbr,
        }

        return self.wallet._sign(message=constructed_signing_message)

    def construct_signatures(self, signature: str):
        return {
            "pub_key": {"type": "tendermint/PubKeySecp256k1", "value": self.wallet.base64_public_key},
            "signature": signature,
        }

    def construct_transaction(self, message: list, signatures: list, fees: dict, memo: str = None):
        return {
            "fee": fees,
            "msg": message,
            "memo": memo if memo else '',
            "signatures": signatures
        }

    def construct_complete_transaction(self, transaction: dict, mode: str = None):
        return {
            "mode": mode if mode else self.mode,
            "tx": transaction,
        }

    def broadcast_transactions(self, transactions: dict):
        """
        Now that messages are standardized we need to sign the transaction and broadcast it.
        The actual signature part will be handled by the Wallet Client to avoid leaking keys between classes.

        Execution of this function is as follows::

            broadcast_transactions(transactions=transactions)

        The expected return result for this function is as follows::

        {
            'height': str,
            'txhash': str,
            'raw_log': str,
            'logs': [{
                'msg_index': int,
                'log': str,
                'events': [{
                    'type': str,
                    'attributes': [{
                        'key': str,
                        'value': str
                    }]
                }]
            }],
            'gas_wanted': str,
            'gas_used': str
        }

        :return: Dictionary for the transaction submitted on-chain.
        """
        return self.tradehub_post_request(path='/txs', json_data=transactions)


1			"""
2			Description:
3
4			Transaction Class used as part of the Authenticated Client to standardize how on-chain messages are constructed.
5			This client is the basis any on-chain transaction and handles message construction, standards, signing, and broadcasting.
6
7			Usage::
8
9			from tradehub.transactions import Transactions
10			"""
11			import jsons
12
13			from tradehub.public_account_client import PublicClient as TradehubPublicClient
14			import tradehub.types as types
15			from tradehub.wallet import Wallet
16
17
18			class Transactions(TradehubPublicClient):
19			"""
20			This class constructs, standardizes, signs, and broadcasts messages to the Tradehub network.
21
22			Execution of this function is as follows::
23
24			Transactions(wallet=Wallet(),
25			trusted_ips=None,
26			trusted_uris=None,
27			network="mainnet")
28			"""
29
30			def __init__(self, wallet: Wallet, trusted_ips: list = None, trusted_uris: list = None, network: str = "testnet"):
31			"""
32			:param wallet: Wallet class for Tradehub interaction.
33			:type wallet: Wallet
34			:param trusted_ips: Known and trusted IPs to connect to for your API requests.
35			:type trusted_ips: list
36			:param trusted_uris: Known and trusted URIs to connect to for your API requests.
37			:type trusted_uris: list
38			:param network: Network to submit the transaction.
39			:type network: str
40			"""
41			TradehubPublicClient.__init__(self, network=network, trusted_ips=trusted_ips, trusted_uris=trusted_uris)
42			self.wallet = wallet
43			self.update_account_details()
44			self.network_variables = {
45			"testnet": {"chain_id": "switcheochain", },
46			"mainnet": {"chain_id": "switcheo-tradehub-1", },
47			}
48			self.chain_id = self.network_variables[network]["chain_id"]
49			self.fees = self.get_transactions_fees()
50			self.mode = "block" # Need to automate
51			self.gas = "100000000000" # Need to automate
52			self.tokens = self.get_token_details()
53
54			def update_account_details(self) -> None:
55			"""
56			Triggers an update to fetch current sequence number.
57
58			:return: None
59			"""
60			self.account_blockchain_dict = self.get_account_details()
61			self.account_nbr = self.account_blockchain_dict["result"]["value"]["account_number"]
62			self.account_sequence_nbr = self.account_blockchain_dict["result"]["value"]["sequence"]
63
64			def get_account_details(self):
65			"""
66			Retrieve Wallet account details required for submitting transactions on Tradehub.
67
68			Execution of this function is as follows::
69
70			get_account_details()
71
72			The expected return result for this function is as follows::
73
74			{
75			'height': '1950991',
76			'result': {
77			'type': 'cosmos-sdk/Account',
78			'value': {
79			'address': 'tswth1urfldmcspc9nk4w8vcfakxdf2rc5ee4fr2dn76',
80			'coins': [{
81			'denom': 'btc',
82			'amount': '99033160'
83			}, {
84			...
85			}],
86			'public_key': {
87			'type': 'tendermint/PubKeySecp256k1',
88			'value': 'Ao8h4bhZr1/m8ZEwPaOizNCcIMDX/yqwQh7LXhI77FLW'
89			},
90			'account_number': '56',
91			'sequence': '3195'
92			}
93			}
94			}
95
96			:return: Dictionary for the current state of the wallet, including sequence number and balances.
97			"""
98			return self.get_account(swth_address=self.wallet.address)
99
100			def get_transaction_fee_type(self, transaction_type: str):
101			return types.fee_types[transaction_type]
102
103			def submit_transaction_on_chain(self, messages: list, transaction_type: str, fee: dict):
104			"""
105			This is the function that every function in higher classes should call.
106			Every function in the authenticated_client calls this function.
107			It standardizes the transaction construction and the signing and broadcasting.
108
109			Execution of this function is as follows::
110
111			submit_transaction_on_chain(messages=[message], transaction_type="UPDATE_PROFILE_MSG_TYPE", fee=None)
112
113			:return: Dictionary for the transaction submitted on-chain.
114			"""
115			messages_list, transactions_list, fee_dict = self.set_transaction_standards(messages=messages,
116			transaction_type=transaction_type,
117			fee=fee)
118			return self.sign_and_broadcast(messages=messages_list, transaction_types=transactions_list, fee=fee_dict)
119
120			def set_transaction_standards(self, messages: list, transaction_type: str, fee: dict):
121			"""
122			This function sets standards for the messages in the transaction submitted to the chain.
123			Fees, Integer precision as string, and Transaction messages are required.
124
125			Execution of this function is as follows::
126
127			submit_transaction_standards(messages=[message], transaction_type="UPDATE_PROFILE_MSG_TYPE", fee=None)
128
129			:return: Three return variables, 1 for each input but run through standardizations.
130			"""
131			messages_list = self.set_message_standards(messages=messages)
132			tradehub_transaction_type = types.transaction_types[transaction_type]
133			transactions_list = [tradehub_transaction_type] * len(messages_list)
134			fee_dict = self.set_fees(transaction_cnt=len(messages), transaction_type=tradehub_transaction_type, fee=fee)
135			return messages_list, transactions_list, fee_dict
136
137			def set_message_standards(self, messages: list):
138			"""
139			Standardize messages, adding an originator or takingo ut keys that don't exist.
140
141			Execution of this function is as follows::
142
143			set_message_standards(messages=[message])
144
145			:return: Three return variables, 1 for each input but run through standardizations.
146			"""
147			messages_list = []
148			for message in messages:
149			if hasattr(message, 'originator') and message.originator in [None, ""]:
150			message.originator = self.wallet.address
151			message_json = jsons.dump(message)
152			message_dict = {}
153			for key, value in message_json.items():
154			if message_json[key] is not None:
155			message_dict[key] = value
156			messages_list.append(message_dict)
157			return messages_list
158
159			def set_fees(self, transaction_cnt: int, transaction_type: str, fee: dict):
160			"""
161			Standardize, find, and calculate fees for the number of transactions.
162
163			Execution of this function is as follows::
164
165			set_fees(transaction_cnt=1, transaction_type="UPDATE_PROFILE_MSG_TYPE", fee=None)
166
167			:return: Dictionary for the fee's that will be needed for this transaction.
168			"""
169			if fee:
170			fee_dict = fee
171			else:
172			fee_type = self.get_transaction_fee_type(transaction_type)
173			fee_amount = str(int(self.fees[fee_type]) * transaction_cnt)
174			fee_dict = {
175			"amount": [{"amount": fee_amount, "denom": "swth"}],
176			"gas": self.gas,
177			}
178			return fee_dict
179
180			def sign_and_broadcast(self, messages: list, transaction_types: list, fee: dict): # Eventually need to add memo to this.
181			"""
182			Now that messages are standardized we need to sign the transaction and broadcast it.
183			The actual signature part will be handled by the Wallet Client to avoid leaking keys between classes.
184
185			Execution of this function is as follows::
186
187			sign_and_broadcast(messages=messages_list, transaction_types=transactions_list, fee=fee_dict)
188
189			:return: Dictionary for the transaction submitted on-chain.
190			"""
191			transactions = self.sign_transaction(messages=messages, transaction_types=transaction_types, fee=fee)
192			broadcast_response = self.broadcast_transactions(transactions=transactions)
193			if 'code' not in broadcast_response:
194			self.account_sequence_nbr = str(int(self.account_sequence_nbr) + 1)
195			elif broadcast_response['code'] == '3':
196			self.update_account_details()
197			elif broadcast_response['code'] == '19':
198			self.account_sequence_nbr = str(int(self.account_sequence_nbr) + 1)
199			return broadcast_response
200
201			def sign_transaction(self,
202			messages: list,
203			transaction_types: list,
204			memo: str = None,
205			fee: dict = None):
206			"""
207			A Signature has the following sequence. https://docs.switcheo.org/#/?id=authentication
208			(1) Construct a list of dictionaries combining message and message types together. <- construct_concrete_messages
209			(2) Sign the list of messages generated in step (1). <- sign_message
210			(3) Take the signature from step (2) and create a signature JSON message. <- construct_signatures
211			(4) Take the signature JSON from step (3) and wrape it in a transaction JSON message. <- construct_transaction
212			(5) Take the transaction JSON from step (4) and create the final layer of the transaction JSON message. <- construct_complete_transaction
213
214			Execution of this function is as follows::
215
216			sign_transaction(messages=messages_list, transaction_types=transactions_list, fee=fee_dict)
217
218			:return: Dictionary for a complete transaction message.
219			"""
220			concrete_messages = self.construct_concrete_messages(messages=messages, transaction_types=transaction_types)
221			signature = self.sign_message(messages=concrete_messages, memo=memo, fee=fee)
222			signatures = self.construct_signatures(signature=signature)
223			transaction = self.construct_transaction(message=concrete_messages, signatures=[signatures], fees=fee)
224			return self.construct_complete_transaction(transaction=transaction)
225
226			def construct_concrete_messages(self, messages: list, transaction_types: list): # both of these are lists of strings
227			"""
228			The first step to building a transaction is merging the messages with the transaction types for each.
229			Once constructed then each message can be sent off for signature.
230
231			Execution of this function is as follows::
232
233			construct_concrete_messages(messages=messages_list, transaction_types=transactions_list)
234
235			:return: List of dictionaries of messages with transaction types.
236			"""
237			if len(messages) != len(transaction_types):
238			raise ValueError('Message length is not equal to transaction types length.')
239			if len(messages) > 100:
240			raise ValueError('Cannot broadcast more than 100 messages in 1 transaction')
241
242			concrete_messages = [] # ConcreteMsg[] from JS code -> {type: string, value: object}
243
244			for (message, transaction_type) in zip(messages, transaction_types):
245			concrete_messages.append({
246			"type": transaction_type,
247			"value": message,
248			})
249
250			return concrete_messages
251
252			def sign_message(self,
253			messages: list,
254			memo: str = None,
255			fee: dict = None):
256			"""
257			Sign the messages constructed in the previous function.
258
259			Execution of this function is as follows::
260
261			sign_message(messages=messages_list, memo=None, fee=None)
262
263			:return: String as a result of the signed messages.
264			"""
265			if self.account_sequence_nbr is None or self.account_nbr is None or self.account_nbr == '0': # no sequence override, get latest from blockchain
266			self.update_account_details()
267			if self.account_nbr == '0' or self.account_nbr is None:
268			raise ValueError('Account number still 0 after refetching. This suggests your account is not initialized with funds.')
269
270			fee_dict = self.set_fees(transaction_cnt=len(messages), transaction_type=messages[0]["type"], fee=fee)
271
272			constructed_signing_message = {
273			"account_number": self.account_nbr,
274			"chain_id": self.chain_id,
275			"fee": fee_dict,
276			"memo": memo if memo else '',
277			"msgs": messages,
278			"sequence": self.account_sequence_nbr,
279			}
280
281			return self.wallet._sign(message=constructed_signing_message)
282
283			def construct_signatures(self, signature: str):
284			return {
285			"pub_key": {"type": "tendermint/PubKeySecp256k1", "value": self.wallet.base64_public_key},
286			"signature": signature,
287			}
288
289			def construct_transaction(self, message: list, signatures: list, fees: dict, memo: str = None):
290			return {
291			"fee": fees,
292			"msg": message,
293			"memo": memo if memo else '',
294			"signatures": signatures
295			}
296
297			def construct_complete_transaction(self, transaction: dict, mode: str = None):
298			return {
299			"mode": mode if mode else self.mode,
300			"tx": transaction,
301			}
302
303			def broadcast_transactions(self, transactions: dict):
304			"""
305			Now that messages are standardized we need to sign the transaction and broadcast it.
306			The actual signature part will be handled by the Wallet Client to avoid leaking keys between classes.
307
308			Execution of this function is as follows::
309
310			broadcast_transactions(transactions=transactions)
311
312			The expected return result for this function is as follows::
313
314			{
315			'height': str,
316			'txhash': str,
317			'raw_log': str,
318			'logs': [{
319			'msg_index': int,
320			'log': str,
321			'events': [{
322			'type': str,
323			'attributes': [{
324			'key': str,
325			'value': str
326			}]
327			}]
328			}],
329			'gas_wanted': str,
330			'gas_used': str
331			}
332
333			:return: Dictionary for the transaction submitted on-chain.
334			"""
335			return self.tradehub_post_request(path='/txs', json_data=transactions)
336

Mai-Te-Pora / tradehub-python

Transactions.submit_transaction_on_chain() A last analyzed 2021-03-26 05:54 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

Transactions.submit_transaction_on_chain() A
last analyzed 2021-03-26 05:54 UTC