[explorer/rest] feat: Added account endpoint

explorer/rest/rest/__init__.py

@@ -4,6 +4,7 @@
 
 from flask import Flask, abort, jsonify, request
 from flask_cors import CORS
+from symbolchain.CryptoTypes import PublicKey
 from zenlog import log
 
 from rest.facade.NemRestFacade import NemRestFacade
@@ -79,6 +80,34 @@ def api_get_nem_blocks():
 
 		return jsonify(nem_api_facade.get_blocks(limit=limit, offset=offset, min_height=min_height, sort=sort))
 
+	@app.route('/api/nem/account')
+	def api_get_nem_account():
+		address = request.args.get('address', '').strip() or None
+		public_key = request.args.get('publicKey', '').strip() or None
+
+		# Validate that exactly one of address or public_key is provided
+		if bool(address) == bool(public_key):
+			abort(400)
+
+		if address and not nem_api_facade.nem_db.network.is_valid_address_string(address):
+			abort(400)
+
+		if public_key:
+			try:
+				PublicKey(public_key)
+			except ValueError:
+				abort(400)
+
+		if address:
+			result = nem_api_facade.get_account_by_address(address)
+		else:
+			result = nem_api_facade.get_account_by_public_key(public_key)
+
+		if not result:
+			abort(404)
+
+		return jsonify(result)
+
 
 def setup_error_handlers(app):
 	@app.errorhandler(404)

explorer/rest/rest/db/NemDatabase.py

@@ -1,9 +1,9 @@
 from binascii import hexlify
 
 from symbolchain.CryptoTypes import PublicKey
-from symbolchain.nem.Network import Network
-from symbolchain.Network import NetworkLocator
+from symbolchain.nem.Network import Address
 
+from rest.model.Account import AccountView
 from rest.model.Block import BlockView
 
 from .DatabaseConnection import DatabaseConnectionPool
@@ -20,9 +20,9 @@ def _format_xem_relative(amount):
 class NemDatabase(DatabaseConnectionPool):
 	"""Database containing Nem blockchain data."""
 
-	def __init__(self, db_config, network_name):
+	def __init__(self, db_config, network):
 		super().__init__(db_config)
-		self.network = NetworkLocator.find_by_name(Network.NETWORKS, network_name)
+		self.network = network
 
 	def _create_block_view(self, result):
 		harvest_public_key = PublicKey(_format_bytes(result[7]))
@@ -38,6 +38,70 @@ def _create_block_view(self, result):
 			size=result[9]
 		)
 
+	def _create_account_view(self, result):  # pylint: disable=no-self-use,too-many-locals
+		(
+			address,
+			public_key,
+			remote_address,
+			importance,
+			balance,
+			vested_balance,
+			mosaics,
+			harvested_fees,
+			harvested_blocks,
+			status,
+			remote_status,
+			last_harvested_height,
+			min_cosignatories,
+			cosignatory_of,
+			cosignatories
+		) = result
+
+		return AccountView(
+			address=str(Address(address)),
+			public_key=str(PublicKey(public_key)) if public_key else None,
+			remote_address=str(Address(remote_address)) if remote_address else None,
+			importance=importance,
+			balance=_format_xem_relative(balance),
+			vested_balance=_format_xem_relative(vested_balance),
+			mosaics=[{
+				'namespace_name': mosaic['namespace'],
+				'quantity': mosaic['quantity'],
+			} for mosaic in mosaics],
+			harvested_fees=_format_xem_relative(harvested_fees),
+			harvested_blocks=harvested_blocks,
+			status=status,
+			remote_status=remote_status,
+			last_harvested_height=last_harvested_height,
+			min_cosignatories=min_cosignatories,
+			cosignatory_of=[str(Address(address)) for address in cosignatory_of] if cosignatory_of else None,
+			cosignatories=[str(Address(address)) for address in cosignatories] if cosignatories else None
+		)
+
+	def _generate_account_query(self, where_condition):  # pylint: disable=no-self-use
+		"""Base account query."""
+
+		return f'''
+			SELECT
+				address,
+				public_key,
+				remote_address,
+				importance::float,
+				balance,
+				vested_balance,
+				mosaics,
+				harvested_fees,
+				harvested_blocks,
+				status,
+				remote_status,
+				last_harvested_height,
+				min_cosignatories,
+				cosignatory_of,
+				cosignatories
+			FROM accounts
+			WHERE {where_condition}
+		'''
+
 	def get_block(self, height):
 		"""Gets block by height in database."""
 
