blockbuster_async_blocking_call_detection_quickstart.py

1import blockbuster
2import time
3import asyncio
4
5# 1. Activate blockbuster to monitor the current thread
6blockbuster.activate()
7
8async def main():
9    print("Starting async loop...")
10    
11    # 2. This synchronous sleep would normally block the entire loop.
12    # Blockbuster will intercept this and raise a BlockedError or log it.
13    try:
14        time.sleep(1)
15    except Exception as e:
16        print(f"Captured blocking call: {e}")
17
18    print("Async loop finished.")
19
20if __name__ == "__main__":
21    asyncio.run(main())
22
23# 3. Deactivate when finished
24blockbuster.deactivate()