[explorer/rest]: added mosaic endpoint

explorer/rest/rest/__init__.py

@@ -102,6 +102,13 @@ def api_get_nem_namespaces():
 
 		return jsonify(nem_api_facade.get_namespaces(limit=limit, offset=offset, sort=sort))
 
+	@app.route('/api/nem/mosaic/<name>')
+	def api_get_nem_mosaic_by_name(name):
+		result = nem_api_facade.get_mosaic(name)
+		if not result:
+			abort(404)
+		return jsonify(result)
+
 
 def setup_error_handlers(app):
 	@app.errorhandler(404)

explorer/rest/rest/db/NemDatabase.py

@@ -1,15 +1,20 @@
-from binascii import hexlify
+from binascii import hexlify, unhexlify
 
 from symbolchain.CryptoTypes import PublicKey
-from symbolchain.nem.Network import Network
+from symbolchain.nem.Network import Address, Network
 from symbolchain.Network import NetworkLocator
 
 from rest.model.Block import BlockView
+from rest.model.Mosaic import MosaicView
 from rest.model.Namespace import NamespaceView
 
 from .DatabaseConnection import DatabaseConnectionPool
 
 
+def _format_address_bytes(buffer):
+	return unhexlify(_format_bytes(buffer))
+
+
 def _format_bytes(buffer):
 	return hexlify(buffer).decode('utf8').upper()
 
@@ -18,6 +23,10 @@ def _format_xem_relative(amount):
 	return amount / (10 ** 6)
 
 
+def _format_relative(amount, divisibility):
+	return amount / (10 ** divisibility)
+
+
 class NemDatabase(DatabaseConnectionPool):
 	"""Database containing Nem blockchain data."""
 
@@ -112,6 +121,41 @@ def _create_namespace_view(self, result):
 			mosaics=mosaics
 		)
 
+	def _create_mosaic_view(self, result):
+		levy_types = {
+			1: 'absolute fee',
+			2: 'percentile'
+		}
+
+		creator_public_key = PublicKey(_format_bytes(result[2]))
+		levy_type = levy_types.get(result[10], None)
+		levy_fee = _format_relative(result[13], result[12]) if levy_type else None
+
+		namespace_mosaic_name = result[0].split('.')
+		namespace_name = '.'.join(namespace_mosaic_name[:-1])
+		mosaic_name = namespace_mosaic_name[-1]
+
+		return MosaicView(
+			mosaic_name=mosaic_name,
+			namespace_name=namespace_name,
+			description=result[1],
+			creator=self.network.public_key_to_address(creator_public_key),
+			registered_height=result[3],
+			registered_timestamp=str(result[4]),
+			initial_supply=result[5],
+			total_supply=result[6],
+			divisibility=result[7],
+			supply_mutable=result[8],
+			transferable=result[9],
+			levy_type=levy_type,
+			levy_namespace=result[11],
+			levy_fee=levy_fee,
+			levy_recipient=Address(_format_address_bytes(result[14])) if result[14] else None,
+			root_namespace_registered_height=result[15],
+			root_namespace_registered_timestamp=str(result[16]),
+			root_namespace_expiration_height=result[17],
+		)
+
 	def get_block(self, height):
 		"""Gets block by height in database."""
 
@@ -173,3 +217,47 @@ def get_namespaces(self, limit, offset, sort):
 			results = cursor.fetchall()
 
 			return [self._create_namespace_view(result) for result in results]
+
+	def get_mosaic(self, namespace_name):
+		"""Gets mosaic by namespace name in database."""
+
+		with self.connection() as connection:
+			cursor = connection.cursor()
+			cursor.execute('''
+				SELECT
+					m1.namespace_name,
+					m1.description,
+					m1.creator,
+					m1.registered_height as mosaic_registered_height,
+					b2.timestamp as mosaic_registered_timestamp,
+					m1.initial_supply,
+					m1.total_supply,
+					m1.divisibility,
+					m1.supply_mutable,
+					m1.transferable,
+					m1.levy_type,
+					m1.levy_namespace_name,
+					CASE
+						WHEN m1.levy_namespace_name = 'nem.xem' THEN 6
+						WHEN m1.levy_namespace_name IS NULL THEN NULL
+						ELSE m2.divisibility
+					END AS levy_namespace_divisibility,
+					m1.levy_fee,
+					m1.levy_recipient,
+					n.registered_height AS root_namespace_registered_height,
+					b1.timestamp AS root_namespace_registered_timestamp,
+					n.expiration_height
+				FROM mosaics m1
+				LEFT JOIN mosaics m2
+					ON m1.levy_namespace_name = m2.namespace_name AND m1.levy_namespace_name IS NOT NULL
+				LEFT JOIN namespaces n
+					ON m1.root_namespace = n.root_namespace
+				LEFT JOIN blocks b1
+					ON b1.height = n.registered_height
+				LEFT JOIN blocks b2
+					ON b2.height = m1.registered_height
+				WHERE m1.namespace_name = %s
+			''', (namespace_name,))
+			result = cursor.fetchone()
+
+			return self._create_mosaic_view(result) if result else None