@@ -67,3 +131,31 @@ def get_blocks(self, limit, offset, min_height, sort):
 			results = cursor.fetchall()
 
 			return [self._create_block_view(result) for result in results]
+
+	def get_account_by_address(self, address):
+		"""Gets account by address."""
+
+		where_clause = 'address = %s'
+
+		sql = self._generate_account_query(where_clause)
+
+		with self.connection() as connection:
+			cursor = connection.cursor()
+			cursor.execute(sql, (address.bytes,))
+			result = cursor.fetchone()
+
+			return self._create_account_view(result) if result else None
+
+	def get_account_by_public_key(self, public_key):
+		"""Gets account by public key."""
+
+		where_clause = 'public_key = %s'
+
+		sql = self._generate_account_query(where_clause)
+
+		with self.connection() as connection:
+			cursor = connection.cursor()
+			cursor.execute(sql, (public_key.bytes,))
+			result = cursor.fetchone()
+
+			return self._create_account_view(result) if result else None

explorer/rest/rest/facade/NemRestFacade.py

@@ -1,3 +1,7 @@
+from symbolchain.CryptoTypes import PublicKey
+from symbolchain.nem.Network import Address, Network
+from symbolchain.Network import NetworkLocator
+
 from rest.db.NemDatabase import NemDatabase
 
 
@@ -7,7 +11,8 @@ class NemRestFacade:
 	def __init__(self, db_config, network_name):
 		"""Creates a facade object."""
 
-		self.nem_db = NemDatabase(db_config, network_name)
+		self.network = NetworkLocator.find_by_name(Network.NETWORKS, network_name)
+		self.nem_db = NemDatabase(db_config, self.network)
 
 	def get_block(self, height):
 		"""Gets block by height."""
@@ -22,3 +27,17 @@ def get_blocks(self, limit, offset, min_height, sort):
 		blocks = self.nem_db.get_blocks(limit, offset, min_height, sort)
 
 		return [block.to_dict() for block in blocks]
+
+	def get_account_by_address(self, address):
+		"""Gets account by address."""
+
+		account = self.nem_db.get_account_by_address(Address(address))
+
+		return account.to_dict() if account else None
+
+	def get_account_by_public_key(self, public_key):
+		"""Gets account by public key."""
+
+		account = self.nem_db.get_account_by_public_key(PublicKey(public_key))
+
+		return account.to_dict() if account else None

explorer/rest/rest/model/Account.py

