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

Completed

Push — main ( 4fccdc...3e84ec )

unknown

created 2021-02-22 08:52 UTC

tradehub.transactions.Transactions.set_fees() A

↳ Parent: tradehub.transactions

Complexity

Conditions

Size

Total Lines	11
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	9
nop	4
dl	0
loc	11
rs	9.95
c	0
b	0
f	0

import jsons

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


class Transactions(TradehubPublicClient):

    def __init__(self, wallet: Wallet, node_ip: str, node_port: int = 5001, network: str = "testnet"):
        """
        """
        TradehubPublicClient.__init__(self, node_ip=node_ip, node_port=node_port)
        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 get_account_details(self):
        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):
        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):
        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):
        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):
        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.
        '''
            This is the entry point for all signatures in this Class.
            All the signatures should be handled in the Wallet Client to avoid leaking keys.

        '''
        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_concreate_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. <- construact_transaction
            (5) Take the transaction JSON from step (4) and create the final layer of the transaction JSON message. <- construct_complete_transaction
        '''

        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
        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):

        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):
        return self.request.post(path='/txs', json_data=transactions)


1			import jsons
2
3			from tradehub.public_client import PublicClient as TradehubPublicClient
4			import tradehub.types as types
5			from tradehub.wallet import Wallet
6
7
8			class Transactions(TradehubPublicClient):
9
10			def __init__(self, wallet: Wallet, node_ip: str, node_port: int = 5001, network: str = "testnet"):
11			"""
12			"""
13			TradehubPublicClient.__init__(self, node_ip=node_ip, node_port=node_port)
14			self.wallet = wallet
15			self.account_blockchain_dict = self.get_account_details()
16			self.account_nbr = self.account_blockchain_dict["result"]["value"]["account_number"]
17			self.account_sequence_nbr = self.account_blockchain_dict["result"]["value"]["sequence"]
18			self.network_variables = {
19			"testnet": {"chain_id": "switcheochain", },
20			"mainnet": {"chain_id": "switcheo-tradehub-1", },
21			}
22			self.chain_id = self.network_variables[network]["chain_id"]
23			self.fees = self.get_transactions_fees()
24			self.mode = "block" # Need to automate
25			self.gas = "100000000000" # Need to automate
26			self.tokens = self.get_token_details()
27
28			def get_account_details(self):
29			return self.get_account(swth_address=self.wallet.address)
30
31			def get_transaction_fee_type(self, transaction_type: str):
32			return types.fee_types[transaction_type]
33
34			def submit_transaction_on_chain(self, messages: list, transaction_type: str, fee: dict):
35			messages_list, transactions_list, fee_dict = self.set_transaction_standards(messages=messages,
36			transaction_type=transaction_type,
37			fee=fee)
38			return self.sign_and_broadcast(messages=messages_list, transaction_types=transactions_list, fee=fee_dict)
39
40			def set_transaction_standards(self, messages: list, transaction_type: str, fee: dict):
41			messages_list = self.set_message_standards(messages=messages)
42			tradehub_transaction_type = types.transaction_types[transaction_type]
43			transactions_list = [tradehub_transaction_type] * len(messages_list)
44			fee_dict = self.set_fees(transaction_cnt=len(messages), transaction_type=tradehub_transaction_type, fee=fee)
45			return messages_list, transactions_list, fee_dict
46
47			def set_message_standards(self, messages: list):
48			messages_list = []
49			for message in messages:
50			if hasattr(message, 'originator') and message.originator in [None, ""]:
51			message.originator = self.wallet.address
52			message_json = jsons.dump(message)
53			message_dict = {}
54			for key, value in message_json.items():
55			if message_json[key] is not None:
56			message_dict[key] = value
57			messages_list.append(message_dict)
58			return messages_list
59
60			def set_fees(self, transaction_cnt: int, transaction_type: str, fee: dict):
61			if fee:
62			fee_dict = fee
63			else:
64			fee_type = self.get_transaction_fee_type(transaction_type)
65			fee_amount = str(int(self.fees[fee_type]) * transaction_cnt)
66			fee_dict = {
67			"amount": [{"amount": fee_amount, "denom": "swth"}],
68			"gas": self.gas,
69			}
70			return fee_dict
71
72			def sign_and_broadcast(self, messages: list, transaction_types: list, fee: dict): # Eventually need to add memo to this.
73			'''
74			This is the entry point for all signatures in this Class.
75			All the signatures should be handled in the Wallet Client to avoid leaking keys.
76
77			'''
78			transactions = self.sign_transaction(messages=messages, transaction_types=transaction_types, fee=fee)
79			broadcast_response = self.broadcast_transactions(transactions=transactions)
80			if 'code' not in broadcast_response:
81			self.account_sequence_nbr = str(int(self.account_sequence_nbr) + 1)
82			return broadcast_response
83
84			def sign_transaction(self,
85			messages: list,
86			transaction_types: list,
87			memo: str = None,
88			fee: dict = None):
89			'''
90			A Signature has the following sequence. https://docs.switcheo.org/#/?id=authentication
91			(1) Construct a list of dictionaries combining message and message types together. <- construct_concreate_messages
92			(2) Sign the list of messages generated in step (1). <- sign_message
93			(3) Take the signature from step (2) and create a signature JSON message. <- construct_signatures
94			(4) Take the signature JSON from step (3) and wrape it in a transaction JSON message. <- construact_transaction
95			(5) Take the transaction JSON from step (4) and create the final layer of the transaction JSON message. <- construct_complete_transaction
96			'''
97
98			concrete_messages = self.construct_concrete_messages(messages=messages, transaction_types=transaction_types)
99			signature = self.sign_message(messages=concrete_messages, memo=memo, fee=fee)
100			signatures = self.construct_signatures(signature=signature)
101			transaction = self.construct_transaction(message=concrete_messages, signatures=[signatures], fees=fee)
102			return self.construct_complete_transaction(transaction=transaction)
103
104			def construct_concrete_messages(self, messages: list, transaction_types: list): # both of these are lists of strings
105			if len(messages) != len(transaction_types):
106			raise ValueError('Message length is not equal to transaction types length.')
107			if len(messages) > 100:
108			raise ValueError('Cannot broadcast more than 100 messages in 1 transaction')
109
110			concrete_messages = [] # ConcreteMsg[] from JS code -> {type: string, value: object}
111
112			for (message, transaction_type) in zip(messages, transaction_types):
113			concrete_messages.append({
114			"type": transaction_type,
115			"value": message,
116			})
117
118			return concrete_messages
119
120			def sign_message(self,
121			messages: list,
122			memo: str = None,
123			fee: dict = None):
124
125			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
126			self.account_blockchain_dict = self.get_account_details()
127			self.account_nbr = self.account_blockchain_dict["result"]["value"]["account_number"]
128			self.account_sequence_nbr = self.account_blockchain_dict["result"]["value"]["sequence"]
129			if self.account_nbr == '0' or self.account_nbr is None:
130			raise ValueError('Account number still 0 after refetching. This suggests your account is not initialized with funds.')
131
132			fee_dict = self.set_fees(transaction_cnt=len(messages), transaction_type=messages[0]["type"], fee=fee)
133
134			constructed_signing_message = {
135			"account_number": self.account_nbr,
136			"chain_id": self.chain_id,
137			"fee": fee_dict,
138			"memo": memo if memo else '',
139			"msgs": messages,
140			"sequence": self.account_sequence_nbr,
141			}
142
143			return self.wallet._sign(message=constructed_signing_message)
144
145			def construct_signatures(self, signature: str):
146			return {
147			"pub_key": {"type": "tendermint/PubKeySecp256k1", "value": self.wallet.base64_public_key},
148			"signature": signature,
149			}
150
151			def construct_transaction(self, message: list, signatures: list, fees: dict, memo: str = None):
152			return {
153			"fee": fees,
154			"msg": message,
155			"memo": memo if memo else '',
156			"signatures": signatures
157			}
158
159			def construct_complete_transaction(self, transaction: dict, mode: str = None):
160			return {
161			"mode": mode if mode else self.mode,
162			"tx": transaction,
163			}
164
165			def broadcast_transactions(self, transactions: dict):
166			return self.request.post(path='/txs', json_data=transactions)
167

Mai-Te-Pora / tradehub-python

Push — main ( 4fccdc...3e84ec )

tradehub.transactions.Transactions.set_fees() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like