Cookbook — bidiwave¶
Example 1: Console logger¶
Capture all console logs from a page.
import asyncio
from datetime import datetime
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9222/session") as client:
logs: list[str] = []
async def on_log(entry):
ts = datetime.fromtimestamp(entry.timestamp / 1000).strftime("%H:%M:%S.%f")[:-3]
logs.append(f"[{ts}] [{entry.level.upper()}] {entry.text}")
print(logs[-1])
client.on("log.entryAdded", on_log)
await client.session.subscribe(["log.entryAdded"])
async with await client.browsing.open("https://example.com") as page:
await asyncio.sleep(5)
asyncio.run(main())
Example 2: Full page screenshot¶
import asyncio
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9222/session") as client:
async with await client.browsing.open("https://example.com") as page:
screenshot = await page.screenshot()
with open("screenshot.png", "wb") as f:
f.write(screenshot)
print("Screenshot saved to screenshot.png")
asyncio.run(main())
Example 3: Extract data from a page¶
import asyncio
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9222/session") as client:
async with await client.browsing.open("https://news.ycombinator.com") as page:
result = await page.call(
"""() => {
const rows = document.querySelectorAll('.athing');
return Array.from(rows).slice(0, 10).map(row => {
const title = row.querySelector('.titleline > a');
return {
id: row.id,
title: title?.textContent,
url: title?.href,
};
});
}""",
)
for item in result.value:
print(f"[{item['id']}] {item['title']} — {item['url']}")
asyncio.run(main())
Example 4: Multiple tabs in parallel¶
import asyncio
from bidiwave import BiDiClient, StringValue
async def scrape_page(client, url):
async with await client.browsing.open(url) as page:
result = await page.evaluate("document.title")
match result:
case StringValue(value=title):
return title
case _:
return "unknown"
async def main():
async with await BiDiClient.connect("ws://localhost:9222/session") as client:
urls = [
"https://example.com",
"https://example.org",
"https://example.net",
]
titles = await asyncio.gather(*[scrape_page(client, url) for url in urls])
for url, title in zip(urls, titles):
print(f"{url} → {title}")
asyncio.run(main())
Example 5: Wait for an element to appear¶
import asyncio
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9222/session") as client:
async with await client.browsing.open("https://example.com") as page:
found = await page.wait_for_selector("h1", timeout=5)
print(f"Element found: {found}")
asyncio.run(main())
Example 6: Reconnect handling¶
import asyncio
from bidiwave import BiDiClient, ClientConfig
async def main():
config = ClientConfig(max_retries=5)
async with await BiDiClient.connect("ws://localhost:9222/session", config=config) as client:
async def on_reconnect(_):
print("Reconnected. Recreating session...")
await client.session.subscribe(["log.entryAdded"])
client.on_reconnect(on_reconnect)
async def on_disconnect(_):
print("Connection lost. Retrying...")
client.on_disconnect(on_disconnect)
while True:
await asyncio.sleep(1)
asyncio.run(main())
Example 7: Capability-aware code¶
import asyncio
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9222/session") as client:
print(f"Browser: {client.capabilities.browser_name} {client.capabilities.browser_version}")
print(f"Browsing: {client.capabilities.supports_browsing}")
print(f"Script: {client.capabilities.supports_script}")
print(f"Network: {client.capabilities.supports_network}")
if client.capabilities.supports_network:
# Use network module
...
else:
print("Network not supported on this browser")
asyncio.run(main())
Example 8: Simulate user input¶
import asyncio
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9222/session") as client:
async with await client.browsing.open("https://example.com") as page:
ctx = page.id
# Click an element
await client.input.click(ctx, x=100, y=200)
# Type text
await client.input.type_text(ctx, "Hello, world!")
# Press Enter
await client.input.press_key(ctx, "Enter")
# Scroll down 500px
await client.input.scroll(ctx, delta_y=500)
# Drag and drop
await client.input.drag_and_drop(ctx, 100, 100, 300, 300)
asyncio.run(main())
Example 9: Block network requests¶
import asyncio
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9222/session") as client:
# Block all requests to ad domains
intercept = await client.network.add_intercept(
phases=["beforeRequestSent"],
url_patterns=["*ads.example.com*", "*doubleclick.net*"],
)
async with await client.browsing.open("https://example.com") as page:
await asyncio.sleep(3)
# Remove the intercept
await client.network.remove_intercept(intercept.intercept)
asyncio.run(main())
Example 10: Monitor network requests¶
import asyncio
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9222/session") as client:
requests: list[str] = []
async def on_request(event):
url = event.request.url
method = event.request.method
requests.append(f"{method} {url}")
print(f"→ {method} {url}")
async def on_response(event):
status = event.response.status
url = event.request.url
print(f"← {status} {url}")
client.on_request(on_request)
client.on_response(on_response)
await client.session.subscribe([
"network.beforeRequestSent",
"network.responseCompleted",
])
async with await client.browsing.open("https://example.com") as page:
await asyncio.sleep(3)
print(f"\nTotal requests: {len(requests)}")
asyncio.run(main())
Example 11: Mock an HTTP response¶
import asyncio
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9222/session") as client:
# Intercept at beforeRequestSent
intercept = await client.network.add_intercept(
phases=["beforeRequestSent"],
url_patterns=["*api.example.com/data*"],
)
# When intercepted, provide a synthetic response
async def on_request(event):
await client.network.provide_response(
request=event.request.request,
status_code=200,
reason_phrase="OK",
body="eyJtZXNzYWdlIjogIm1vY2tlZCJ9", # {"message": "mocked"}
)
client.on_request(on_request)
async with await client.browsing.open("https://example.com") as page:
await asyncio.sleep(3)
await client.network.remove_intercept(intercept.intercept)
asyncio.run(main())
Example 12: Set viewport for mobile testing¶
import asyncio
from bidiwave import BiDiClient, ViewportSize
async def main():
async with await BiDiClient.connect("ws://localhost:9515/session") as client:
async with await client.browsing.open("https://example.com") as page:
# Simulate iPhone 14 Pro viewport
await client.browsing.set_viewport(
page.id,
viewport=ViewportSize(width=393, height=852),
device_pixel_ratio=3.0,
)
# Take a mobile screenshot
screenshot = await page.screenshot()
with open("mobile.png", "wb") as f:
f.write(screenshot)
asyncio.run(main())
Example 13: Bypass the browser cache¶
import asyncio
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9515/session") as client:
async with await client.browsing.open("https://example.com") as page:
# Force every request in this context to hit the network
await client.network.set_cache_behavior("bypass", contexts=[page.id])
await page.reload()
# Restore normal caching
await client.network.set_cache_behavior("default", contexts=[page.id])
asyncio.run(main())
Example 14: Emulate a mobile device¶
import asyncio
from bidiwave import BiDiClient, ViewportSize
async def main():
async with await BiDiClient.connect("ws://localhost:9515/session") as client:
async with await client.browsing.open("https://example.com") as page:
ctx = page.id
# Set mobile viewport
await client.browsing.set_viewport(
ctx,
viewport=ViewportSize(width=375, height=812),
device_pixel_ratio=3.0,
)
# Simulate Tokyo location
await client.emulation.set_geolocation(
coordinates={"latitude": 35.6762, "longitude": 139.6503, "accuracy": 1.0},
contexts=[ctx],
)
# Simulate offline mode
await client.emulation.set_network_conditions(offline=True, contexts=[ctx])
# Set Tokyo timezone
await client.emulation.set_timezone("Asia/Tokyo", contexts=[ctx])
# Set mobile user agent
await client.emulation.set_user_agent(
"Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X)",
contexts=[ctx],
)
# Grant geolocation permission
await client.permissions.set_permission(
descriptor={"name": "geolocation"},
state="granted",
origin="https://example.com",
)
# Now test the page under these conditions
result = await page.evaluate("navigator.userAgent")
print(f"UA: {result.value}")
asyncio.run(main())
Example 15: Inject a preload script¶
import asyncio
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9515/session") as client:
# Inject a preload script that intercepts console.log
script_id = await client.preload.add_script(
"""() => {
const origLog = console.log;
window.__logs = [];
console.log = function(...args) {
window.__logs.push(args.join(' '));
origLog.apply(console, args);
};
}""",
)
async with await client.browsing.open("https://example.com") as page:
await page.evaluate("console.log('test message')")
logs = await page.evaluate("window.__logs")
print(f"Captured logs: {logs.value}")
await client.preload.remove_script(script_id)
asyncio.run(main())
Example 16: Auto-accept dialogs¶
import asyncio
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9515/session") as client:
# Accept every alert/confirm/prompt automatically
await client.set_auto_prompt(accept=True)
async with await client.browsing.open("https://example.com") as page:
await page.evaluate("alert('hello')")
await asyncio.sleep(1)
asyncio.run(main())
For prompts that need a value, use set_auto_prompt(accept=True,
user_text="..."). Disable with await client.disable_auto_prompt().
Example 17: Retrieve response body¶
import asyncio
import base64
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9515/session") as client:
collector = await client.network.add_data_collector(
data_types=["response"],
max_encoded_data_size=1_000_000,
)
request_ids: list[str] = []
async def on_response(event):
request_ids.append(event.request.request)
client.on_response(on_response)
await client.session.subscribe(["network.responseCompleted"])
async with await client.browsing.open("https://example.com") as page:
await asyncio.sleep(3)
for rid in request_ids:
data = await client.network.get_data(rid, data_type="response")
content = base64.b64decode(data["bytes"]["value"])
print(f"Response body: {len(content)} bytes")
await client.network.remove_data_collector(collector)
asyncio.run(main())
Example 18: CDP performance metrics¶
import asyncio
from bidiwave import BiDiClient
async def main():
async with await BiDiClient.connect("ws://localhost:9515/session") as client:
# Enable Performance domain via CDP
await client.cdp.send_command("Performance.enable")
async with await client.browsing.open("https://example.com") as page:
await asyncio.sleep(2)
# Get performance metrics
metrics = await client.cdp.send_command("Performance.getMetrics")
for metric in metrics["metrics"]:
print(f"{metric['name']}: {metric['value']}")
asyncio.run(main())