explorer/rest/rest/facade/NemRestFacade.py

@@ -36,3 +36,10 @@ def get_namespaces(self, limit, offset, sort):
 		namespaces = self.nem_db.get_namespaces(limit, offset, sort)
 
 		return [namespace.to_dict() for namespace in namespaces]
+
+	def get_mosaic(self, name):
+		"""Gets mosaic by namespace name."""
+
+		mosaic = self.nem_db.get_mosaic(name)
+
+		return mosaic.to_dict() if mosaic else None

explorer/rest/rest/model/Mosaic.py

@@ -0,0 +1,90 @@
+class MosaicView:  # pylint: disable=too-many-instance-attributes, too-many-locals
+	def __init__(
+		self,
+		mosaic_name,
+		namespace_name,
+		description,
+		creator,
+		registered_height,
+		registered_timestamp,
+		initial_supply,
+		total_supply,
+		divisibility,
+		supply_mutable,
+		transferable,
+		levy_type,
+		levy_namespace,
+		levy_fee,
+		levy_recipient,
+		root_namespace_registered_height,
+		root_namespace_registered_timestamp,
+		root_namespace_expiration_height,
+	):
+		"""Create mosaic view."""
+
+		# pylint: disable=too-many-arguments
+
+		self.mosaic_name = mosaic_name
+		self.namespace_name = namespace_name
+		self.description = description
+		self.creator = creator
+		self.registered_height = registered_height
+		self.registered_timestamp = registered_timestamp
+		self.initial_supply = initial_supply
+		self.total_supply = total_supply
+		self.divisibility = divisibility
+		self.supply_mutable = supply_mutable
+		self.transferable = transferable
+		self.levy_type = levy_type
+		self.levy_namespace = levy_namespace
+		self.levy_fee = levy_fee
+		self.levy_recipient = levy_recipient
+		self.root_namespace_registered_height = root_namespace_registered_height
+		self.root_namespace_registered_timestamp = root_namespace_registered_timestamp
+		self.root_namespace_expiration_height = root_namespace_expiration_height
+
+	def __eq__(self, other):
+		return isinstance(other, MosaicView) and all([
+			self.mosaic_name == other.mosaic_name,
+			self.namespace_name == other.namespace_name,
+			self.description == other.description,
+			self.creator == other.creator,
+			self.registered_height == other.registered_height,
+			self.registered_timestamp == other.registered_timestamp,
+			self.initial_supply == other.initial_supply,
+			self.total_supply == other.total_supply,
+			self.divisibility == other.divisibility,
+			self.supply_mutable == other.supply_mutable,
+			self.transferable == other.transferable,
+			self.levy_type == other.levy_type,
+			self.levy_namespace == other.levy_namespace,
+			self.levy_fee == other.levy_fee,
+			self.levy_recipient == other.levy_recipient,
+			self.root_namespace_registered_height == other.root_namespace_registered_height,
+			self.root_namespace_registered_timestamp == other.root_namespace_registered_timestamp,
+			self.root_namespace_expiration_height == other.root_namespace_expiration_height
+		])
+
+	def to_dict(self):
+		"""Formats the mosaic info as a dictionary."""
+
+		return {
+			'mosaicName': self.mosaic_name,
+			'namespaceName': self.namespace_name,
+			'description': self.description,
+			'creator': str(self.creator),
+			'registeredHeight': self.registered_height,
+			'registeredTimestamp': str(self.registered_timestamp),
+			'initialSupply': self.initial_supply,
+			'totalSupply': self.total_supply,
+			'divisibility': self.divisibility,
+			'supplyMutable': self.supply_mutable,
+			'transferable': self.transferable,
+			'levyType': self.levy_type,
+			'levyNamespace': self.levy_namespace,
+			'levyFee': self.levy_fee,
+			'levyRecipient': str(self.levy_recipient) if self.levy_recipient else None,
+			'rootNamespaceRegisteredHeight': self.root_namespace_registered_height,
+			'rootNamespaceRegisteredTimestamp': str(self.root_namespace_registered_timestamp),
+			'rootNamespaceExpirationHeight': self.root_namespace_expiration_height
+		}