@@ -0,0 +1,78 @@
+class AccountView:  # pylint: disable=too-many-locals,too-many-instance-attributes
+	def __init__(
+		self,
+		address,
+		public_key,
+		remote_address,
+		importance,
+		balance,
+		vested_balance,
+		mosaics,
+		harvested_fees,
+		harvested_blocks,
+		status,
+		remote_status,
+		last_harvested_height,
+		min_cosignatories,
+		cosignatory_of,
+		cosignatories
+	):
+		"""Create account view."""
+
+		# pylint: disable=too-many-arguments,too-many-positional-arguments
+
+		self.address = address
+		self.public_key = public_key
+		self.remote_address = remote_address
+		self.importance = importance
+		self.balance = balance
+		self.vested_balance = vested_balance
+		self.mosaics = mosaics
+		self.harvested_fees = harvested_fees
+		self.harvested_blocks = harvested_blocks
+		self.status = status
+		self.remote_status = remote_status
+		self.last_harvested_height = last_harvested_height
+		self.min_cosignatories = min_cosignatories
+		self.cosignatory_of = cosignatory_of
+		self.cosignatories = cosignatories
+
+	def __eq__(self, other):
+		return isinstance(other, AccountView) and all([
+			self.address == other.address,
+			self.public_key == other.public_key,
+			self.remote_address == other.remote_address,
+			self.importance == other.importance,
+			self.balance == other.balance,
+			self.vested_balance == other.vested_balance,
+			self.mosaics == other.mosaics,
+			self.harvested_fees == other.harvested_fees,
+			self.harvested_blocks == other.harvested_blocks,
+			self.status == other.status,
+			self.remote_status == other.remote_status,
+			self.last_harvested_height == other.last_harvested_height,
+			self.min_cosignatories == other.min_cosignatories,
+			self.cosignatory_of == other.cosignatory_of,
+			self.cosignatories == other.cosignatories
+		])
+
+	def to_dict(self):
+		"""Formats the account info as a dictionary."""
+
+		return {
+			'address': self.address,
+			'publicKey': self.public_key,
+			'remoteAddress': self.remote_address,
+			'importance': self.importance,
+			'balance': self.balance,
+			'vestedBalance': self.vested_balance,
+			'mosaics': self.mosaics,
+			'harvestedFees': self.harvested_fees,
+			'harvestedBlocks': self.harvested_blocks,
+			'status': self.status,
+			'remoteStatus': self.remote_status,
+			'lastHarvestedHeight': self.last_harvested_height,
+			'minCosignatories': self.min_cosignatories,
+			'cosignatoryOf': self.cosignatory_of,
+			'cosignatories': self.cosignatories
+		}

explorer/rest/tests/db/test_NemDatabase.py

@@ -2,7 +2,7 @@
 
 from rest.db.NemDatabase import NemDatabase
 
-from ..test.DatabaseTestUtils import BLOCK_VIEWS, DatabaseTestBase
+from ..test.DatabaseTestUtils import ACCOUNT_VIEWS, ACCOUNTS, BLOCK_VIEWS, DatabaseTestBase
 
 BlockQueryParams = namedtuple('BlockQueryParams', ['limit', 'offset', 'min_height', 'sort'])
 
@@ -12,14 +12,18 @@
 
 EXPECTED_BLOCK_VIEW_2 = BLOCK_VIEWS[1]
 
+EXPECTED_ACCOUNT_VIEW_1 = ACCOUNT_VIEWS[0]
+
 # endregion
 
 
 class NemDatabaseTest(DatabaseTestBase):
 
+	# region block
+
 	def _assert_can_query_block_by_height(self, height, expected_block):
 		# Arrange:
-		nem_db = NemDatabase(self.db_config, self.network_name)
+		nem_db = NemDatabase(self.db_config, self.network)
 
 		# Act:
 		block_view = nem_db.get_block(height)
@@ -29,7 +33,7 @@ def _assert_can_query_block_by_height(self, height, expected_block):
 
 	def _assert_can_query_blocks_with_filter(self, query_params, expected_blocks):
 		# Arrange:
-		nem_db = NemDatabase(self.db_config, self.network_name)
+		nem_db = NemDatabase(self.db_config, self.network)
 
 		# Act:
 		blocks_view = nem_db.get_blocks(query_params.limit, query_params.offset, query_params.min_height, query_params.sort)
@@ -66,3 +70,29 @@ def test_can_query_blocks_sorted_by_height_asc(self):
 
 	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])
+
+	# endregion
+
+	# region account
+
+	def test_can_query_account_by_address(self):
+		# Arrange:
+		nem_db = NemDatabase(self.db_config, self.network)
+
+		# Act:
+		account_view = nem_db.get_account_by_address(address=ACCOUNTS[0].address)
+
+		# Assert:
+		self.assertEqual(EXPECTED_ACCOUNT_VIEW_1, account_view)
+
+	def test_can_query_account_by_public_key(self):
+		# Arrange:
+		nem_db = NemDatabase(self.db_config, self.network)
+
+		# Act:
+		account_view = nem_db.get_account_by_public_key(public_key=ACCOUNTS[0].public_key)
+
+		# Assert:
+		self.assertEqual(EXPECTED_ACCOUNT_VIEW_1, account_view)
+
+	# endregion