fluxwright

Run hundreds of headless Chrome jobs on one machine, with a Playwright-style API. Each new_page is a fresh browser context on a pooled Chrome. The engine queues jobs when every slot is busy, restarts browsers before they bloat, and replaces browsers that crash.

Docs: https://fluxwright.vercel.app

Requires a local Chrome/Chromium binary (FLUXWRIGHT_CHROMIUM, CHROME, CHROMIUM, or a standard install path). For headless jobs, chrome-headless-shell is picked up from PATH or Puppeteer's/Playwright's cache and opens pages much faster: npx @puppeteer/browsers install chrome-headless-shell@stable.

Install

pip install fluxwright

Wheels cover Windows x64, macOS (Intel and Apple Silicon) and Linux x86_64, for Python 3.10 and newer. Anywhere else, pip builds from source, which needs Rust.

Use

asyncio:

import asyncio
from fluxwright.async_api import async_playwright

async def main():
    async with async_playwright() as p:
        browser = await p.chromium.launch(max_browsers=4)
        page = await browser.new_page()
        await page.goto("https://example.com")
        print(await page.get_by_role("heading").text_content())  # "Example Domain"
        print(await page.evaluate("() => location.hostname"))
        png = await page.screenshot(full_page=True)  # bytes
        await page.close()
        await browser.close()

asyncio.run(main())

Blocking code:

from fluxwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch()
    page = browser.new_page()
    page.goto("https://example.com")
    print(page.title())
    browser.close()

The names are Playwright's, so a Playwright script for the features below runs after changing playwright.async_api to fluxwright.async_api (or sync_api). chromium can also be imported directly: from fluxwright.async_api import chromium.

Many jobs at once: start them all and let the engine queue them.

async def job(browser, url):
    page = await browser.new_page()
    try:
        await page.goto(url)
        return await page.title()
    finally:
        await page.close()

titles = await asyncio.gather(*(job(browser, url) for url in urls))

The sync API runs every call on one event loop in a background thread, so its objects work from any thread: a ThreadPoolExecutor of jobs sharing one browser is fine (Playwright's sync API does not allow that).

API

Timeouts are in milliseconds, as in Playwright. Errors are fluxwright.Error, and fluxwright.TimeoutError when something ran out of time. Close pages when a job is done; a page you forget is cleaned up when Python collects it.

In the sync API, route handlers run on worker threads and event listeners one at a time on a thread of their own, so both can call the page.

Develop

pip install maturin pytest
maturin develop   # needs Rust
pytest            # against a real Chrome

License

Licensed under the Apache License, Version 2.0.