()
| 24 | |
| 25 | |
| 26 | async def main(): |
| 27 | bearer_token = os.getenv("ENVIO_API_TOKEN") |
| 28 | if not bearer_token: |
| 29 | raise ValueError("ENVIO_API_TOKEN environment variable is required. Please set it in your .env file.") |
| 30 | |
| 31 | client = hypersync.HypersyncClient(ClientConfig( |
| 32 | url="https://eth.hypersync.xyz/", |
| 33 | bearer_token=bearer_token |
| 34 | )) |
| 35 | height = await client.get_height() |
| 36 | start_block = height - 8000 |
| 37 | total_blocks = height - start_block |
| 38 | |
| 39 | query = hypersync.preset_query_blocks_and_transactions(start_block, height) |
| 40 | |
| 41 | # start the stream |
| 42 | receiver = await client.stream(query, hypersync.StreamConfig()) |
| 43 | |
| 44 | print(f"Starting the stream from block {start_block} to {height}...") |
| 45 | start_time = time.time() |
| 46 | total_processed_blocks = 0 |
| 47 | total_transactions = 0 |
| 48 | |
| 49 | with tqdm(total=total_blocks, desc="Processing blocks", unit="blocks") as pbar: |
| 50 | while True: |
| 51 | res = await receiver.recv() |
| 52 | # exit if the stream finished |
| 53 | if res is None: |
| 54 | break |
| 55 | |
| 56 | blocks_in_batch = len(res.data.blocks) |
| 57 | total_processed_blocks += blocks_in_batch |
| 58 | total_transactions += len(res.data.transactions) |
| 59 | |
| 60 | # Update progress bar |
| 61 | pbar.update(blocks_in_batch) |
| 62 | pbar.set_postfix({ |
| 63 | "Current height": res.next_block, |
| 64 | "Transactions": total_transactions |
| 65 | }) |
| 66 | |
| 67 | end_time = time.time() |
| 68 | duration = end_time - start_time |
| 69 | |
| 70 | print("\nStream completed!") |
| 71 | print(f"Total time taken: {duration:.2f} seconds") |
| 72 | print(f"Total blocks processed: {total_processed_blocks}") |
| 73 | print(f"Total transactions processed: {total_transactions}") |
| 74 | |
| 75 | if __name__ == "__main__": |
| 76 | asyncio.run(main()) |
no test coverage detected