FastAPIでPlaywrightを使う方法

はじめに

何も意識せずにFastAPIでPlaywright使用すると、「It looks like you are using Playwright Sync API inside the asyncio loop. Please use the Async API instead.」というエラーが発生します。
FastAPIのような非同期 (Async) な実行環境の中で、Playwrightの同期 (Sync) APIを使おうとしているためです。
FastAPIはデフォルトでasyncioという非同期で動作します。同期的な処理を実行すると、その処理が完了するまでイベントループ全体をブロックしてしまい、他のリクエスト処理が停止するため、このエラーが発生します。
そこで、Playwrightの非同期API（playwright.async_api）を使用します。

ソース

main.py

from fastapi import FastAPI
from playwright.async_api import async_playwright

app = FastAPI()

@app.get("/scrape/")
async def run_scrape(url) -> dict[str, str]:

    # 非同期コンテキストマネージャを使用
    async with async_playwright() as p:
        # ブラウザの起動（ヘッドレスモード）
        browser = await p.chromium.launch(headless=True)
        # 新しいページを開く
        page = await browser.new_page()

        # ページナビゲーション
        await page.goto(url, wait_until="networkidle", timeout=10000)

        # コンテンツの取得
        title = await page.title()
        content_element = await page.query_selector("body")
        content = await content_element.inner_text() if content_element else "コンテンツが見つかりませんでした"

        # ブラウザを閉じる
        await browser.close()

        return {
            "url": url,
            "title": title,
            "content_snippet": content[:200] + "..."
        }

使い方

・サーバ起動。「--reload」をつけるとエラーになるので注意。
uvicorn main:app

・URLを指定してコマンドラインで実行
curl -X GET "http://127.0.0.1:8000/scrape/?url=https://www.google.com"

おまけ

main.py

from fastapi import FastAPI
from playwright.async_api import async_playwright
from pathlib import Path

app = FastAPI()

@app.get("/scrape/")
async def run_scrape(url) -> dict[str, str]:

    # 非同期コンテキストマネージャを使用
    async with async_playwright() as p:
        # ブラウザの起動（ヘッドレスモード）
        browser = await p.chromium.launch(headless=True)
        # 新しいページを開く
        page = await browser.new_page()

        # ページナビゲーション
        await page.goto(url, wait_until="networkidle", timeout=10000)

        # ダウンロードボタンをクリック
        download_button = page.locator(".開発者ツールでボタンのクラス名を調べてここに記入")
        async with page.expect_download() as download_info:
            await download_button.click()

        # ダウンロードしたファイルを保存
        download = await download_info.value
        path = Path("保存ファイル名.pdf").resolve()
        await download.save_as(path)

        # ブラウザを閉じる
        await browser.close()

        return {
            "status": "success",
            "message": f"PDFが正常にダウンロードされ、'{path}'に保存されました。"
        }