generator_the_graph_ens.py

# This Python file module is responsible for  compose and request to a GraphQL API using the "requests" library.
# The main function, "post_generator" takes in a URL, a query function, and optional parameters for "skip" and "block_number".
# This script requests ens domain subgraph. It can be changed by the use of another subgraph and query data.
# Recommended to use Graph Explorer with "API_KEY" and "SUBGRAPH_ID". For more information, visit https://thegraph.com/explorer
# In some situations, you can use "Hosted Service". But this option is not recommended due to 
# the fact that this method will soon become deprecated
# The "post_generator" function in the provided code is a generator that retrieves data from the GraphQL API
# in a continuous manner. It uses the "yield" keyword to generate a sequence of API responses. 
# This allows for data to be fetched in batches or pages, making it useful for handling large datasets.
# To launch the script, it is necessary to install the requests library and the python-dotenv library
# pip install python-dotenv
# pip install requests

import requests
import time
import json
from settings import API_KEY, SUBGRAPH_ID
import os
from dotenv import load_dotenv

load_dotenv()
API_KEY = os.environ.get('API_KEY')
SUBGRAPH_ID = os.environ.get('SUBGRAPH_ID')

def post_generator(url, query, interval = 0, interval_increase = 60, skip=0, block_number=0):

    # # get last block from db
    # domain = SomeModel.objects.order_by('block_number').last()
    #  
    # if domain is not None:
    #     block_number = domain.block_number
    # else:
    #     block_number = 0

    while True:
            
        response = requests.post(url, json = query(block_number, skip))

        # decode byte -> json
        json_response = response.content.decode('utf-8')

        # load json -> dict
        dict_response = json.loads(json_response)

        # Проверяем, что поле "nameRegistereds" является пустым списком
        try:

            if not dict_response['data']['nameRegistereds']:
                
                interval = interval_increase  # increase the interval between requests   
                time.sleep(interval) 

            else:

                interval = 0  # reset the interval in case of receiving a non-empty response
                skip += 1
                yield dict_response
   
            print('Data  response:', dict_response)

        except (KeyError, TypeError):
            
            print(f"Error response:", dict_response)


url = f"https://gateway.thegraph.com/api/{API_KEY}/subgraphs/id/{SUBGRAPH_ID}"
data = lambda block_number, skip : { "query": f'''
                    {{  nameRegistereds(where: {{blockNumber_gt: {block_number}}},
                                      first: 1000, skip: {1000 * skip}) {{
        				blockNumber
	    				registration {{
        					labelName
                  			registrant {{
                                  id
                                    }}
                            domain {{
                            	owner {{
                                    id
                                    }}
                                resolver {{
                                    texts
                                        }}
                                }}
                            }}                                      }}
                }}             
                        '''}


for response in post_generator(url, data, block_number=0):
    pass
    # now we can batch save this data,
    # perform additional queries, and so on