How to Subscribe to Real-Time Data Stream using Bitquery API and Python GraphQL Client

GraphQL Tutorials
1

We will use the python_graphql_client package to get real-time data from the Streaming APIs.
Here’s a step-by-step tutorial on how to use this code:

Step 1: Import required libraries

from python_graphql_client import GraphqlClient
from requests.auth import HTTPBasicAuth
import http.client
import asyncio

This code block imports the GraphqlClient library, HTTPBasicAuth, and http.client libraries required for this script.

Step 2: Set up authentication

auth = HTTPBasicAuth('', 'YOUR API KEY')

The Bitquery API requires authentication, and the above code sets up the authentication using HTTPBasicAuth with the API key as the password.

Step 3: Set up the GraphQL client

ws = GraphqlClient(endpoint="wss://streaming.bitquery.io/graphql",auth=auth)

This code sets up a GraphqlClient instance with the Bitquery WebSocket endpoint and authentication.

Step 4: Define the subscription callback function

def callback(response):
  print("got new OHLC data:")
  print(response)

This function is called every time new data is received from the WebSocket stream. The function prints the response

Step 5: Define the GraphQL subscription query

query = """
 subscription {
    EVM(network: bsc) {
      Transfers {
        Block {
          Number
          Hash
        }
        Transfer {
          Currency {
            Symbol
            Name
          }
          Amount
        }
      }
    }
  }
"""
await ws.subscribe(query=query, handle=callback)

This code block defines the GraphQL subscription query, which is sent to the Bitquery API to subscribe to the Binance Smart Chain network’s Transfer events.

Step 6: Make an asynchronous subscription request

asyncio.run(ws.subscribe(query=query, handle=print))

This code block makes an asynchronous subscription request to the Bitquery API with the defined subscription query and the handle parameter set to print. The handle parameter specifies the function that should be called when new data is received.

Final Code

import asyncio
from python_graphql_client import GraphqlClient
from requests.auth import HTTPBasicAuth

auth = HTTPBasicAuth('', 'YOUR API KEY')
ws = GraphqlClient(endpoint="wss://streaming.bitquery.io/graphql", auth=auth)


def callback(response):
  print("got new OHLC data:")
  print(response)


async def subscribe_to_transfers():
  query = """
    subscription {
    EVM(network: bsc) {
      Transfers {
        Block {
          Number
          Hash
        }
        Transfer {
          Currency {
            Symbol
            Name
          }
          Amount
        }
      }
    }
  }
  """

  await ws.subscribe(query=query, handle=callback)


asyncio.run(subscribe_to_transfers())

Finally, the asyncio.run method is called to run the subscription request asynchronously.

That’s it! By running this code, you should be able to see the data in your terminal as shown below.

image
image1431×304 52.4 KB

2 Likes