kopia lustrzana https://github.com/bugout-dev/moonstream
fixed address_info response schema
rodzic
52677e59fb
commit
dd071642b5
|
|
@ -51,15 +51,6 @@ 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"
|
||||
|
|
@ -68,11 +59,6 @@ class LabelNames(Enum):
|
|||
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
|
||||
)
|
||||
|
|
@ -80,6 +66,9 @@ def get_ethereum_address_info(
|
|||
if id is None:
|
||||
return None
|
||||
|
||||
address_info = data.EthereumAddressInfo(address=address)
|
||||
etherscan_address_url = f"https://etherscan.io/address/{address}"
|
||||
blockchain_com_url = f"https://www.blockchain.com/eth/address/{address}"
|
||||
# Checking for token:
|
||||
coinmarketcap_label: Optional[EthereumLabel] = (
|
||||
db_session.query(EthereumLabel)
|
||||
|
|
@ -90,13 +79,15 @@ def get_ethereum_address_info(
|
|||
.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
|
||||
address_info.token = data.EthereumTokenDetails(
|
||||
name=coinmarketcap_label.label_data["name"],
|
||||
symbol=coinmarketcap_label.label_data["symbol"],
|
||||
external_url=[
|
||||
coinmarketcap_label.label_data["coinmarketcap_url"],
|
||||
etherscan_address_url,
|
||||
blockchain_com_url,
|
||||
],
|
||||
)
|
||||
|
||||
# Checking for smart contract
|
||||
etherscan_label: Optional[EthereumLabel] = (
|
||||
|
|
@ -108,10 +99,10 @@ def get_ethereum_address_info(
|
|||
.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
|
||||
address_info.smart_contract = data.EthereumSmartContractDetails(
|
||||
name=etherscan_label.label_data["name"],
|
||||
external_url=[etherscan_address_url, blockchain_com_url],
|
||||
)
|
||||
|
||||
return address_info
|
||||
|
||||
|
|
|
|||
|
|
@ -157,16 +157,21 @@ class EthereumSmartContractSourceInfo(BaseModel):
|
|||
compiler_version: str
|
||||
|
||||
|
||||
class EthereumAddressDetails(BaseModel):
|
||||
class EthereumTokenDetails(BaseModel):
|
||||
name: Optional[str]
|
||||
symbol: Optional[str]
|
||||
external_url: List[str] = []
|
||||
|
||||
|
||||
class EthereumSmartContractDetails(BaseModel):
|
||||
name: Optional[str]
|
||||
external_url: List[str] = []
|
||||
|
||||
|
||||
class EthereumAddressInfo(BaseModel):
|
||||
address: str
|
||||
address_type: str
|
||||
details: EthereumAddressDetails = EthereumAddressDetails()
|
||||
token: Optional[EthereumTokenDetails]
|
||||
smart_contract: Optional[EthereumSmartContractDetails]
|
||||
|
||||
|
||||
class TxinfoEthereumBlockchainResponse(BaseModel):
|
||||
|
|
|
|||
|
|
@ -1,10 +1,9 @@
|
|||
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 import FastAPI, Depends, Query, HTTPException
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from moonstreamdb.db import yield_db_session
|
||||
from sqlalchemy.orm import Session
|
||||
|
|
@ -18,7 +17,8 @@ from ..version import MOONSTREAM_VERSION
|
|||
logger = logging.getLogger(__name__)
|
||||
|
||||
tags_metadata = [
|
||||
{"name": "addressinfo", "description": "Addresses public information."},
|
||||
{"name": "addressinfo", "description": "Address public information."},
|
||||
{"name": "labels", "description": "Addresses label information."},
|
||||
]
|
||||
|
||||
app = FastAPI(
|
||||
|
|
@ -55,3 +55,33 @@ async def addressinfo_handler(
|
|||
) -> Optional[data.EthereumAddressInfo]:
|
||||
response = actions.get_ethereum_address_info(db_session, address)
|
||||
return response
|
||||
|
||||
|
||||
@app.get(
|
||||
"/labels/ethereum_blockchain",
|
||||
tags=["labels bul"],
|
||||
response_model=data.AddressListLabelsResponse,
|
||||
)
|
||||
async def addresses_labels_bulk_handler(
|
||||
addresses: Optional[str] = Query(None),
|
||||
start: int = Query(0),
|
||||
limit: int = Query(100),
|
||||
db_session: Session = Depends(yield_db_session),
|
||||
) -> data.AddressListLabelsResponse:
|
||||
"""
|
||||
Fetch labels with additional public information
|
||||
about known addresses.
|
||||
"""
|
||||
if limit > 100:
|
||||
raise HTTPException(
|
||||
status_code=406, detail="The limit cannot exceed 100 addresses"
|
||||
)
|
||||
try:
|
||||
addresses_response = actions.get_address_labels(
|
||||
db_session=db_session, start=start, limit=limit, addresses=addresses
|
||||
)
|
||||
except Exception as err:
|
||||
logger.error(f"Unable to get info about Ethereum addresses {err}")
|
||||
raise HTTPException(status_code=500)
|
||||
|
||||
return addresses_response
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ from typing import Dict, Optional
|
|||
|
||||
from sqlalchemy.sql.expression import true
|
||||
|
||||
from fastapi import FastAPI, Depends, HTTPException, Query
|
||||
from fastapi import FastAPI, Depends
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from moonstreamdb.db import yield_db_session
|
||||
from moonstreamdb.models import EthereumAddress
|
||||
|
|
@ -27,7 +27,6 @@ logger = logging.getLogger(__name__)
|
|||
|
||||
tags_metadata = [
|
||||
{"name": "txinfo", "description": "Ethereum transactions info."},
|
||||
{"name": "address info", "description": "Addresses public information."},
|
||||
]
|
||||
|
||||
app = FastAPI(
|
||||
|
|
@ -92,31 +91,3 @@ async def txinfo_ethereum_blockchain_handler(
|
|||
response.smart_contract_address = txinfo_request.tx.to_address
|
||||
response.is_smart_contract_call = True
|
||||
return response
|
||||
|
||||
|
||||
@app.get(
|
||||
"/addresses", tags=["address info"], response_model=data.AddressListLabelsResponse
|
||||
)
|
||||
async def addresses_labels_handler(
|
||||
addresses: Optional[str] = Query(None),
|
||||
start: int = Query(0),
|
||||
limit: int = Query(100),
|
||||
db_session: Session = Depends(yield_db_session),
|
||||
) -> data.AddressListLabelsResponse:
|
||||
"""
|
||||
Fetch labels with additional public information
|
||||
about known addresses.
|
||||
"""
|
||||
if limit > 100:
|
||||
raise HTTPException(
|
||||
status_code=406, detail="The limit cannot exceed 100 addresses"
|
||||
)
|
||||
try:
|
||||
addresses_response = actions.get_address_labels(
|
||||
db_session=db_session, start=start, limit=limit, addresses=addresses
|
||||
)
|
||||
except Exception as err:
|
||||
logger.error(f"Unable to get info about Ethereum addresses {err}")
|
||||
raise HTTPException(status_code=500)
|
||||
|
||||
return addresses_response
|
||||
|
|
|
|||
Ładowanie…
Reference in New Issue