explorer/rest/tests/db/test_NemDatabase.py

@@ -2,23 +2,11 @@
 
 from rest.db.NemDatabase import NemDatabase
 
-from ..test.DatabaseTestUtils import BLOCK_VIEWS, NAMESPACE_VIEWS, DatabaseTestBase
+from ..test.DatabaseTestUtils import BLOCK_VIEWS, MOSAIC_VIEWS, NAMESPACE_VIEWS, DatabaseTestBase
 
 BlockQueryParams = namedtuple('BlockQueryParams', ['limit', 'offset', 'min_height', 'sort'])
 PaginationQueryParams = namedtuple('PaginationQueryParams', ['limit', 'offset', 'sort'])
 
-# region test data
-
-EXPECTED_BLOCK_VIEW_1 = BLOCK_VIEWS[0]
-
-EXPECTED_BLOCK_VIEW_2 = BLOCK_VIEWS[1]
-
-EXPECTED_NAMESPACE_VIEW_1 = NAMESPACE_VIEWS[0]
-
-EXPECTED_NAMESPACE_VIEW_2 = NAMESPACE_VIEWS[1]
-
-# endregion
-
 
 class NemDatabaseTest(DatabaseTestBase):
 
@@ -45,44 +33,42 @@ def _assert_can_query_blocks_with_filter(self, query_params, expected_blocks):
 		self.assertEqual(expected_blocks, blocks_view)
 
 	def test_can_query_block_by_height_1(self):
-		self._assert_can_query_block_by_height(1, EXPECTED_BLOCK_VIEW_1)
+		self._assert_can_query_block_by_height(1, BLOCK_VIEWS[0])
 
 	def test_cannot_query_nonexistent_block(self):
 		self._assert_can_query_block_by_height(3, None)
 
 	def test_can_query_blocks_filtered_limit(self):
-		self._assert_can_query_blocks_with_filter(BlockQueryParams(1, 0, 1, 'desc'), [EXPECTED_BLOCK_VIEW_2])
+		self._assert_can_query_blocks_with_filter(BlockQueryParams(1, 0, 1, 'desc'), [BLOCK_VIEWS[1]])
 
 	def test_can_query_blocks_filtered_offset_0(self):
-		self._assert_can_query_blocks_with_filter(BlockQueryParams(1, 0, 0, 'desc'), [EXPECTED_BLOCK_VIEW_2])
+		self._assert_can_query_blocks_with_filter(BlockQueryParams(1, 0, 0, 'desc'), [BLOCK_VIEWS[1]])
 
 	def test_can_query_blocks_filtered_offset_1(self):
-		self._assert_can_query_blocks_with_filter(BlockQueryParams(1, 1, 0, 'desc'), [EXPECTED_BLOCK_VIEW_1])
+		self._assert_can_query_blocks_with_filter(BlockQueryParams(1, 1, 0, 'desc'), [BLOCK_VIEWS[0]])
 
 	def test_can_query_blocks_filtered_min_height_1(self):
-		self._assert_can_query_blocks_with_filter(BlockQueryParams(10, 0, 1, 'desc'), [EXPECTED_BLOCK_VIEW_2, EXPECTED_BLOCK_VIEW_1])
+		self._assert_can_query_blocks_with_filter(BlockQueryParams(10, 0, 1, 'desc'), [BLOCK_VIEWS[1], BLOCK_VIEWS[0]])
 
 	def test_can_query_blocks_filtered_min_height_2(self):
-		self._assert_can_query_blocks_with_filter(BlockQueryParams(10, 0, 2, 'desc'), [EXPECTED_BLOCK_VIEW_2])
+		self._assert_can_query_blocks_with_filter(BlockQueryParams(10, 0, 2, 'desc'), [BLOCK_VIEWS[1]])
 
 	def test_can_query_blocks_filtered_min_height_3(self):
 		self._assert_can_query_blocks_with_filter(BlockQueryParams(10, 0, 3, 'desc'), [])
 
 	def test_can_query_blocks_sorted_by_height_asc(self):
