Source code for langchain_community.document_loaders.mintbase

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


[docs]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
[docs] 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}")
[docs] 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
[docs] 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.")