[lightapi/python] task: refactor, add transaction handler

lightapi/python/symbollightapi/connector/NemConnector.py

@@ -1,21 +1,13 @@
-from collections import namedtuple
-
 from symbolchain.CryptoTypes import PublicKey
 from symbolchain.nc import TransactionType
 from symbolchain.nem.Network import Address
 
 from ..model.Block import Block
 from ..model.Endpoint import Endpoint
 from ..model.NodeInfo import NodeInfo
-from ..model.Transaction import CosignSignatureTransaction, TransactionFactory
+from ..model.Transaction import TransactionFactory, TransactionHandler
 from .BasicConnector import BasicConnector
 
-Message = namedtuple('Message', ['payload', 'is_plain'])
-Mosaic = namedtuple('Mosaic', ['namespace_name', 'quantity'])
-Modification = namedtuple('Modification', ['modification_type', 'cosignatory_account'])
-MosaicLevy = namedtuple('MosaicLevy', ['fee', 'recipient', 'type', 'namespace_name'])
-MosaicProperties = namedtuple('MosaicProperties', ['divisibility', 'initial_supply', 'supply_mutable', 'transferable'])
-
 MICROXEM_PER_XEM = 1000000
 
 
@@ -117,8 +109,7 @@ async def get_block(self, height):
 		# block_dict = Block_1
 		return self._map_to_block(block)
 
-	@staticmethod
-	def _map_to_block(block_dict):
+	def _map_to_block(self, block_dict):
 		block = block_dict['block']
 		difficulty = block_dict['difficulty']
 		block_hash = block_dict['hash']
@@ -128,7 +119,7 @@ def _map_to_block(block_dict):
 			block['height'],
 			block['timeStamp'],
 			[
-				NemConnector._map_to_transaction(transaction, False, block['height'])
+				self._map_to_transaction(transaction, block['height'])
 				for transaction in transactions
 			],
 			difficulty,
@@ -138,124 +129,28 @@ def _map_to_block(block_dict):
 		)
 
 	@staticmethod
-	def _map_to_transaction(transaction, is_embedded=False, block_height=None):
+	def _map_to_transaction(transaction, block_height):
 		"""Maps a transaction dictionary to a transaction object."""
 
-		tx_dict = transaction if is_embedded else transaction['tx']
+		tx_dict = transaction['tx']
 		tx_type = tx_dict['type']
 
 		# Define common arguments for all transactions
 		common_args = {
-			'transaction_hash': None if is_embedded else transaction['hash'],
-			'height': None if is_embedded else block_height,
+			'transaction_hash': transaction['hash'],
+			'height': block_height,
 			'sender': tx_dict['signer'],
 			'fee': tx_dict['fee'],
 			'timestamp': tx_dict['timeStamp'],
 			'deadline': tx_dict['deadline'],
-			'signature': None if is_embedded else tx_dict['signature'],
+			'signature': tx_dict['signature'],
 		}
 
 		specific_args = {}
 