-		self._assert_can_query_blocks_with_filter(BlockQueryParams(10, 0, 0, 'asc'), [EXPECTED_BLOCK_VIEW_1, EXPECTED_BLOCK_VIEW_2])
+		self._assert_can_query_blocks_with_filter(BlockQueryParams(10, 0, 0, 'asc'), [BLOCK_VIEWS[0], BLOCK_VIEWS[1]])
 
 	def test_can_query_blocks_sorted_by_height_desc(self):
-		self._assert_can_query_blocks_with_filter(BlockQueryParams(10, 0, 0, 'desc'), [EXPECTED_BLOCK_VIEW_2, EXPECTED_BLOCK_VIEW_1])
+		self._assert_can_query_blocks_with_filter(BlockQueryParams(10, 0, 0, 'desc'), [BLOCK_VIEWS[1], BLOCK_VIEWS[0]])
 
 	# endregion
 
 	# region namespace tests
 
 	def _assert_can_query_namespace_by_name(self, name, expected_namespace):
-		# Arrange:
 		nem_db = NemDatabase(self.db_config, self.network_name)
 
-		# Act:
 		namespace_view = nem_db.get_namespace(name)
 
 		# Assert:
@@ -99,27 +85,47 @@ def _assert_can_query_namespaces_with_filter(self, query_params, expected_namesp
 		self.assertEqual(expected_namespaces, namespaces_view)
 
 	def test_can_query_namespace_by_name(self):
-		self._assert_can_query_namespace_by_name('oxford', EXPECTED_NAMESPACE_VIEW_1)
+		self._assert_can_query_namespace_by_name('oxford', NAMESPACE_VIEWS[0])
 
 	def test_cannot_query_nonexistent_namespace(self):
 		self._assert_can_query_namespace_by_name('non_exist', None)
 
 	def test_can_query_namespaces_filtered_limit(self):
-		self._assert_can_query_namespaces_with_filter(PaginationQueryParams(1, 0, 'desc'), [EXPECTED_NAMESPACE_VIEW_2])
+		self._assert_can_query_namespaces_with_filter(PaginationQueryParams(1, 0, 'desc'), [NAMESPACE_VIEWS[1]])
 
 	def test_can_query_namespaces_filtered_offset_0(self):
-		self._assert_can_query_namespaces_with_filter(PaginationQueryParams(1, 0, 'desc'), [EXPECTED_NAMESPACE_VIEW_2])
+		self._assert_can_query_namespaces_with_filter(PaginationQueryParams(1, 0, 'desc'), [NAMESPACE_VIEWS[1]])
 
 	def test_can_query_namespaces_filtered_offset_1(self):
-		self._assert_can_query_namespaces_with_filter(PaginationQueryParams(1, 1, 'desc'), [EXPECTED_NAMESPACE_VIEW_1])
+		self._assert_can_query_namespaces_with_filter(PaginationQueryParams(1, 1, 'desc'), [NAMESPACE_VIEWS[0]])
 
 	def test_can_query_namespaces_sorted_by_id_asc(self):
-		self._assert_can_query_namespaces_with_filter(PaginationQueryParams(10, 0, 'asc'), [EXPECTED_NAMESPACE_VIEW_1, EXPECTED_NAMESPACE_VIEW_2])
+		self._assert_can_query_namespaces_with_filter(PaginationQueryParams(10, 0, 'asc'), [NAMESPACE_VIEWS[0], NAMESPACE_VIEWS[1]])
 
 	def test_can_query_namespaces_sorted_by_id_desc(self):
 		self._assert_can_query_namespaces_with_filter(
 			PaginationQueryParams(10, 0, 'desc'),
-			[EXPECTED_NAMESPACE_VIEW_2, EXPECTED_NAMESPACE_VIEW_1]
+			[NAMESPACE_VIEWS[1], NAMESPACE_VIEWS[0]]
 		)
 
 	# endregion
+
+	# region mosaic tests
+
+	def _assert_can_query_mosaic_by_name(self, namespace_name, expected_mosaic):
+		# Arrange:
+		nem_db = NemDatabase(self.db_config, self.network_name)
+
+		# Act:
+		mosaic_view = nem_db.get_mosaic(namespace_name)
+
+		# Assert:
+		self.assertEqual(expected_mosaic, mosaic_view)
+
+	def test_can_query_mosaic_by_name(self):
+		self._assert_can_query_mosaic_by_name('dragon.dragonfly', MOSAIC_VIEWS[0])
+
+	def test_cannot_query_nonexistent_mosaic(self):
+		self._assert_can_query_mosaic_by_name('non-exist-mosaic', None)
+
+	# endregion