tradehub.transactions.Transactions.construct_transaction() - Code Metrics - Inspection of "Merge pull request #48 from Mai-Te-Pora/bugfix/err..." - Mai-Te-Pora/tradehub-python - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — main ( 40d33d...2ccf5e )

by Switcheolytics

created 2021-03-22 20:34 UTC

Transactions.construct_transaction() A

↳ Parent: tradehub.transactions

Complexity

Conditions

Size

Total Lines	6
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	6
nop	5
dl	0
loc	6
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.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"]
        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)
        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.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"]
            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.account_blockchain_dict = self.get_account_details()
44			self.account_nbr = self.account_blockchain_dict["result"]["value"]["account_number"]
45			self.account_sequence_nbr = self.account_blockchain_dict["result"]["value"]["sequence"]
46			self.network_variables = {
47			"testnet": {"chain_id": "switcheochain", },
48			"mainnet": {"chain_id": "switcheo-tradehub-1", },
49			}
50			self.chain_id = self.network_variables[network]["chain_id"]
51			self.fees = self.get_transactions_fees()
52			self.mode = "block" # Need to automate
53			self.gas = "100000000000" # Need to automate
54			self.tokens = self.get_token_details()
55
56			def update_account_details(self) -> None:
57			"""
58			Triggers an update to fetch current sequence number.
59
60			:return: None
61			"""
62			self.account_blockchain_dict = self.get_account_details()
63			self.account_nbr = self.account_blockchain_dict["result"]["value"]["account_number"]
64			self.account_sequence_nbr = self.account_blockchain_dict["result"]["value"]["sequence"]
65
66			def get_account_details(self):
67			"""
68			Retrieve Wallet account details required for submitting transactions on Tradehub.
69
70			Execution of this function is as follows::
71
72			get_account_details()
73
74			The expected return result for this function is as follows::
75
76			{
77			'height': '1950991',
78			'result': {
79			'type': 'cosmos-sdk/Account',
80			'value': {
81			'address': 'tswth1urfldmcspc9nk4w8vcfakxdf2rc5ee4fr2dn76',
82			'coins': [{
83			'denom': 'btc',
84			'amount': '99033160'
85			}, {
86			...
87			}],
88			'public_key': {
89			'type': 'tendermint/PubKeySecp256k1',
90			'value': 'Ao8h4bhZr1/m8ZEwPaOizNCcIMDX/yqwQh7LXhI77FLW'
91			},
92			'account_number': '56',
93			'sequence': '3195'
94			}
95			}
96			}
97
98			:return: Dictionary for the current state of the wallet, including sequence number and balances.
99			"""
100			return self.get_account(swth_address=self.wallet.address)
101
102			def get_transaction_fee_type(self, transaction_type: str):
103			return types.fee_types[transaction_type]
104
105			def submit_transaction_on_chain(self, messages: list, transaction_type: str, fee: dict):
106			"""
107			This is the function that every function in higher classes should call.
108			Every function in the authenticated_client calls this function.
109			It standardizes the transaction construction and the signing and broadcasting.
110
111			Execution of this function is as follows::
112
113			submit_transaction_on_chain(messages=[message], transaction_type="UPDATE_PROFILE_MSG_TYPE", fee=None)
114
115			:return: Dictionary for the transaction submitted on-chain.
116			"""
117			messages_list, transactions_list, fee_dict = self.set_transaction_standards(messages=messages,
118			transaction_type=transaction_type,
119			fee=fee)
120			return self.sign_and_broadcast(messages=messages_list, transaction_types=transactions_list, fee=fee_dict)
121
122			def set_transaction_standards(self, messages: list, transaction_type: str, fee: dict):
123			"""
124			This function sets standards for the messages in the transaction submitted to the chain.
125			Fees, Integer precision as string, and Transaction messages are required.
126
127			Execution of this function is as follows::
128
129			submit_transaction_standards(messages=[message], transaction_type="UPDATE_PROFILE_MSG_TYPE", fee=None)
130
131			:return: Three return variables, 1 for each input but run through standardizations.
132			"""
133			messages_list = self.set_message_standards(messages=messages)
134			tradehub_transaction_type = types.transaction_types[transaction_type]
135			transactions_list = [tradehub_transaction_type] * len(messages_list)
136			fee_dict = self.set_fees(transaction_cnt=len(messages), transaction_type=tradehub_transaction_type, fee=fee)
137			return messages_list, transactions_list, fee_dict
138
139			def set_message_standards(self, messages: list):
140			"""
141			Standardize messages, adding an originator or takingo ut keys that don't exist.
142
143			Execution of this function is as follows::
144
145			set_message_standards(messages=[message])
146
147			:return: Three return variables, 1 for each input but run through standardizations.
148			"""
149			messages_list = []
150			for message in messages:
151			if hasattr(message, 'originator') and message.originator in [None, ""]:
152			message.originator = self.wallet.address
153			message_json = jsons.dump(message)
154			message_dict = {}
155			for key, value in message_json.items():
156			if message_json[key] is not None:
157			message_dict[key] = value
158			messages_list.append(message_dict)
159			return messages_list
160
161			def set_fees(self, transaction_cnt: int, transaction_type: str, fee: dict):
162			"""
163			Standardize, find, and calculate fees for the number of transactions.
164
165			Execution of this function is as follows::
166
167			set_fees(transaction_cnt=1, transaction_type="UPDATE_PROFILE_MSG_TYPE", fee=None)
168
169			:return: Dictionary for the fee's that will be needed for this transaction.
170			"""
171			if fee:
172			fee_dict = fee
173			else:
174			fee_type = self.get_transaction_fee_type(transaction_type)
175			fee_amount = str(int(self.fees[fee_type]) * transaction_cnt)
176			fee_dict = {
177			"amount": [{"amount": fee_amount, "denom": "swth"}],
178			"gas": self.gas,
179			}
180			return fee_dict
181
182			def sign_and_broadcast(self, messages: list, transaction_types: list, fee: dict): # Eventually need to add memo to this.
183			"""
184			Now that messages are standardized we need to sign the transaction and broadcast it.
185			The actual signature part will be handled by the Wallet Client to avoid leaking keys between classes.
186
187			Execution of this function is as follows::
188
189			sign_and_broadcast(messages=messages_list, transaction_types=transactions_list, fee=fee_dict)
190
191			:return: Dictionary for the transaction submitted on-chain.
192			"""
193			transactions = self.sign_transaction(messages=messages, transaction_types=transaction_types, fee=fee)
194			broadcast_response = self.broadcast_transactions(transactions=transactions)
195			if 'code' not in broadcast_response:
196			self.account_sequence_nbr = str(int(self.account_sequence_nbr) + 1)
197			return broadcast_response
198
199			def sign_transaction(self,
200			messages: list,
201			transaction_types: list,
202			memo: str = None,
203			fee: dict = None):
204			"""
205			A Signature has the following sequence. https://docs.switcheo.org/#/?id=authentication
206			(1) Construct a list of dictionaries combining message and message types together. <- construct_concrete_messages
207			(2) Sign the list of messages generated in step (1). <- sign_message
208			(3) Take the signature from step (2) and create a signature JSON message. <- construct_signatures
209			(4) Take the signature JSON from step (3) and wrape it in a transaction JSON message. <- construct_transaction
210			(5) Take the transaction JSON from step (4) and create the final layer of the transaction JSON message. <- construct_complete_transaction
211
212			Execution of this function is as follows::
213
214			sign_transaction(messages=messages_list, transaction_types=transactions_list, fee=fee_dict)
215
216			:return: Dictionary for a complete transaction message.
217			"""
218			concrete_messages = self.construct_concrete_messages(messages=messages, transaction_types=transaction_types)
219			signature = self.sign_message(messages=concrete_messages, memo=memo, fee=fee)
220			signatures = self.construct_signatures(signature=signature)
221			transaction = self.construct_transaction(message=concrete_messages, signatures=[signatures], fees=fee)
222			return self.construct_complete_transaction(transaction=transaction)
223
224			def construct_concrete_messages(self, messages: list, transaction_types: list): # both of these are lists of strings
225			"""
226			The first step to building a transaction is merging the messages with the transaction types for each.
227			Once constructed then each message can be sent off for signature.
228
229			Execution of this function is as follows::
230
231			construct_concrete_messages(messages=messages_list, transaction_types=transactions_list)
232
233			:return: List of dictionaries of messages with transaction types.
234			"""
235			if len(messages) != len(transaction_types):
236			raise ValueError('Message length is not equal to transaction types length.')
237			if len(messages) > 100:
238			raise ValueError('Cannot broadcast more than 100 messages in 1 transaction')
239
240			concrete_messages = [] # ConcreteMsg[] from JS code -> {type: string, value: object}
241
242			for (message, transaction_type) in zip(messages, transaction_types):
243			concrete_messages.append({
244			"type": transaction_type,
245			"value": message,
246			})
247
248			return concrete_messages
249
250			def sign_message(self,
251			messages: list,
252			memo: str = None,
253			fee: dict = None):
254			"""
255			Sign the messages constructed in the previous function.
256
257			Execution of this function is as follows::
258
259			sign_message(messages=messages_list, memo=None, fee=None)
260
261			:return: String as a result of the signed messages.
262			"""
263			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
264			self.account_blockchain_dict = self.get_account_details()
265			self.account_nbr = self.account_blockchain_dict["result"]["value"]["account_number"]
266			self.account_sequence_nbr = self.account_blockchain_dict["result"]["value"]["sequence"]
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

Push — main ( 40d33d...2ccf5e )

Transactions.construct_transaction() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like