-		if TransactionType.TRANSFER.value == tx_type:
-			message = tx_dict['message']
-
-			if 'payload' in message and 'type' in message:
-				message = Message(
-					message['payload'],
-					message['type']
-				)
-
-			mosaics = None
-			if 'mosaics' in tx_dict:
-				mosaics = [
-					Mosaic(
-						f'{mosaic["mosaicId"]["namespaceId"]}.{mosaic["mosaicId"]["name"]}',
-						mosaic['quantity']
-					)
-					for mosaic in tx_dict['mosaics']
-				]
-
-			specific_args = {
-				'amount': tx_dict['amount'],
-				'recipient': tx_dict['recipient'],
-				'message': message,
-				'mosaics': mosaics,
-			}
-		elif TransactionType.ACCOUNT_KEY_LINK.value == tx_type:
-			specific_args = {
-				'mode': tx_dict['mode'],
-				'remote_account': tx_dict['remoteAccount'],
-			}
-		elif TransactionType.MULTISIG_ACCOUNT_MODIFICATION.value == tx_type:
-			specific_args = {
-				'min_cosignatories': tx_dict['minCosignatories']['relativeChange'],
-				'modifications': [
-					Modification(
-						modification['modificationType'],
-						modification['cosignatoryAccount'])
-					for modification in tx_dict['modifications']
-				]
-			}
-		elif TransactionType.MULTISIG_TRANSACTION.value == tx_type:
-			specific_args = {
-				'signatures': [
-					CosignSignatureTransaction(
-						signature['timeStamp'],
-						signature['otherHash']['data'],
-						signature['otherAccount'],
-						signature['signer'],
-						signature['fee'],
-						signature['deadline'],
-						signature['signature']
-					)
-					for signature in tx_dict['signatures']
-				],
-				'other_transaction': NemConnector._map_to_transaction(tx_dict['otherTrans'], True),
-				'inner_hash': transaction['innerHash'],
-			}
-		elif TransactionType.NAMESPACE_REGISTRATION.value == tx_type:
-			specific_args = {
-				'rental_fee_sink': tx_dict['rentalFeeSink'],
-				'rental_fee': tx_dict['rentalFee'],
-				'parent': tx_dict['parent'],
-				'namespace': tx_dict['newPart'],
-			}
-		elif TransactionType.MOSAIC_DEFINITION.value == tx_type:
-			mosaic_definition = tx_dict['mosaicDefinition']
-			mosaic_id = mosaic_definition['id']
-			mosaic_levy = mosaic_definition['levy']
-			mosaic_properties_dict = {
-				item['name']: item['value']
-				for item in mosaic_definition['properties']
-			}
-
-			specific_args = {
-				'creation_fee': tx_dict['creationFee'],
-				'creation_fee_sink': tx_dict['creationFeeSink'],
-				'creator': mosaic_definition['creator'],
-				'description': mosaic_definition['description'],
-				'properties': MosaicProperties(
-					int(mosaic_properties_dict['divisibility']),
-					int(mosaic_properties_dict['initialSupply']),
-					mosaic_properties_dict['supplyMutable'] != 'false',
-					mosaic_properties_dict['transferable'] != 'false'
-				),
-				'levy': MosaicLevy(
-					mosaic_levy['fee'],
-					mosaic_levy['recipient'],
-					mosaic_levy['type'],
-					f'{mosaic_levy["mosaicId"]["namespaceId"]}.{mosaic_levy["mosaicId"]["name"] }'
-				),
-				'namespace_name': f'{mosaic_id["namespaceId"]}.{mosaic_id["name"] }'
-			}
-		elif TransactionType.MOSAIC_SUPPLY_CHANGE.value == tx_type:
-			mosaic_id = tx_dict['mosaicId']
-
-			specific_args = {
-				'supply_type': tx_dict['supplyType'],
-				'delta': tx_dict['delta'],
-				'namespace_name': f'{mosaic_id["namespaceId"]}.{mosaic_id["name"] }',
-			}
+		if TransactionType.MULTISIG_TRANSACTION.value == tx_type:
+			specific_args = TransactionHandler().map[tx_type](tx_dict, transaction['innerHash'])
+		else:
+			specific_args = TransactionHandler().map[tx_type](tx_dict)
 
 		return TransactionFactory.create_transaction(tx_type, common_args, specific_args)

lightapi/python/symbollightapi/model/Transaction.py

@@ -1,7 +1,15 @@
+from collections import namedtuple
+
 from symbolchain.nc import TransactionType
 
 from ..model.Exceptions import UnknownTransactionType
 
+Message = namedtuple('Message', ['payload', 'is_plain'])
+Mosaic = namedtuple('Mosaic', ['namespace_name', 'quantity'])
+Modification = namedtuple('Modification', ['modification_type', 'cosignatory_account'])
+MosaicLevy = namedtuple('MosaicLevy', ['fee', 'recipient', 'type', 'namespace_name'])
+MosaicProperties = namedtuple('MosaicProperties', ['divisibility', 'initial_supply', 'supply_mutable', 'transferable'])
+
 
 class Transaction:
 	def __init__(
@@ -383,6 +391,152 @@ def __eq__(self, other):
 		])
 
 
