kopia lustrzana https://github.com/bugout-dev/moonstream
adding address_info endpoint
rodzic
949683efbb
commit
7107ffdc48
|
@ -3,7 +3,7 @@ import logging
|
||||||
|
|
||||||
|
|
||||||
from typing import Dict, Any, List, Optional, Union
|
from typing import Dict, Any, List, Optional, Union
|
||||||
|
from enum import Enum
|
||||||
from sqlalchemy.engine.base import Transaction
|
from sqlalchemy.engine.base import Transaction
|
||||||
from moonstreamdb.models import (
|
from moonstreamdb.models import (
|
||||||
EthereumBlock,
|
EthereumBlock,
|
||||||
|
@ -265,6 +265,52 @@ def parse_search_query_to_sqlalchemy_filters(q: str, allowed_addresses: List[str
|
||||||
return constructed_filters
|
return constructed_filters
|
||||||
|
|
||||||
|
|
||||||
|
class AdressType:
|
||||||
|
TOKEN = 0
|
||||||
|
SMART_CONTRACT = 1
|
||||||
|
NFT = 2
|
||||||
|
EXCHANGE = 3
|
||||||
|
|
||||||
|
|
||||||
|
class LabelNames(Enum):
|
||||||
|
ETHERSCAN_SMARTCONTRACT = "etherscan_smartcontract"
|
||||||
|
COINMARKETCAP_TOKEN = "coinmarketcap_token"
|
||||||
|
EXCHANGE = "EXCANGE"
|
||||||
|
|
||||||
|
|
||||||
|
def get_ethereum_address_info(
|
||||||
|
db_session: Session, address: str
|
||||||
|
) -> Optional[data.EthereumAddressInfo]:
|
||||||
|
query = db_session.query(EthereumAddress.id).filter(
|
||||||
|
EthereumAddress.address == address
|
||||||
|
)
|
||||||
|
id = query.one_or_none()
|
||||||
|
if id is None:
|
||||||
|
return None
|
||||||
|
labels = (
|
||||||
|
db_session.query(EthereumLabel).filter(EthereumLabel.address_id == id[0]).all()
|
||||||
|
)
|
||||||
|
address_info = data.EthereumAddressInfo(address=address)
|
||||||
|
for label in labels:
|
||||||
|
if label.label == LabelNames.ETHERSCAN_SMARTCONTRACT:
|
||||||
|
address_info.address_type = AdressType.SMART_CONTRACT
|
||||||
|
address_info.details.name = label.label_data["name"]
|
||||||
|
address_info.details.external_URL = (
|
||||||
|
f"https://etherscan.io/address/{address}"
|
||||||
|
)
|
||||||
|
elif label.label == LabelNames.COINMARKETCAP_TOKEN:
|
||||||
|
address_info.address_type = AdressType.TOKEN
|
||||||
|
address_info.details.name = label.label_data["name"]
|
||||||
|
address_info.details.symbol = label.label_data["symbol"]
|
||||||
|
address_info.details.external_URL = label.label_data["coinmarketcap_url"]
|
||||||
|
elif label.label == LabelNames.EXCHANGE:
|
||||||
|
address_info.address_type = AdressType.EXCHANGE
|
||||||
|
address_info.details.name = label.label_data["name"]
|
||||||
|
address_info.details.symbol = label.label_data["label"]
|
||||||
|
|
||||||
|
return address_info
|
||||||
|
|
||||||
|
|
||||||
def get_contract_source_info(
|
def get_contract_source_info(
|
||||||
db_session: Session, contract_address: str
|
db_session: Session, contract_address: str
|
||||||
) -> Optional[data.EthereumSmartContractSourceInfo]:
|
) -> Optional[data.EthereumSmartContractSourceInfo]:
|
||||||
|
|
|
@ -140,6 +140,18 @@ class EthereumSmartContractSourceInfo(BaseModel):
|
||||||
compiler_version: str
|
compiler_version: str
|
||||||
|
|
||||||
|
|
||||||
|
class EthereumAddressDetails(BaseModel):
|
||||||
|
name: Optional[str]
|
||||||
|
symbol: Optional[str]
|
||||||
|
external_URL: Optional[str]
|
||||||
|
|
||||||
|
|
||||||
|
class EthereumAddressInfo(BaseModel):
|
||||||
|
address: str
|
||||||
|
address_type: Optional[int]
|
||||||
|
details: EthereumAddressDetails = EthereumAddressDetails()
|
||||||
|
|
||||||
|
|
||||||
class TxinfoEthereumBlockchainResponse(BaseModel):
|
class TxinfoEthereumBlockchainResponse(BaseModel):
|
||||||
tx: EthereumTransaction
|
tx: EthereumTransaction
|
||||||
is_smart_contract_deployment: bool = False
|
is_smart_contract_deployment: bool = False
|
||||||
|
|
|
@ -0,0 +1,10 @@
|
||||||
|
import logging
|
||||||
|
from typing import Dict, Optional
|
||||||
|
|
||||||
|
from sqlalchemy.sql.expression import true
|
||||||
|
|
||||||
|
from fastapi import FastAPI, Depends, HTTPException, Query
|
||||||
|
from fastapi.middleware.cors import CORSMiddleware
|
||||||
|
from moonstreamdb.db import yield_db_session
|
||||||
|
from moonstreamdb.models import EthereumAddress
|
||||||
|
from sqlalchemy.orm import Session
|
|
@ -10,7 +10,7 @@ from typing import Dict, Optional
|
||||||
|
|
||||||
from sqlalchemy.sql.expression import true
|
from sqlalchemy.sql.expression import true
|
||||||
|
|
||||||
from fastapi import FastAPI, Depends, HTTPException, Query
|
from fastapi import FastAPI, Depends, HTTPException, Query, Form
|
||||||
from fastapi.middleware.cors import CORSMiddleware
|
from fastapi.middleware.cors import CORSMiddleware
|
||||||
from moonstreamdb.db import yield_db_session
|
from moonstreamdb.db import yield_db_session
|
||||||
from moonstreamdb.models import EthereumAddress
|
from moonstreamdb.models import EthereumAddress
|
||||||
|
@ -94,6 +94,18 @@ async def txinfo_ethereum_blockchain_handler(
|
||||||
return response
|
return response
|
||||||
|
|
||||||
|
|
||||||
|
@app.get(
|
||||||
|
"/ethereum_blockchain/address_info",
|
||||||
|
tags=["address info"],
|
||||||
|
response_model=data.EthereumAddressInfo,
|
||||||
|
)
|
||||||
|
async def address_info_handler(
|
||||||
|
address: str = Form(...), db_session: Session = Depends(yield_db_session)
|
||||||
|
) -> Optional[data.EthereumAddressInfo]:
|
||||||
|
response = actions.get_ethereum_address_info(db_session, address)
|
||||||
|
return response
|
||||||
|
|
||||||
|
|
||||||
@app.get(
|
@app.get(
|
||||||
"/addresses", tags=["address info"], response_model=data.AddressListLabelsResponse
|
"/addresses", tags=["address info"], response_model=data.AddressListLabelsResponse
|
||||||
)
|
)
|
||||||
|
|
Ładowanie…
Reference in New Issue