diff --git a/scrapegraphai/docloaders/browser_base.py b/scrapegraphai/docloaders/browser_base.py index 50c6cd18..e1e3c494 100644 --- a/scrapegraphai/docloaders/browser_base.py +++ b/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)) diff --git a/scrapegraphai/docloaders/chromium.py b/scrapegraphai/docloaders/chromium.py index a522582c..c6a30a15 100644 --- a/scrapegraphai/docloaders/chromium.py +++ b/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) diff --git a/scrapegraphai/nodes/generate_answer_from_image_node.py b/scrapegraphai/nodes/generate_answer_from_image_node.py index 808804fd..8a07bb0f 100644 --- a/scrapegraphai/nodes/generate_answer_from_image_node.py +++ b/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)) diff --git a/scrapegraphai/nodes/graph_iterator_node.py b/scrapegraphai/nodes/graph_iterator_node.py index 82171258..0a634585 100644 --- a/scrapegraphai/nodes/graph_iterator_node.py +++ b/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 diff --git a/scrapegraphai/utils/__init__.py b/scrapegraphai/utils/__init__.py index 3a4e7b13..f013dfe7 100644 --- a/scrapegraphai/utils/__init__.py +++ b/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", diff --git a/scrapegraphai/utils/event_loop.py b/scrapegraphai/utils/event_loop.py new file mode 100644 index 00000000..c052e781 --- /dev/null +++ b/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()