Playwright Python同步客户端为何切换至异步上下文？

我用Playwright Python库写爬虫，预期全程在同步上下文运行，但调用sync_playwright()后，异步事件循环被启动了，导致执行Django ORM操作时抛出SynchronousOnlyOperation错误。

原始爬虫代码 #

from contextlib import contextmanager
from playwright.sync_api import sync_playwright
import asyncio

def is_async():
    return asyncio.get_event_loop().is_running()

class BaseScrapper(object):
    
    @property
    @contextmanager
    def playwright(self):
        print(f'BaseScrapper.playwright - is_async={is_async()}')
        with sync_playwright() as p:
            print(f'BaseScrapper.playwright with - is_async={is_async()}')
            yield p
    
    @property
    @contextmanager
    def browser(self):
        print(f'BaseScrapper.browser - is_async={is_async()}')
        with self.playwright as p:
            print(f'BaseScrapper.browser with - is_async={is_async()}')
            yield p.chromium.launch(headless=True)
    
    @contextmanager
    def open_page(self, url):
        print(f'BaseScrapper.open_page - is_async={is_async()}')
        with self.browser as browser:
            print(f'BaseScrapper.open_page.with - is_async={is_async()}')
            new_page = browser.new_page()
            
            # 绑定响应监听器
            new_page.on("response", self.intercept_response)
            
            new_page.goto(url, wait_until="domcontentloaded")
            yield new_page
    
    def intercept_response(self, response):
        pass

class ScrapeTest(BaseScrapper):
    @contextmanager
    def run(self):
        print(f'ScrapeTest.run - is_async={is_async()}')
        with self.open_page(url='www.google.com') as page:
            print(f'ScrapeTest.run - is_async={is_async()}')
            yield page

def run():
    print(f'running ... is_async={is_async()}')
    s = ScrapeTest()
    with s.run() as p:
        print(f'run.with is_async={is_async()}')

运行输出 #

>>> run()
running ... is_async=False
ScrapeTest.run - is_async=False
BaseScrapper.open_page - is_async=False
BaseScrapper.browser - is_async=False
BaseScrapper.playwright - is_async=False
BaseScrapper.playwright with - is_async=True
BaseScrapper.browser with - is_async=True
BaseScrapper.open_page.with - is_async=True

触发的Django错误 #

执行Django ORM调用时抛出：

django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.

简化测试代码 #

为了定位问题，我写了更简化的测试代码：

from playwright.sync_api import sync_playwright
import asyncio

class BaseScrapper(object):
    @property
    def playwright(self):
        print('basescrapper.playwright')
        return sync_playwright().start()
    
    @property
    def browser(self):
        print('basescrapper.browser')
        return self.playwright.chromium.launch(headless=True)
    
    def open_page(self, url):
        print('basescrapper.open page')
        page = self.browser.new_page()
        page.goto(url)
        return page

class SampleScrapper(BaseScrapper):
    def run(self):
        print(f'before - {"async" if asyncio.get_event_loop().is_running() else "sync"}')
        page = self.open_page(url='https://www.google.com')
        print(f'after - {"async" if asyncio.get_event_loop().is_running() else "sync"}')

def run():
    print(f'run 1 - {"async" if asyncio.get_event_loop().is_running() else "sync"}')
    s = SampleScrapper()
    print(f'run 2 - {"async" if asyncio.get_event_loop().is_running() else "sync"}')
    s.run()
    print(f'run 3 - {"async" if asyncio.get_event_loop().is_running() else "sync"}')

简化代码运行输出 #

>>> run()
run 1 - sync
run 2 - sync
before - sync
basescrapper.open page
basescrapper.browser
basescrapper.playwright
after - async
run 3 - async

问题原因 #

Playwright的同步API（sync_playwright）底层基于异步实现，它会在后台启动一个asyncio事件循环来驱动浏览器操作。虽然对外暴露的是同步接口，但调用sync_playwright().start()或进入其上下文管理器时，会自动创建并运行异步事件循环，这就是is_async()从False变为True的原因。

Django的ORM操作被标记为仅允许在同步上下文执行，它通过检查当前是否有asyncio事件循环在运行来判断上下文类型。只要事件循环存在，无论是否真的在异步函数中执行，都会触发该错误。

解决方法 #

方法1：在独立线程中执行Playwright操作 #

把爬虫逻辑放到单独线程中运行，线程内的事件循环不会影响主线程的同步上下文：

import threading
from django.db import models

def scrape_in_thread():
    s = ScrapeTest()
    with s.run() as page:
        # 执行爬虫操作
        pass

def run_with_thread():
    print(f'主线程上下文: {"async" if asyncio.get_event_loop().is_running() else "sync"}')
    thread = threading.Thread(target=scrape_in_thread)
    thread.start()
    thread.join()
    # 主线程仍为同步上下文，可安全执行ORM操作
    obj = models.MyModel.objects.first()

方法2：用Django的sync_to_async包装ORM操作 #

如果必须在Playwright启动后的上下文里执行ORM操作，用Django提供的sync_to_async将同步ORM调用转为异步兼容：

from asgiref.sync import sync_to_async

class BaseScrapper(object):
    def intercept_response(self, response):
        # 用sync_to_async包装ORM操作
        async def get_db_data():
            return await sync_to_async(models.MyModel.objects.first)()
        
        # 在Playwright回调中执行异步任务
        asyncio.create_task(get_db_data())

方法3：手动关闭事件循环（不推荐） #

可以在Playwright使用完毕后手动关闭事件循环，但这种方式可能引发线程安全问题，不建议生产环境使用：

def run():
    s = SampleScrapper()
    s.run()
    # 关闭事件循环
    loop = asyncio.get_event_loop()
    loop.close()
    print(f'关闭后上下文: {"async" if asyncio.get_event_loop().is_running() else "sync"}')

内容的提问来源于stack exchange，提问作者user28275621