feat: metadata api

.github/workflows/deploy.yml

@@ -80,7 +80,7 @@ jobs:
           REDIS_HOST: ${{ secrets.REDIS_HOST }}
           REDIS_PORT: 6379
           REDIS_DB: 0
-          TOKEN_METADATA_BUCKET: token-metadata-testnet
+          METADATA_BUCKET: metadata-testnet
           CORS_ALLOWED_ORIGIN: https://explorer.testnet.hathor.network
       - name: Deploy Lambdas Mainnet
         if: ${{ needs.init.outputs.environment == 'mainnet' }}
@@ -94,7 +94,7 @@ jobs:
           REDIS_HOST: ${{ secrets.REDIS_HOST }}
           REDIS_PORT: 6379
           REDIS_DB: 0
-          TOKEN_METADATA_BUCKET: token-metadata-mainnet
+          METADATA_BUCKET: metadata-mainnet
           CORS_ALLOWED_ORIGIN: https://explorer.hathor.network
       - name: Deploy Daemons Testnet
         if: ${{ needs.init.outputs.environment == 'testnet' }}

common/configuration.py

@@ -43,6 +43,6 @@ def default(cls) -> 'Environment':
 REDIS_PORT = config('REDIS_PORT', default=None)
 REDIS_DB = config('REDIS_DB', default='0', cast=int)
 
-TOKEN_METADATA_BUCKET = config('TOKEN_METADATA_BUCKET', default=None)
+METADATA_BUCKET = config('METADATA_BUCKET', default=None)
 
 CORS_ALLOWED_ORIGIN = config('CORS_ALLOWED_ORIGIN', default=None)

domain/metadata/metadata.py

@@ -0,0 +1,32 @@
+from dataclasses import asdict, dataclass
+from enum import Enum
+
+
+class MetadataType(str, Enum):
+    TOKEN = 'TOKEN'
+    TRANSACTION = 'TRANSACTION'
+
+
+@dataclass
+class Metadata:
+    id: str
+
+    @property
+    def type(self) -> MetadataType:
+        raise NotImplementedError
+
+    @property
+    def data(self):
+        raise NotImplementedError
+
+    @classmethod
+    def from_dict(cls, dikt: dict) -> 'Metadata':
+        raise NotImplementedError
+
+    def to_dict(self) -> dict:
+        """ Convert a Metadata instance into dict
+
+        :return: Dict representations of Metadata
+        :rtype: dict
+        """
+        return asdict(self)

domain/metadata/token_metadata.py

@@ -0,0 +1,92 @@
+from dataclasses import dataclass
+from enum import Enum
+from typing import Optional
+
+from domain.metadata.metadata import Metadata, MetadataType
+
+
+class TokenNFTMediaType(str, Enum):
+    VIDEO = 'VIDEO'
+    IMAGE = 'IMAGE'
+    AUDIO = 'AUDIO'
+
+
+@dataclass
+class TokenNFTMedia:
+    """Data of token NFT
+
+    :param type: NFT Type. image or video.
+    :type type: :py:class:`domain.metadata.token_metdata.TokenNFTMediaType`
+
+    :param file: Media file of the NFT
+    :type file: str
+
+    :param loop: If media will play in loop or not. Works for audio and video as well
+    :type loop: Optional[bool]
+
+    :param autoplay: If media will play automatically or not. Works for audio and video as well
+    :type autoplay: Optional[bool]
+    """
+    type: TokenNFTMediaType
+    file: str
+    loop: Optional[bool] = False
+    autoplay: Optional[bool] = False
+
+
+@dataclass
+class MetaToken:
+    """Metadata of a token
+
+    :param id: Token unique id
+    :type id: str
+
+    :param verified: If token is verified or not. None and False are equivalent
+    :type verified: Optional[bool]
+
+    :param banned: If token is banned or not. None and False are equivalent
+    :type banned: Optional[bool]
+
+    :param reason: Bannishment reason
+    :type reason: Optional[str]
+
+    :param nft: If token is a nft or not. None and False are equivalent
+    :type nft: Optional[bool]
+
+    :param nft_media: NFT media data, if any.
+    :type nft_media: Optional[:py:class:`domain.metadata.token_metdata.TokenNFTMedia`]
+    """
+    id: str
+    verified: Optional[bool] = False
+    banned: Optional[bool] = False
+    reason: Optional[str] = ''
+    nft: Optional[bool] = False
+    nft_media: Optional[TokenNFTMedia] = None
+
+
+@dataclass
+class TokenMetadata(Metadata):
+    data: Optional[MetaToken] = None
+    type: MetadataType = MetadataType.TOKEN
+
+    @classmethod
+    def from_dict(cls, dikt: dict) -> 'TokenMetadata':
+        """ Creates a new TokenMetadata instance from a given dict (inverse operation of `to_dict`)
+
+        :param dikt: Dict with TokenMetadata structure and data
+        :type dikt: dict
+
+        :return: The new instance
+        :rtype: :py:class:`domain.metadata.token_metdata.TokenMetadata`
+        """
+        data = dikt.get('data', {})
+        if data.get('nft_media'):
+            data['nft_media'] = TokenNFTMedia(
+                TokenNFTMediaType(data['nft_media']['type'].upper()),
+                data['nft_media']['file'],
+                data['nft_media'].get('loop', None)
+            )
+
+        return cls(
+            id=dikt['id'],
+            data=MetaToken(**data)
+        )

