Source code for langchain_community.document_loaders.etherscan
import os
import re
from typing import Iterator, List
import requests
from langchain_core.documents import Document
from langchain_community.document_loaders.base import BaseLoader
[docs]class EtherscanLoader(BaseLoader):
"""Load transactions from `Ethereum` mainnet.
The Loader use Etherscan API to interact with Ethereum mainnet.
ETHERSCAN_API_KEY environment variable must be set use this loader.
"""
[docs] def __init__(
self,
account_address: str,
api_key: str = "docs-demo",
filter: str = "normal_transaction",
page: int = 1,
offset: int = 10,
start_block: int = 0,
end_block: int = 99999999,
sort: str = "desc",
):
self.account_address = account_address
self.api_key = os.environ.get("ETHERSCAN_API_KEY") or api_key
self.filter = filter
self.page = page
self.offset = offset
self.start_block = start_block
self.end_block = end_block
self.sort = sort
if not self.api_key:
raise ValueError("Etherscan API key not provided")
if not re.match(r"^0x[a-fA-F0-9]{40}$", self.account_address):
raise ValueError(f"Invalid contract address {self.account_address}")
if filter not in [
"normal_transaction",
"internal_transaction",
"erc20_transaction",
"eth_balance",
"erc721_transaction",
"erc1155_transaction",
]:
raise ValueError(f"Invalid filter {filter}")
[docs] def lazy_load(self) -> Iterator[Document]:
"""Lazy load Documents from table."""
result = []
if self.filter == "normal_transaction":
result = self.getNormTx()
elif self.filter == "internal_transaction":
result = self.getInternalTx()
elif self.filter == "erc20_transaction":
result = self.getERC20Tx()
elif self.filter == "eth_balance":
result = self.getEthBalance()
elif self.filter == "erc721_transaction":
result = self.getERC721Tx()
elif self.filter == "erc1155_transaction":
result = self.getERC1155Tx()
else:
raise ValueError(f"Invalid filter {filter}")
for doc in result:
yield doc
[docs] def getNormTx(self) -> List[Document]:
url = (
f"https://api.etherscan.io/api?module=account&action=txlist&address={self.account_address}"
f"&startblock={self.start_block}&endblock={self.end_block}&page={self.page}"
f"&offset={self.offset}&sort={self.sort}&apikey={self.api_key}"
)
try:
response = requests.get(url)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print("Error occurred while making the request:", e) # noqa: T201
items = response.json()["result"]
result = []
if len(items) == 0:
return [Document(page_content="")]
for item in items:
content = str(item)
metadata = {"from": item["from"], "tx_hash": item["hash"], "to": item["to"]}
result.append(Document(page_content=content, metadata=metadata))
print(len(result)) # noqa: T201
return result
[docs] def getEthBalance(self) -> List[Document]:
url = (
f"https://api.etherscan.io/api?module=account&action=balance"
f"&address={self.account_address}&tag=latest&apikey={self.api_key}"
)
try:
response = requests.get(url)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print("Error occurred while making the request:", e) # noqa: T201
return [Document(page_content=response.json()["result"])]
[docs] def getInternalTx(self) -> List[Document]:
url = (
f"https://api.etherscan.io/api?module=account&action=txlistinternal"
f"&address={self.account_address}&startblock={self.start_block}"
f"&endblock={self.end_block}&page={self.page}&offset={self.offset}"
f"&sort={self.sort}&apikey={self.api_key}"
)
try:
response = requests.get(url)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print("Error occurred while making the request:", e) # noqa: T201
items = response.json()["result"]
result = []
if len(items) == 0:
return [Document(page_content="")]
for item in items:
content = str(item)
metadata = {"from": item["from"], "tx_hash": item["hash"], "to": item["to"]}
result.append(Document(page_content=content, metadata=metadata))
return result
[docs] def getERC20Tx(self) -> List[Document]:
url = (
f"https://api.etherscan.io/api?module=account&action=tokentx"
f"&address={self.account_address}&startblock={self.start_block}"
f"&endblock={self.end_block}&page={self.page}&offset={self.offset}"
f"&sort={self.sort}&apikey={self.api_key}"
)
try:
response = requests.get(url)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print("Error occurred while making the request:", e) # noqa: T201
items = response.json()["result"]
result = []
if len(items) == 0:
return [Document(page_content="")]
for item in items:
content = str(item)
metadata = {"from": item["from"], "tx_hash": item["hash"], "to": item["to"]}
result.append(Document(page_content=content, metadata=metadata))
return result
[docs] def getERC721Tx(self) -> List[Document]:
url = (
f"https://api.etherscan.io/api?module=account&action=tokennfttx"
f"&address={self.account_address}&startblock={self.start_block}"
f"&endblock={self.end_block}&page={self.page}&offset={self.offset}"
f"&sort={self.sort}&apikey={self.api_key}"
)
try:
response = requests.get(url)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print("Error occurred while making the request:", e) # noqa: T201
items = response.json()["result"]
result = []
if len(items) == 0:
return [Document(page_content="")]
for item in items:
content = str(item)
metadata = {"from": item["from"], "tx_hash": item["hash"], "to": item["to"]}
result.append(Document(page_content=content, metadata=metadata))
return result
[docs] def getERC1155Tx(self) -> List[Document]:
url = (
f"https://api.etherscan.io/api?module=account&action=token1155tx"
f"&address={self.account_address}&startblock={self.start_block}"
f"&endblock={self.end_block}&page={self.page}&offset={self.offset}"
f"&sort={self.sort}&apikey={self.api_key}"
)
try:
response = requests.get(url)
response.raise_for_status()
except requests.exceptions.RequestException as e:
print("Error occurred while making the request:", e) # noqa: T201
items = response.json()["result"]
result = []
if len(items) == 0:
return [Document(page_content="")]
for item in items:
content = str(item)
metadata = {"from": item["from"], "tx_hash": item["hash"], "to": item["to"]}
result.append(Document(page_content=content, metadata=metadata))
return result
