Querying Horizon

py-stellar-base gives you access to all the endpoints exposed by Horizon.

Building requests

py-stellar-base uses the Builder pattern to create the requests to send to Horizon. Starting with a Server object, you can chain methods together to generate a query. (See the Horizon reference documentation for what methods are possible.)

 1"""
 2See: https://stellar-sdk.readthedocs.io/en/latest/querying_horizon.html#building-requests
 3"""
 4
 5from stellar_sdk import Server
 6
 7server = Server(horizon_url="https://horizon.stellar.org")
 8account = "GB6NVEN5HSUBKMYCE5ZOWSK5K23TBWRUQLZY3KNMXUZ3AQ2ESC4MY4AQ"
 9
10# get a list of transactions that occurred in ledger 1400
11transactions = server.transactions().for_ledger(1400).call()
12print(transactions)
13
14# get a list of transactions submitted by a particular account
15transactions = server.transactions().for_account(account_id=account).call()
16print(transactions)
17
18# The following example will show you how to handle paging
19print(f"Gets all payment operations associated with {account}.")
20payments_records = []
21payments_call_builder = (
22    server.payments().for_account(account).order(desc=False).limit(10)
23)  # limit can be set to a maximum of 200
24payments_records += payments_call_builder.call()["_embedded"]["records"]
25page_count = 0
26while page_records := payments_call_builder.next()["_embedded"]["records"]:
27    payments_records += page_records
28    print(f"Page {page_count} fetched")
29    print(f"data: {page_records}")
30    page_count += 1
31print(f"Payments count: {len(payments_records)}")

Once the request is built, it can be invoked with call() or with stream(). call() will return the response given by Horizon.

Streaming requests

Many requests can be invoked with stream(). Instead of returning a result like call() does, stream() will return an EventSource. Horizon will start sending responses from either the beginning of time or from the point specified with cursor(). (See the Horizon reference documentation to learn which endpoints support streaming.)

For example, to log instances of transactions from a particular account:

 1"""
 2See: https://stellar-sdk.readthedocs.io/en/latest/querying_horizon.html#streaming-requests
 3"""
 4
 5from stellar_sdk import Server
 6
 7server = Server(horizon_url="https://horizon-testnet.stellar.org")
 8account_id = "GASOCNHNNLYFNMDJYQ3XFMI7BYHIOCFW3GJEOWRPEGK2TDPGTG2E5EDW"
 9last_cursor = "now"  # or load where you left off
10
11
12def tx_handler(tx_response):
13    print(tx_response)
14
15
16for tx in server.transactions().for_account(account_id).cursor(last_cursor).stream():
17    tx_handler(tx)