domain/metadata/transaction_metadata.py

@@ -0,0 +1,33 @@
+from dataclasses import dataclass
+from typing import Optional
+
+from domain.metadata.metadata import Metadata, MetadataType
+
+
+@dataclass
+class MetaTransaction:
+    id: str
+    context: Optional[str]
+    genesis: Optional[bool]
+
+
+@dataclass
+class TransactionMetadata(Metadata):
+    data: Optional[MetaTransaction] = None
+    type: MetadataType = MetadataType.TRANSACTION
+
+    @classmethod
+    def from_dict(cls, dikt: dict) -> 'TransactionMetadata':
+        """ Creates a new TransactionMetadata instance from a given dict (inverse operation of `to_dict`)
+
+        :param dikt: Dict with TransactionMetadata structure and data
+        :type dikt: dict
+
+        :return: The new instance
+        :rtype: :py:class:`domain.metadata.token_metdata.TransactionMetadata`
+        """
+
+        return cls(
+            id=dikt['id'],
+            data=MetaTransaction(**dikt['data'])
+        )

domain/tx/token.py

@@ -1,79 +1,8 @@
 from dataclasses import asdict, dataclass
-from enum import Enum
-from typing import Optional
 
 from dacite import from_dict
 
 
-class TokenNFTType(str, Enum):
-    VIDEO = 'VIDEO'
-    IMAGE = 'IMAGE'
-    AUDIO = 'AUDIO'
-
-
-@dataclass
-class TokenNFT:
-    """Data of token NFT
-
-    :param type: NFT Type. image or video.
-    :type type: :py:class:`domain.tx.token.TokenNFTType`
-
-    :param file: Media file of the NFT
-    :type file: str
-    """
-    type: TokenNFTType
-    file: str
-
-
-@dataclass
-class TokenMetadata:
-    """Metadata of token
-
-    :param id: Token unique id
-    :type id: str
-
-    :param verified: If token is verified or not
-    :type verified: bool
-
-    :param banned: If token is banned or not
-    :type banned: bool
-
-    :param reason: Bannishment reason
-    :type reason: str
-
-    :param nft: NFT data, if any.
-    :type nft: :py:class:`domain.tx.token.TokenNFT`
-    """
-    id: str
-    verified: bool
-    banned: bool
-    reason: Optional[str]
-    nft: Optional[TokenNFT]
-
-    def to_dict(self) -> dict:
-        """ Convert a TokenMetadata instance into dict
-
-        :return: Dict representations of TokenMetadata
-        :rtype: dict
-        """
-        return asdict(self)
-
-    @classmethod
-    def from_dict(cls, dikt: dict) -> 'TokenMetadata':
-        """ Creates a new TokenMetadata instance from a given dict (inverse operation of `to_dict`)
-
-        :param dikt: Dict with TokenMetadata structure and data
-        :type dikt: dict
-
-        :return: The new instance
-        :rtype: :py:class:`domain.tx.token.TokenMetadata`
-        """
-        if dikt.get('nft'):
-            dikt['nft'] = TokenNFT(TokenNFTType(dikt['nft']['type'].upper()), dikt['nft']['file'])
-
-        return TokenMetadata(**dikt)
-
-
 @dataclass
 class Token:
     """Hathor Token

gateways/metadata_gateway.py

@@ -0,0 +1,80 @@
+import json
+from typing import Optional
+
+from common.configuration import METADATA_BUCKET
+from domain.metadata.metadata import MetadataType
+from domain.metadata.token_metadata import TokenMetadata
+from domain.metadata.transaction_metadata import TransactionMetadata
+from gateways.clients.s3_client import S3Client
+
+TYPE_TO_FOLDER = {
+    MetadataType.TRANSACTION: 'transaction',
+    MetadataType.TOKEN: 'token'
+}
+
+
+class MetadataGateway:
+
+    def __init__(self, s3_client: Optional[S3Client] = None) -> None:
+        self.s3_client = s3_client or S3Client()
+
+    def get_transaction_metadata(self, id: str) -> Optional[TransactionMetadata]:
+        """Retrieve transaction metadata from a json file stored in s3
+
+        :param id: transaction id
+        :type id: str
+        :raises Exception: The name of the bucket used to store the jsons must be on config
+        :return: TransactionMetadata object or None if not found
+        :rtype: TokenMetadata | None
+        """
+        folder = TYPE_TO_FOLDER[MetadataType.TRANSACTION]
+        transaction_metadata = self._get_metadata(f"{folder}/{id}.json")
+
+        if transaction_metadata is None:
+            return None
+
+        return TransactionMetadata.from_dict(transaction_metadata)
+
+    def get_token_metadata(self, id: str) -> Optional[TokenMetadata]:
+        """Retrieve token metadata from a json file stored in s3
+
+        :param id: token id
+        :type id: str
+
+        :return: TokenMetadata object or None if not found
+        :rtype: TokenMetadata | None
+        """
+        folder = TYPE_TO_FOLDER[MetadataType.TOKEN]
+        token_metadata = self._get_metadata(f"{folder}/{id}.json")
+
+        if token_metadata is None:
+            return None
+
+        return TokenMetadata.from_dict(token_metadata)
+
+    def _get_metadata(self, s3_object_name: str) -> Optional[dict]:
+        """Retrive metadata from file in s3
+
+        :param s3_object_name: name of object
+        :type s3_object_name: str
+
+        :raises Exception: The name of the bucket used to store the jsons must be on config
+
+        :return: file content as dict
+        :rtype: Optional[dict]
+        """
+        metadata_bucket = self._metadata_bucket()
+
+        raw_metadata = self.s3_client.load_file(metadata_bucket, s3_object_name)
+        if raw_metadata is None:
+            return None
+
+        return json.loads(raw_metadata)
+
+    def _metadata_bucket(self) -> str:
+        metadata_bucket = METADATA_BUCKET
+
+        if metadata_bucket is None:
+            raise Exception('No bucket name in config')
+
+        return metadata_bucket

gateways/token_gateway.py

@@ -1,48 +1,19 @@
-import json
 from typing import Union
 
-from common.configuration import TOKEN_METADATA_BUCKET
-from domain.tx.token import Token, TokenMetadata
+from domain.tx.token import Token
 from gateways.clients.hathor_core_client import TOKEN_ENDPOINT, HathorCoreClient
 from gateways.clients.responses.hathor_core.hathor_core_token_response import HathorCoreTokenResponse
-from gateways.clients.s3_client import S3Client
 
 
 class TokenGateway:
     """Gateway for Token
 
