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.
