github-starred/flowsint
2
0
Fork 0
mirror of https://github.com/reconurge/flowsint.git synced 2026-07-22 18:41:11 -05:00
Code Issues 980 Packages Projects Releases 14 Wiki Activity

fix(transforms): etherscan api url

Browse Source
This commit is contained in:
dextmorgn
2025-11-01 13:16:06 +01:00
parent 7ab921f0da
commit f47f1de3f1
2 changed files with 8 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
flowsint-transforms/src/flowsint_transforms/crypto/to_nfts.py
+5 -7
View File
@@ -9,11 +9,9 @@ from dotenv import load_dotenv
load_dotenv()
ETHERSCAN_API_URL = os.getenv("ETHERSCAN_API_URL")
class CryptoWalletAddressToNFTs(Transform):
"""Resolve NFTs for a wallet address (ETH)."""
"""[ETHERSCAN] Resolve NFTs for a wallet address (ETH)."""
# Define types as class attributes - base class handles schema generation automatically
InputType = List[CryptoWallet]
@@ -51,15 +49,15 @@ class CryptoWalletAddressToNFTs(Transform):
{
"name": "ETHERSCAN_API_KEY",
"type": "vaultSecret",
"description": "The Etherscan API key to use for the transaction lookup.",
"description": "The Etherscan API key to use for the NFT lookup.",
"required": True,
},
{
"name": "ETHERSCAN_API_URL",
"type": "url",
"description": "The Etherscan API URL to use for the transaction lookup.",
"description": "The Etherscan API URL to use for the NFT lookup.",
"required": False,
"default": ETHERSCAN_API_URL,
"default": "https://api.etherscan.io/v2/api",
},
]
@@ -92,7 +90,7 @@ class CryptoWalletAddressToNFTs(Transform):
async def scan(self, data: InputType) -> OutputType:
results: OutputType = []
api_key = self.get_secret("ETHERSCAN_API_KEY", os.getenv("ETHERSCAN_API_KEY"))
api_url = self.get_params().get("ETHERSCAN_API_URL", None)
api_url = self.get_params().get("ETHERSCAN_API_URL", "https://api.etherscan.io/v2/api")
for d in data:
try:
nfts = self._get_nfts(d.address, api_key, api_url)
flowsint-transforms/src/flowsint_transforms/crypto/to_transactions.py
+3 -4
View File
@@ -11,14 +11,13 @@ from flowsint_core.core.logger import Logger
load_dotenv()
ETHERSCAN_API_URL = os.getenv("ETHERSCAN_API_URL")
def wei_to_eth(wei_str):
return int(wei_str) / 10**18
class CryptoWalletAddressToTransactions(Transform):
"""[ETHERSCAN] Resolve transactions for a wallet address (ETH)."""
# Define types as class attributes - base class handles schema generation automatically
InputType = List[CryptoWallet]
@@ -64,7 +63,7 @@ class CryptoWalletAddressToTransactions(Transform):
"type": "url",
"description": "The Etherscan API URL to use for the transaction lookup.",
"required": False,
"default": ETHERSCAN_API_URL,
"default": "https://api.etherscan.io/v2/api",
},
]
@@ -97,7 +96,7 @@ class CryptoWalletAddressToTransactions(Transform):
async def scan(self, data: InputType) -> OutputType:
results: OutputType = []
api_key = self.get_secret("ETHERSCAN_API_KEY", os.getenv("ETHERSCAN_API_KEY"))
api_url = self.get_params().get("ETHERSCAN_API_URL", ETHERSCAN_API_URL)
api_url = self.get_params().get("ETHERSCAN_API_URL", "https://api.etherscan.io/v2/api")
for d in data:
try:
transactions = await self._get_transactions(d.address, api_key, api_url)