+class TransactionHandler:
+	"""Transaction handle mapper."""
+	def __init__(self):
+		self.map = {
+			TransactionType.TRANSFER.value: self._map_transfer_args,
+			TransactionType.ACCOUNT_KEY_LINK.value: self._map_account_key_link_args,
+			TransactionType.MULTISIG_ACCOUNT_MODIFICATION.value: self._map_multisig_account_modification_args,
+			TransactionType.MULTISIG_TRANSACTION.value: self._map_multisig_transaction_args,
+			TransactionType.NAMESPACE_REGISTRATION.value: self._map_namespace_registration_args,
+			TransactionType.MOSAIC_DEFINITION.value: self._map_mosaic_definition_args,
+			TransactionType.MOSAIC_SUPPLY_CHANGE.value: self._map_mosaic_supply_change_args,
+		}
+
+	@staticmethod
+	def _map_transfer_args(tx_dict):
+		message = tx_dict['message']
+
+		if message:
+			message = Message(
+				message['payload'],
+				message['type']
+			)
+
+		mosaics = None
+		if 'mosaics' in tx_dict:
+			mosaics = [
+				Mosaic(
+					f'{mosaic["mosaicId"]["namespaceId"]}.{mosaic["mosaicId"]["name"]}',
+					mosaic['quantity']
+				)
+				for mosaic in tx_dict['mosaics']
+			]
+
+		return {
+			'amount': tx_dict['amount'],
+			'recipient': tx_dict['recipient'],
+			'message': message,
+			'mosaics': mosaics,
+		}
+
+	@staticmethod
+	def _map_account_key_link_args(tx_dict):
+		return {
+			'mode': tx_dict['mode'],
+			'remote_account': tx_dict['remoteAccount'],
+		}
+
+	@staticmethod
+	def _map_multisig_account_modification_args(tx_dict):
+		return {
+			'min_cosignatories': tx_dict['minCosignatories']['relativeChange'],
+			'modifications': [
+				Modification(
+					modification['modificationType'],
+					modification['cosignatoryAccount'])
+				for modification in tx_dict['modifications']
+			]
+		}
+
+	def _map_multisig_transaction_args(self, tx_dict, inner_hash):
+
+		other_transaction = tx_dict['otherTrans']
+
+		specific_args = self.map[other_transaction['type']](other_transaction)
+
+		common_args = {
+			'transaction_hash': None,
+			'height': None,
+			'sender': other_transaction['signer'],
+			'fee': other_transaction['fee'],
+			'timestamp': other_transaction['timeStamp'],
+			'deadline': other_transaction['deadline'],
+			'signature': None,
+		}
+
+		return {
+			'signatures': [
+				CosignSignatureTransaction(
+					signature['timeStamp'],
+					signature['otherHash']['data'],
+					signature['otherAccount'],
+					signature['signer'],
+					signature['fee'],
+					signature['deadline'],
+					signature['signature']
+				)
+				for signature in tx_dict['signatures']
+			],
+			'other_transaction': TransactionFactory.create_transaction(other_transaction['type'], common_args, specific_args),
+			'inner_hash': inner_hash,
+		}
+
+	@staticmethod
+	def _map_namespace_registration_args(tx_dict):
+		return {
+			'rental_fee_sink': tx_dict['rentalFeeSink'],
+			'rental_fee': tx_dict['rentalFee'],
+			'parent': tx_dict['parent'],
+			'namespace': tx_dict['newPart'],
+		}
+
+	@staticmethod
+	def _map_mosaic_definition_args(tx_dict):
+		mosaic_definition = tx_dict['mosaicDefinition']
+		mosaic_id = mosaic_definition['id']
+		mosaic_levy = mosaic_definition['levy']
+		mosaic_properties_dict = {
+			item['name']: item['value']
+			for item in mosaic_definition['properties']
+		}
+
+		mosaic_properties = MosaicProperties(
+			int(mosaic_properties_dict['divisibility']),
+			int(mosaic_properties_dict['initialSupply']),
+			mosaic_properties_dict['supplyMutable'] != 'false',
+			mosaic_properties_dict['transferable'] != 'false'
+		)
+
+		if mosaic_levy:
+			mosaic_levy = MosaicLevy(
+				mosaic_levy['fee'],
+				mosaic_levy['recipient'],
+				mosaic_levy['type'],
+				f'{mosaic_levy["mosaicId"]["namespaceId"]}.{mosaic_levy["mosaicId"]["name"] }'
+			)
+
+		return {
+			'creation_fee': tx_dict['creationFee'],
+			'creation_fee_sink': tx_dict['creationFeeSink'],
+			'creator': mosaic_definition['creator'],
+			'description': mosaic_definition['description'],
+			'namespace_name': f'{mosaic_id["namespaceId"]}.{mosaic_id["name"] }',
+			'properties': mosaic_properties,
+			'levy': mosaic_levy,
+		}
+
+	@staticmethod
+	def _map_mosaic_supply_change_args(tx_dict):
+		mosaic_id = tx_dict['mosaicId']
+		return {
+			'supply_type': tx_dict['supplyType'],
+			'delta': tx_dict['delta'],
+			'namespace_name': f'{mosaic_id["namespaceId"]}.{mosaic_id["name"] }',
+		}
+
+
 class TransactionFactory:
 	"""Create transaction models."""
 

lightapi/python/tests/connector/test_NemConnector.py

@@ -788,7 +788,7 @@ async def test_can_query_account_info_without_public_key(server):  # pylint: dis
 )
 
 
-async def test_can_query_blocks_after(server):  # pylint: disable=redefined-outer-name
+async def can_query_blocks_after(server):  # pylint: disable=redefined-outer-name
 	# Arrange:
 	connector = NemConnector(server.make_url(''))