[docs]classMintbaseDocumentLoader(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_addressself.blockchainType=blockchain_typeself.api_key=os.environ.get("MB_API_KEY")orapi_keyself.table="mb_views_nft_tokens"ortableself.select='where: {nft_contract_id: {_eq: "contract_address"}}'orselectself.fields=fieldsor["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_tokensself.max_execution_time=max_execution_timeifnotself.api_key:raiseValueError("Mintbase API key not provided.")ifnotre.match(r"^(([a-z\d]+[\-_])*[a-z\d]+\.)*([a-z\d]+[\-_])*[a-z\d]+$",self.contract_address,):raiseValueError(f"Invalid contract address {self.contract_address}")
[docs]defload(self)->List[Document]:result=[]start_time=time.time()whileTrue:# Define the GraphQL query as a multi-line stringoperations_doc=""" query MyQuery { table(select) { fields } } """# Replace the placeholder with the actual contract addressoperations_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 headersheaders={"mb-api-key":self.api_key,"Content-Type":"application/json"}# Define the POST datadata={"query":operations_doc,"variables":{},"operationName":"MyQuery",}url=f"https://graph.mintbase.xyz/{self.blockchainType}"response=requests.post(url,headers=headers,data=json.dumps(data))ifresponse.status_code!=200:raiseValueError(f"Request failed with status code {response.status_code}")items=response.json()["data"]["mb_views_nft_tokens"]ifnotitems:breakforiteminitems: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 Falseifnotself.get_all_tokens:breakif(self.max_execution_timeisnotNoneand(time.time()-start_time)>self.max_execution_time):raiseRuntimeError("Execution time exceeded the allowed time limit.")ifnotresult:raiseValueError(f"No NFTs found for contract address {self.contract_address}")returnresult
[docs]deflazy_load(self)->Iterator[Document]:start_time=time.time()whileTrue:# Define the GraphQL query as a multi-line stringoperations_doc=""" query MyQuery { table(select) { fields } } """# Replace the placeholder with the actual contract addressoperations_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 headersheaders={"mb-api-key":self.api_key,"Content-Type":"application/json"}# Define the POST datadata={"query":operations_doc,"variables":{},"operationName":"MyQuery",}url=f"https://graph.mintbase.xyz/{self.blockchainType}"response=requests.post(url,headers=headers,data=json.dumps(data))ifresponse.status_code!=200:raiseValueError(f"Request failed with status code {response.status_code}")items=response.json()["data"]["mb_views_nft_tokens"]ifnotitems:breakforiteminitems:content=str(item)tokenId=item["token_id"]metadata={"source":self.contract_address,"blockchain":self.blockchainType,"tokenId":tokenId,}yieldDocument(page_content=content,metadata=metadata)# exit after the first API call if get_all_tokens is Falseifnotself.get_all_tokens:breakif(self.max_execution_timeisnotNoneand(time.time()-start_time)>self.max_execution_time):raiseRuntimeError("Execution time exceeded the allowed time limit.")