fix: allow running graphs inside an existing event loop (Jupyter, FastAPI)

scrapegraphai/docloaders/browser_base.py

@@ -5,6 +5,8 @@
 import asyncio
 from typing import List
 
+from ..utils.event_loop import run_coroutine_sync
+
 
 def browser_base_fetch(
     api_key: str,
@@ -53,7 +55,7 @@ async def _async_browser_base_fetch():
                 result.append(await _async_fetch_link(url))
             return result
 
-        result = asyncio.run(_async_browser_base_fetch())
+        result = run_coroutine_sync(_async_browser_base_fetch())
     else:
         for url in link:
             result.append(session.load(url, text_content=text_content))

scrapegraphai/docloaders/chromium.py

@@ -6,6 +6,7 @@
 from langchain_core.documents import Document
 
 from ..utils import Proxy, dynamic_import, get_logger, parse_or_search_proxy
+from ..utils.event_loop import run_coroutine_sync
 
 logger = get_logger("web-loader")
 
@@ -460,7 +461,7 @@ def lazy_load(self) -> Iterator[Document]:
         )
 
         for url in self.urls:
-            html_content = asyncio.run(scraping_fn(url))
+            html_content = run_coroutine_sync(scraping_fn(url))
             metadata = {"source": url}
             yield Document(page_content=html_content, metadata=metadata)
 

scrapegraphai/nodes/generate_answer_from_image_node.py

@@ -8,6 +8,7 @@
 
 import aiohttp
 
+from ..utils.event_loop import run_coroutine_sync
 from .base_node import BaseNode
 
 
@@ -113,15 +114,4 @@ def execute(self, state: dict) -> dict:
         """
         Wrapper to run the asynchronous execute_async function in a synchronous context.
         """
-        try:
-            eventloop = asyncio.get_event_loop()
-        except RuntimeError:
-            eventloop = None
-
-        if eventloop and eventloop.is_running():
-            task = eventloop.create_task(self.execute_async(state))
-            state = eventloop.run_until_complete(asyncio.gather(task))[0]
-        else:
-            state = asyncio.run(self.execute_async(state))
-
-        return state
+        return run_coroutine_sync(self.execute_async(state))

scrapegraphai/nodes/graph_iterator_node.py

@@ -8,6 +8,7 @@
 from pydantic import BaseModel
 from tqdm.asyncio import tqdm
 
+from ..utils.event_loop import run_coroutine_sync
 from .base_node import BaseNode
 
 DEFAULT_BATCHSIZE = 16
@@ -66,15 +67,7 @@ def execute(self, state: dict) -> dict:
             f"--- Executing {self.node_name} Node with batchsize {batchsize} ---"
         )
 
-        try:
-            eventloop = asyncio.get_event_loop()
-        except RuntimeError:
-            eventloop = None
-
-        if eventloop and eventloop.is_running():
-            state = eventloop.run_until_complete(self._async_execute(state, batchsize))
-        else:
-            state = asyncio.run(self._async_execute(state, batchsize))
+        state = run_coroutine_sync(self._async_execute(state, batchsize))
 
         return state
 

scrapegraphai/utils/__init__.py

@@ -19,6 +19,7 @@
 from .convert_to_md import convert_to_md
 from .data_export import export_to_csv, export_to_json, export_to_xml
 from .dict_content_compare import are_content_equal
+from .event_loop import run_coroutine_sync
 from .llm_callback_manager import CustomLLMCallbackManager
 from .logging import (
     get_logger,
@@ -86,6 +87,7 @@
     "dynamic_import",
     "srcfile_import",
     "num_tokens_calculus",
+    "run_coroutine_sync",
     # Proxy handling
     "Proxy",
     "parse_or_search_proxy",

scrapegraphai/utils/event_loop.py

@@ -0,0 +1,31 @@
+"""
+Event loop helper module
+"""
+
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+
+
+def run_coroutine_sync(coro):
+    """
+    Runs a coroutine from synchronous code.
+
+    Uses asyncio.run() when the current thread has no running event loop.
+    When called from code that already has one (Jupyter, FastAPI, and
+    similar environments), the coroutine runs on its own loop in a worker
+    thread, since asyncio.run() and loop.run_until_complete() both raise
+    RuntimeError in that case.
+
+    Args:
+        coro: The coroutine to execute.
+
+    Returns:
+        The value returned by the coroutine.
+    """
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.run(coro)
+
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        return executor.submit(asyncio.run, coro).result()