-    :param s3_client: Client for s3 manipulation, default to domain S3Client
-    :type s3_client:
+    :param hathor_core_client: Client for make hathor-core requests, default to domain HathorCoreClient
+    :type hathor_core_client:
     """
-    def __init__(
-        self,
-        s3_client: Union[S3Client, None] = None,
-        hathor_core_client: Union[HathorCoreClient, None] = None
-    ) -> None:
-        self.s3_client = s3_client or S3Client()
+    def __init__(self, hathor_core_client: Union[HathorCoreClient, None] = None) -> None:
         self.hathor_core_client = hathor_core_client or HathorCoreClient()
 
-    def get_token_metadata_from_s3(self, file: str) -> Union[TokenMetadata, None]:
-        """Retrieve token metadata from a json file stored in s3
-
-        :param file: file name
-        :type file: str
-        :raises Exception: The name of the bucket used to store the jsons must be on config
-        :return: TokenMetadata object
-        :rtype: TokenMetadata
-        """
-        token_metadata_bucket = TOKEN_METADATA_BUCKET
-
-        if token_metadata_bucket is None:
-            raise Exception('No bucket name in config')
-
-        token_raw_metadata = self.s3_client.load_file(token_metadata_bucket, file)
-        if token_raw_metadata is None:
-            return None
-
-        token_metadata = json.loads(token_raw_metadata)
-        return TokenMetadata.from_dict(token_metadata)
-
     def get_token(self, id: str) -> Union[Token, None]:
         response = self.hathor_core_client.get(TOKEN_ENDPOINT, {'id': id})