DocumentationBlog

Getting started with the Python SDK

After you installed the Python SDK, you can start indexing blockchain data. The SDK provides an IndexerRunner that tracks the state of the indexer and invokes the data handler for each item received from the Apibara stream.

Configuration#

The IndexerRunnerConfiguration object is used to store the indexer configuration. The options are:

  • apibara_url: the URL of the Apibara stream,
  • apibara_ssl: connect to the stream using HTTP/2 SSL,
  • storage_url: a MongoDB connection string to store the indexer state and the application data.

The following configuration is used to connect to the StarkNet Goerli stream and store data to a MongoDB instance running locally.

from apibara.indexer.runner import IndexerRunnerConfiguration
config = IndexerRunnerConfiguration(
apibara_url="goerli.starknet.stream.apibara.com:443",
apibara_ssl=True,
storage_url="mongodb://apibara:apibara@localhost:27017",
),

Indexing Web3 events#

The SDK version 0.5 supports indexing blockchain events, grouped by block.

Start by defining a function to handle the events received from the stream. Notice that the function must be async.

async def handle_events(info: Info, block_events: NewEvents):
# your handler here

Then you can create the IndexerRunner object.

runner = IndexerRunner(
config=config,
reset_state=False,
indexer_id=indexer_id,
new_events_handler=handle_events,
)

The indexer needs to know what events to index, a list of initial event filters can be added as follows:

runner.add_event_filters(
filters=[
EventFilter.from_event_name(name="Transfer")
],
index_from_block=180_000,
)

The EventFilter.from_event_name also supports an optional address argument to only receive events from the specified contract. In this case, we also specify which block to start indexing from to avoid processing 180,000 blocks that don't contain relevant events.

You can share state between calls to the event handler by setting a context object. This context object can be any Python object.

runner.set_context({
"network": "starknet-goerli"
})

The context is available in the event handler as info.context.

The final step is to run the indexer.

await runner.run()

Chain-aware database storage#

The Python SDK provides an abstraction over storage to store events together with information about the block they were generated at. The SDK automatically deletes data in response to chain reorganizations, so that you don't need to handle the low-level details yourself.

Refer to the API reference to learn more about the Storage interface.