mirror of
https://github.com/hwchase17/langchain
synced 2024-11-10 01:10:59 +00:00
fa4124b821
- [x] **Add Near NFT loader**: "community: Load NFT near block chain
using mintbase graph API"
- [x] **PR message**:
- **Description:** a description of the change
- **Twitter handle:**Kurodenjiro
---------
Co-authored-by: Bagatur <22008038+baskaryan@users.noreply.github.com>
Co-authored-by: Bagatur <baskaryan@gmail.com>
265 lines
8.7 KiB
Python
265 lines
8.7 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import time
|
|
from typing import Iterator, List, Literal, Optional
|
|
|
|
import requests
|
|
from langchain_core.documents import Document
|
|
|
|
from langchain_community.document_loaders.base import BaseLoader
|
|
|
|
|
|
class MintbaseDocumentLoader(BaseLoader):
|
|
"""Load elements from a blockchain smart contract.
|
|
|
|
The supported blockchains are: Near mainnet, Near testnet.
|
|
|
|
If no BlockchainType is specified, the default is Near mainnet.
|
|
|
|
The Loader uses the Mintbase API to interact with the blockchain.
|
|
MB_API_KEY environment variable must be set to use this loader.
|
|
|
|
The API returns 100 NFTs per request and can be paginated using the
|
|
startToken parameter.
|
|
|
|
If get_all_tokens is set to True, the loader will get all tokens
|
|
on the contract. Note that for contracts with a large number of tokens,
|
|
this may take a long time (e.g. 10k tokens is 100 requests).
|
|
Default value is false for this reason.
|
|
|
|
The max_execution_time (sec) can be set to limit the execution time
|
|
of the loader.
|
|
|
|
Future versions of this loader can:
|
|
- Support additional Mintbase APIs (e.g. getTokens, etc.)
|
|
|
|
Example:
|
|
.. code-block:: python
|
|
|
|
contractAddress = "nft.yearofchef.near" # Year of chef contract address
|
|
blockchainLoader = MintbaseDocumentLoader(
|
|
contract_address=contractAddress, blockchain_type="mainnet",api_key="omni-site"
|
|
)
|
|
""" # noqa: E501
|
|
|
|
def __init__(
|
|
self,
|
|
contract_address: str,
|
|
*,
|
|
blockchain_type: Literal["mainnet", "testnet"],
|
|
api_key: str = "",
|
|
table: str = "",
|
|
select: str = "",
|
|
fields: Optional[List[str]] = None,
|
|
get_all_tokens: bool = False,
|
|
max_execution_time: Optional[int] = None,
|
|
):
|
|
"""
|
|
|
|
Args:
|
|
contract_address: The address of the smart contract.
|
|
blockchainType: The blockchain type.
|
|
api_key: The Mintbase API key.
|
|
table: name of the table to query
|
|
select: Conditions for querying
|
|
fields: Information to display after query
|
|
get_all_tokens: Whether to get all tokens on the contract.
|
|
max_execution_time: The maximum execution time (sec).
|
|
"""
|
|
self.contract_address = contract_address
|
|
self.blockchainType = blockchain_type
|
|
self.api_key = os.environ.get("MB_API_KEY") or api_key
|
|
self.table = "mb_views_nft_tokens" or table
|
|
self.select = 'where: {nft_contract_id: {_eq: "contract_address"}}' or select
|
|
self.fields = fields or [
|
|
"base_uri",
|
|
"burned_receipt_id",
|
|
"burned_timestamp",
|
|
"copies",
|
|
"description",
|
|
"expires_at",
|
|
"extra",
|
|
"issued_at",
|
|
"last_transfer_receipt_id",
|
|
"last_transfer_timestamp",
|
|
"media",
|
|
"media_hash",
|
|
"metadata_content_flag",
|
|
"metadata_id",
|
|
"mint_memo",
|
|
"minted_receipt_id",
|
|
"minted_timestamp",
|
|
"minter",
|
|
"nft_contract_content_flag",
|
|
"nft_contract_created_at",
|
|
"nft_contract_icon",
|
|
"nft_contract_id",
|
|
"nft_contract_is_mintbase",
|
|
"nft_contract_name",
|
|
"nft_contract_owner_id",
|
|
"nft_contract_reference",
|
|
"nft_contract_spec",
|
|
"nft_contract_symbol",
|
|
"owner",
|
|
"reference",
|
|
"reference_blob",
|
|
"reference_hash",
|
|
"royalties",
|
|
"royalties_percent",
|
|
"splits",
|
|
"starts_at",
|
|
"title",
|
|
"token_id",
|
|
"updated_at",
|
|
]
|
|
|
|
self.get_all_tokens = get_all_tokens
|
|
self.max_execution_time = max_execution_time
|
|
|
|
if not self.api_key:
|
|
raise ValueError("Mintbase API key not provided.")
|
|
|
|
if not re.match(
|
|
r"^(([a-z\d]+[\-_])*[a-z\d]+\.)*([a-z\d]+[\-_])*[a-z\d]+$",
|
|
self.contract_address,
|
|
):
|
|
raise ValueError(f"Invalid contract address {self.contract_address}")
|
|
|
|
def load(self) -> List[Document]:
|
|
result = []
|
|
|
|
start_time = time.time()
|
|
|
|
while True:
|
|
# Define the GraphQL query as a multi-line string
|
|
operations_doc = """
|
|
query MyQuery {
|
|
table(select) {
|
|
fields
|
|
}
|
|
}
|
|
"""
|
|
|
|
# Replace the placeholder with the actual contract address
|
|
operations_doc = operations_doc.replace("select", self.select)
|
|
operations_doc = operations_doc.replace(
|
|
"contract_address", self.contract_address
|
|
)
|
|
operations_doc = operations_doc.replace("table", self.table)
|
|
operations_doc = operations_doc.replace("fields", "\n".join(self.fields))
|
|
|
|
# Define the headers
|
|
headers = {"mb-api-key": self.api_key, "Content-Type": "application/json"}
|
|
# Define the POST data
|
|
data = {
|
|
"query": operations_doc,
|
|
"variables": {},
|
|
"operationName": "MyQuery",
|
|
}
|
|
|
|
url = f"https://graph.mintbase.xyz/{self.blockchainType}"
|
|
|
|
response = requests.post(url, headers=headers, data=json.dumps(data))
|
|
|
|
if response.status_code != 200:
|
|
raise ValueError(
|
|
f"Request failed with status code {response.status_code}"
|
|
)
|
|
|
|
items = response.json()["data"]["mb_views_nft_tokens"]
|
|
|
|
if not items:
|
|
break
|
|
|
|
for item in items:
|
|
content = str(item)
|
|
token_id = item["token_id"]
|
|
metadata = {
|
|
"source": self.contract_address,
|
|
"blockchain": self.blockchainType,
|
|
"tokenId": token_id,
|
|
}
|
|
result.append(Document(page_content=content, metadata=metadata))
|
|
|
|
# exit after the first API call if get_all_tokens is False
|
|
if not self.get_all_tokens:
|
|
break
|
|
|
|
if (
|
|
self.max_execution_time is not None
|
|
and (time.time() - start_time) > self.max_execution_time
|
|
):
|
|
raise RuntimeError("Execution time exceeded the allowed time limit.")
|
|
|
|
if not result:
|
|
raise ValueError(
|
|
f"No NFTs found for contract address {self.contract_address}"
|
|
)
|
|
|
|
return result
|
|
|
|
def lazy_load(self) -> Iterator[Document]:
|
|
start_time = time.time()
|
|
|
|
while True:
|
|
# Define the GraphQL query as a multi-line string
|
|
operations_doc = """
|
|
query MyQuery {
|
|
table(select) {
|
|
fields
|
|
}
|
|
}
|
|
"""
|
|
|
|
# Replace the placeholder with the actual contract address
|
|
operations_doc = operations_doc.replace("select", self.select)
|
|
operations_doc = operations_doc.replace(
|
|
"contract_address", self.contract_address
|
|
)
|
|
operations_doc = operations_doc.replace("table", self.table)
|
|
operations_doc = operations_doc.replace("fields", "\n".join(self.fields))
|
|
|
|
# Define the headers
|
|
headers = {"mb-api-key": self.api_key, "Content-Type": "application/json"}
|
|
# Define the POST data
|
|
data = {
|
|
"query": operations_doc,
|
|
"variables": {},
|
|
"operationName": "MyQuery",
|
|
}
|
|
|
|
url = f"https://graph.mintbase.xyz/{self.blockchainType}"
|
|
|
|
response = requests.post(url, headers=headers, data=json.dumps(data))
|
|
|
|
if response.status_code != 200:
|
|
raise ValueError(
|
|
f"Request failed with status code {response.status_code}"
|
|
)
|
|
|
|
items = response.json()["data"]["mb_views_nft_tokens"]
|
|
|
|
if not items:
|
|
break
|
|
|
|
for item in items:
|
|
content = str(item)
|
|
tokenId = item["token_id"]
|
|
metadata = {
|
|
"source": self.contract_address,
|
|
"blockchain": self.blockchainType,
|
|
"tokenId": tokenId,
|
|
}
|
|
yield Document(page_content=content, metadata=metadata)
|
|
|
|
# exit after the first API call if get_all_tokens is False
|
|
if not self.get_all_tokens:
|
|
break
|
|
|
|
if (
|
|
self.max_execution_time is not None
|
|
and (time.time() - start_time) > self.max_execution_time
|
|
):
|
|
raise RuntimeError("Execution time exceeded the allowed time limit.")
|