Merge pull request #188 from bugout-dev/address-info

Address info

backend/moonstream/actions.py

@@ -1,13 +1,14 @@
 import json
 import logging
 from typing import Dict, Any, List, Optional
-
+from enum import Enum
 import boto3  # type: ignore
 from moonstreamdb.models import (
     EthereumAddress,
     EthereumLabel,
 )
-from sqlalchemy.orm import Session
+from sqlalchemy import text
+from sqlalchemy.orm import Session, query, query_expression
 
 from . import data
 from .settings import ETHERSCAN_SMARTCONTRACTS_BUCKET
@@ -50,6 +51,71 @@ def get_contract_source_info(
     return None
 
 
+class AddressType(Enum):
+    UNKNOWN = "unknown"
+    REGULAR = "regular"
+    TOKEN = "token"
+    SMART_CONTRACT = "smart_contract"
+    NFT = "nft"
+    EXCHANGE = "exchange"
+
+
+class LabelNames(Enum):
+    ETHERSCAN_SMARTCONTRACT = "etherscan_smartcontract"
+    COINMARKETCAP_TOKEN = "coinmarketcap_token"
+
+
+def get_ethereum_address_info(
+    db_session: Session, address: str
+) -> Optional[data.EthereumAddressInfo]:
+
+    address_info = data.EthereumAddressInfo(
+        address=address, address_type=AddressType.UNKNOWN.value
+    )
+
+    query = db_session.query(EthereumAddress.id).filter(
+        EthereumAddress.address == address
+    )
+    id = query.one_or_none()
+    if id is None:
+        return None
+
+    # Checking for token:
+    coinmarketcap_label: Optional[EthereumLabel] = (
+        db_session.query(EthereumLabel)
+        .filter(EthereumLabel.address_id == id[0])
+        .filter(EthereumLabel.label == LabelNames.COINMARKETCAP_TOKEN.value)
+        .order_by(text("created_at desc"))
+        .limit(1)
+        .one_or_none()
+    )
+    if coinmarketcap_label is not None:
+        address_info.address_type = AddressType.TOKEN.value
+        address_info.details.name = coinmarketcap_label.label_data["name"]
+        address_info.details.symbol = coinmarketcap_label.label_data["symbol"]
+        address_info.details.external_url = [
+            coinmarketcap_label.label_data["coinmarketcap_url"]
+        ]
+        return address_info
+
+    # Checking for smart contract
+    etherscan_label: Optional[EthereumLabel] = (
+        db_session.query(EthereumLabel)
+        .filter(EthereumLabel.address_id == id[0])
+        .filter(EthereumLabel.label == LabelNames.ETHERSCAN_SMARTCONTRACT.value)
+        .order_by(text("created_at desc"))
+        .limit(1)
+        .one_or_none()
+    )
+    if etherscan_label is not None:
+        address_info.address_type = AddressType.SMART_CONTRACT.value
+        address_info.details.name = etherscan_label.label_data["name"]
+        address_info.details.external_url = [f"https://etherscan.io/address/{address}"]
+        return address_info
+
+    return address_info
+
+
 def get_address_labels(
     db_session: Session, start: int, limit: int, addresses: Optional[str] = None
 ) -> data.AddressListLabelsResponse:

backend/moonstream/api.py

@@ -12,6 +12,7 @@ from .routes.subscriptions import app as subscriptions_api
 from .routes.users import app as users_api
 from .routes.txinfo import app as txinfo_api
 from .routes.streams import app as streams_api
+from .routes.address_info import app as addressinfo_api
 
 from .settings import ORIGINS
 from .version import MOONSTREAM_VERSION
@@ -49,3 +50,4 @@ app.mount("/subscriptions", subscriptions_api)
 app.mount("/users", users_api)
 app.mount("/streams", streams_api)
 app.mount("/txinfo", txinfo_api)
+app.mount("/address_info", addressinfo_api)

backend/moonstream/data.py

@@ -157,6 +157,18 @@ class EthereumSmartContractSourceInfo(BaseModel):
     compiler_version: str
 
 
+class EthereumAddressDetails(BaseModel):
+    name: Optional[str]
+    symbol: Optional[str]
+    external_url: List[str] = []
+
+
+class EthereumAddressInfo(BaseModel):
+    address: str
+    address_type: str
+    details: EthereumAddressDetails = EthereumAddressDetails()
+
+
 class TxinfoEthereumBlockchainResponse(BaseModel):
     tx: EthereumTransaction
     is_smart_contract_deployment: bool = False

backend/moonstream/routes/address_info.py

@@ -0,0 +1,57 @@
+import logging
+from typing import Dict, List, Optional
+from sqlalchemy.orm.query import Query
+
+from sqlalchemy.sql.expression import true
+
+from fastapi import FastAPI, Depends, Form
+from fastapi.middleware.cors import CORSMiddleware
+from moonstreamdb.db import yield_db_session
+from sqlalchemy.orm import Session
+
+from .. import actions
+from .. import data
+from ..middleware import BroodAuthMiddleware
+from ..settings import DOCS_TARGET_PATH, ORIGINS, DOCS_PATHS
+from ..version import MOONSTREAM_VERSION
+
+logger = logging.getLogger(__name__)
+
+tags_metadata = [
+    {"name": "addressinfo", "description": "Addresses public information."},
+]
+
+app = FastAPI(
+    title=f"Moonstream users API.",
+    description="User, token and password handlers.",
+    version=MOONSTREAM_VERSION,
+    openapi_tags=tags_metadata,
+    openapi_url="/openapi.json",
+    docs_url=None,
+    redoc_url=f"/{DOCS_TARGET_PATH}",
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=ORIGINS,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+whitelist_paths: Dict[str, str] = {}
+whitelist_paths.update(DOCS_PATHS)
+app.add_middleware(BroodAuthMiddleware, whitelist=whitelist_paths)
+
+
+@app.get(
+    "/ethereum_blockchain",
+    tags=["addressinfo"],
+    response_model=data.EthereumAddressInfo,
+)
+async def addressinfo_handler(
+    address: str,
+    db_session: Session = Depends(yield_db_session),
+) -> Optional[data.EthereumAddressInfo]:
+    response = actions.get_ethereum_address_info(db_session, address)
+    return response