skills/d4vinci__scrapling
1
0
Fork 0
mirror of https://github.com/D4Vinci/Scrapling.git synced 2026-09-14 20:07:02 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(fetchers): keep browser tabs open and reuse them across requests

Browse Source 
Browser sessions no longer close a tab after its request is completed. The tab returns to the pool as ready, and the next request reuses it, reapplying its own timeouts, extra headers, and resource/domain routes (after `unroute_all`) so nothing leaks between requests. Tabs that hit an error or got closed by the browser are closed and evicted, and the internal response listener is detached after every request so reused tabs don't stack handlers.

Adds `close_pages()` to all browser sessions to close every open tab; the next request opens a fresh one. `PagePool` gains `get_ready_page`/`remove_page`/`clear` and `PageInfo.mark_ready`, new pages start busy, and the unused `cleanup_error_pages` is removed.

Proxy-rotation contexts keep closing per request. Docs and the agent skill describe the new lifecycle.
This commit is contained in:
Karim shoair
2026-08-23 20:28:30 +03:00
parent 6c5b691db7
commit f164713efa
14 changed files with 632 additions and 308 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent-skill/Scrapling-Skill.zip
BIN
View File
Binary file not shown.
agent-skill/Scrapling-Skill/references/fetching/dynamic.md
+9 -4
View File
@@ -325,14 +325,19 @@ async def scrape_multiple_sites():
return pages
```
You may have noticed the `max_pages` argument. This is a new argument that enables the fetcher to create a **rotating pool of Browser tabs**. Instead of using a single tab for all your requests, you set a limit on the maximum number of pages that can be displayed at once. With each request, the library will close all tabs that have finished their task and check if the number of the current tabs is lower than the maximum allowed number of pages/tabs, then:
You may have noticed the `max_pages` argument. It enables the fetcher to keep a **pool of Browser tabs**, and you set the maximum number of tabs that can be open at once. Tabs stay open after their request finishes, so with each request, the library will:
1. If you are within the allowed range, the fetcher will create a new tab for you, and then all is as normal.
2. Otherwise, it will keep checking every subsecond if creating a new tab is allowed or not for 60 seconds, then raise `TimeoutError`. This can happen when the website you are fetching becomes unresponsive.
1. Reuse a free tab if there's one. Every request applies its own tab-level settings (`timeout`, `extra_headers`, `disable_resources`, `blocked_domains`, etc.) to the tab it gets, so nothing leaks from the previous request.
2. Otherwise, open a new tab if the number of open tabs is lower than `max_pages`.
3. Otherwise, keep checking every subsecond for a tab to become free for 60 seconds, then raise `TimeoutError`. This can happen when the website you are fetching becomes unresponsive.
Tabs that hit an error are closed and replaced, and you can close all the open tabs yourself at any point with `session.close_pages()`, then the next request opens a fresh one.
This logic allows for multiple URLs to be fetched at the same time in the same browser, which saves a lot of resources, but most importantly, is so fast :)
In versions 0.3 and 0.3.1, the pool was reusing finished tabs to save more resources/time. That logic proved flawed, as it's nearly impossible to protect pages/tabs from contamination by the previous configuration used in the request before this one.
Keeping the tabs open also means the page you fetched is still there for the next request, so a `page_setup` function on the next request runs on it before navigating away. That's the building block for chaining automation across requests.
Versions 0.3.2 to 0.4.14 closed every tab after its request because reusing tabs used to leak settings between requests. Since 0.4.15, the settings are reset on every reuse, so the tabs stay open.
### Session Benefits
agent-skill/Scrapling-Skill/references/fetching/stealthy.md
+9 -4
View File
@@ -228,14 +228,19 @@ async def scrape_multiple_sites():
return pages
```
You may have noticed the `max_pages` argument. This is a new argument that enables the fetcher to create a **rotating pool of Browser tabs**. Instead of using a single tab for all your requests, you set a limit on the maximum number of pages that can be displayed at once. With each request, the library will close all tabs that have finished their task and check if the number of the current tabs is lower than the maximum allowed number of pages/tabs, then:
You may have noticed the `max_pages` argument. It enables the fetcher to keep a **pool of Browser tabs**, and you set the maximum number of tabs that can be open at once. Tabs stay open after their request finishes, so with each request, the library will:
1. If you are within the allowed range, the fetcher will create a new tab for you, and then all is as normal.
2. Otherwise, it will keep checking every subsecond if creating a new tab is allowed or not for 60 seconds, then raise `TimeoutError`. This can happen when the website you are fetching becomes unresponsive.
1. Reuse a free tab if there's one. Every request applies its own tab-level settings (`timeout`, `extra_headers`, `disable_resources`, `blocked_domains`, etc.) to the tab it gets, so nothing leaks from the previous request.
2. Otherwise, open a new tab if the number of open tabs is lower than `max_pages`.
3. Otherwise, keep checking every subsecond for a tab to become free for 60 seconds, then raise `TimeoutError`. This can happen when the website you are fetching becomes unresponsive.
Tabs that hit an error are closed and replaced, and you can close all the open tabs yourself at any point with `session.close_pages()`, then the next request opens a fresh one.
This logic allows for multiple URLs to be fetched at the same time in the same browser, which saves a lot of resources, but most importantly, is so fast :)
In versions 0.3 and 0.3.1, the pool was reusing finished tabs to save more resources/time. That logic proved flawed, as it's nearly impossible to protect pages/tabs from contamination by the previous configuration used in the request before this one.
Keeping the tabs open also means the page you fetched is still there for the next request, so a `page_setup` function on the next request runs on it before navigating away. That's the building block for chaining automation across requests.
Versions 0.3.2 to 0.4.14 closed every tab after its request because reusing tabs used to leak settings between requests. Since 0.4.15, the settings are reset on every reuse, so the tabs stay open.
### Session Benefits
docs/fetching/dynamic.md
+9 -4
View File
@@ -151,14 +151,19 @@ async def scrape_multiple_sites():
return pages
```
You may have noticed the `max_pages` argument. This is a new argument that enables the fetcher to create a **rotating pool of Browser tabs**. Instead of using a single tab for all your requests, you set a limit on the maximum number of pages that can be displayed at once. With each request, the library will close all tabs that have finished their task and check if the number of the current tabs is lower than the maximum allowed number of pages/tabs, then:
You may have noticed the `max_pages` argument. It enables the fetcher to keep a **pool of Browser tabs**, and you set the maximum number of tabs that can be open at once. Tabs stay open after their request finishes, so with each request, the library will:
1. If you are within the allowed range, the fetcher will create a new tab for you, and then all is as normal.
2. Otherwise, it will keep checking every subsecond if creating a new tab is allowed or not for 60 seconds, then raise `TimeoutError`. This can happen when the website you are fetching becomes unresponsive.
1. Reuse a free tab if there's one. Every request applies its own tab-level settings (`timeout`, `extra_headers`, `disable_resources`, `blocked_domains`, etc.) to the tab it gets, so nothing leaks from the previous request.
2. Otherwise, open a new tab if the number of open tabs is lower than `max_pages`.
3. Otherwise, keep checking every subsecond for a tab to become free for 60 seconds, then raise `TimeoutError`. This can happen when the website you are fetching becomes unresponsive.
Tabs that hit an error are closed and replaced, and you can close all the open tabs yourself at any point with `session.close_pages()`, then the next request opens a fresh one.
This logic allows for multiple URLs to be fetched at the same time in the same browser, which saves a lot of resources, but most importantly, is so fast :)
In versions 0.3 and 0.3.1, the pool was reusing finished tabs to save more resources/time. That logic proved flawed, as it's nearly impossible to protect pages/tabs from contamination by the previous configuration used in the request before this one.
Keeping the tabs open also means the page you fetched is still there for the next request, so a `page_setup` function on the next request runs on it before navigating away. That's the building block for chaining automation across requests.
Versions 0.3.2 to 0.4.14 closed every tab after its request because reusing tabs used to leak settings between requests. Since 0.4.15, the settings are reset on every reuse, so the tabs stay open.
### Session Benefits
docs/fetching/stealthy.md
+9 -4
View File
@@ -240,14 +240,19 @@ async def scrape_multiple_sites():
return pages
```
You may have noticed the `max_pages` argument. This is a new argument that enables the fetcher to create a **rotating pool of Browser tabs**. Instead of using a single tab for all your requests, you set a limit on the maximum number of pages that can be displayed at once. With each request, the library will close all tabs that have finished their task and check if the number of the current tabs is lower than the maximum allowed number of pages/tabs, then:
You may have noticed the `max_pages` argument. It enables the fetcher to keep a **pool of Browser tabs**, and you set the maximum number of tabs that can be open at once. Tabs stay open after their request finishes, so with each request, the library will:
1. If you are within the allowed range, the fetcher will create a new tab for you, and then all is as normal.
2. Otherwise, it will keep checking every subsecond if creating a new tab is allowed or not for 60 seconds, then raise `TimeoutError`. This can happen when the website you are fetching becomes unresponsive.
1. Reuse a free tab if there's one. Every request applies its own tab-level settings (`timeout`, `extra_headers`, `disable_resources`, `blocked_domains`, etc.) to the tab it gets, so nothing leaks from the previous request.
2. Otherwise, open a new tab if the number of open tabs is lower than `max_pages`.
3. Otherwise, keep checking every subsecond for a tab to become free for 60 seconds, then raise `TimeoutError`. This can happen when the website you are fetching becomes unresponsive.
Tabs that hit an error are closed and replaced, and you can close all the open tabs yourself at any point with `session.close_pages()`, then the next request opens a fresh one.
This logic allows for multiple URLs to be fetched at the same time in the same browser, which saves a lot of resources, but most importantly, is so fast :)
In versions 0.3 and 0.3.1, the pool was reusing finished tabs to save more resources/time. That logic proved flawed, as it's nearly impossible to protect pages/tabs from contamination by the previous configuration used in the request before this one.
Keeping the tabs open also means the page you fetched is still there for the next request, so a `page_setup` function on the next request runs on it before navigating away. That's the building block for chaining automation across requests.
Versions 0.3.2 to 0.4.14 closed every tab after its request because reusing tabs used to leak settings between requests. Since 0.4.15, the settings are reset on every reuse, so the tabs stay open.
### Session Benefits
scrapling/engines/_browsers/_base.py
+55 -31
View File
@@ -1,7 +1,7 @@
from time import time
from re import search as re_search
from asyncio import sleep as asyncio_sleep, Lock
from contextlib import contextmanager, asynccontextmanager
from contextlib import contextmanager, asynccontextmanager, suppress
from playwright.sync_api._generated import Page
from playwright.sync_api import (
@@ -63,11 +63,18 @@ class SyncSession:
def start(self) -> None:
pass
def close_pages(self) -> None:
"""Close every open tab in the session's pool. The next request opens a fresh tab."""
for page_info in self.page_pool.clear():
with suppress(Exception):
page_info.page.close()
def close(self): # pragma: no cover
"""Close all resources"""
if not self._is_alive:
return
self.close_pages()
if self.context:
self.context.close()
self.context = None
@@ -107,21 +114,21 @@ class SyncSession:
blocked_domains: Optional[Set[str]] = None,
context: Optional[BrowserContext] = None,
) -> PageInfo[Page]: # pragma: no cover
"""Get a new page to use"""
# No need to check if a page is available or not in sync code because the code blocked before reaching here till the page closed, ofc.
ctx = context if context is not None else self.context
assert ctx is not None, "Browser context not initialized"
page = ctx.new_page()
"""Get a ready page from the pool, or open a new one"""
page_info = self.page_pool.get_ready_page() if context is None else None
if page_info is None:
ctx = context if context is not None else self.context
assert ctx is not None, "Browser context not initialized"
page_info = self.page_pool.add_page(ctx.new_page())
page = cast(Page, page_info.page)
page.set_default_navigation_timeout(timeout)
page.set_default_timeout(timeout)
if extra_headers:
page.set_extra_http_headers(extra_headers)
page.set_extra_http_headers(extra_headers or {})
page.unroute_all(behavior="ignoreErrors")
if disable_resources or blocked_domains:
page.route("**/*", create_intercept_handler(disable_resources, blocked_domains))
page_info = self.page_pool.add_page(page)
page_info.mark_busy()
return page_info
return cast(PageInfo[Page], page_info)
def get_pool_stats(self) -> Dict[str, int]:
"""Get statistics about the current page pool"""
@@ -202,8 +209,8 @@ class SyncSession:
page_info = self._get_page(timeout, extra_headers, disable_resources, blocked_domains, context=context)
yield page_info
finally:
if page_info is not None and page_info in self.page_pool.pages:
self.page_pool.pages.remove(page_info)
if page_info is not None:
self.page_pool.remove_page(page_info)
context.close()
else:
# Standard mode: use PagePool with persistent context
@@ -211,8 +218,12 @@ class SyncSession:
try:
yield page_info
finally:
page_info.page.close()
self.page_pool.pages.remove(page_info)
if page_info.state == "error" or page_info.page.is_closed():
with suppress(Exception):
page_info.page.close()
self.page_pool.remove_page(page_info)
else:
page_info.mark_ready()
class AsyncSession:
@@ -234,11 +245,18 @@ class AsyncSession:
async def start(self) -> None:
pass
async def close_pages(self) -> None:
"""Close every open tab in the session's pool. The next request opens a fresh tab."""
for page_info in self.page_pool.clear():
with suppress(Exception):
await cast(AsyncPage, page_info.page).close()
async def close(self):
"""Close all resources"""
if not self._is_alive: # pragma: no cover
return
await self.close_pages()
if self.context:
await self.context.close()
self.context = None # pyright: ignore
@@ -280,35 +298,37 @@ class AsyncSession:
blocked_domains: Optional[Set[str]] = None,
context: Optional[AsyncBrowserContext] = None,
) -> PageInfo[AsyncPage]: # pragma: no cover
"""Get a new page to use"""
"""Get a ready page from the pool, or open a new one"""
ctx = context if context is not None else self.context
if TYPE_CHECKING:
assert ctx is not None, "Browser context not initialized"
async with self._lock:
# If we're at max capacity after cleanup, wait for busy pages to finish
if context is None and self.page_pool.pages_count >= self.max_pages:
# Only applies when using persistent context
page_info = self.page_pool.get_ready_page() if context is None else None
if page_info is None and context is None and self.page_pool.pages_count >= self.max_pages:
# At max capacity with the persistent context, so wait for a busy page to become ready
start_time = time()
while time() - start_time < self._max_wait_for_page:
await asyncio_sleep(0.05)
if self.page_pool.pages_count < self.max_pages:
page_info = self.page_pool.get_ready_page()
if page_info is not None:
break
else:
raise TimeoutError(
f"No pages finished to clear place in the pool within the {self._max_wait_for_page}s timeout period"
)
page = await ctx.new_page()
if page_info is None:
page_info = self.page_pool.add_page(await ctx.new_page())
page = cast(AsyncPage, page_info.page)
page.set_default_navigation_timeout(timeout)
page.set_default_timeout(timeout)
if extra_headers:
await page.set_extra_http_headers(extra_headers)
await page.set_extra_http_headers(extra_headers or {})
await page.unroute_all(behavior="ignoreErrors")
if disable_resources or blocked_domains:
await page.route("**/*", create_async_intercept_handler(disable_resources, blocked_domains))
return self.page_pool.add_page(page)
return cast(PageInfo[AsyncPage], page_info)
def get_pool_stats(self) -> Dict[str, int]:
"""Get statistics about the current page pool"""
@@ -391,8 +411,8 @@ class AsyncSession:
)
yield page_info
finally:
if page_info is not None and page_info in self.page_pool.pages:
self.page_pool.pages.remove(page_info)
if page_info is not None:
self.page_pool.remove_page(page_info)
await context.close()
else:
# Standard mode: use PagePool with persistent context
@@ -400,8 +420,12 @@ class AsyncSession:
try:
yield page_info
finally:
await page_info.page.close()
self.page_pool.pages.remove(page_info)
if page_info.state == "error" or page_info.page.is_closed():
with suppress(Exception):
await page_info.page.close()
self.page_pool.remove_page(page_info)
else:
page_info.mark_ready()
class BaseSessionMixin:
scrapling/engines/_browsers/_controllers.py
+110 -110
View File
@@ -144,70 +144,70 @@ class DynamicSession(SyncSession, DynamicSessionMixin):
final_response: List = [None]
xhr_captured: List = []
page = page_info.page
page.on(
"response",
self._create_response_handler(
page_info,
final_response,
xhr_pattern=self._config.capture_xhr,
xhr_container=xhr_captured,
),
handler = self._create_response_handler(
page_info,
final_response,
xhr_pattern=self._config.capture_xhr,
xhr_container=xhr_captured,
)
if params.page_setup:
try:
params.page_setup(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_setup: {e}")
page.on("response", handler)
try:
first_response = page.goto(url, referer=referer)
self._wait_for_page_stability(page, params.load_dom, params.network_idle)
if not first_response:
raise RuntimeError(f"Failed to get response for {url}")
if params.page_action:
if params.page_setup:
try:
_ = params.page_action(page)
params.page_setup(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_action: {e}")
log.error(f"Error executing page_setup: {e}")
if params.wait_selector:
try:
waiter: Locator = page.locator(params.wait_selector)
waiter.first.wait_for(state=params.wait_selector_state)
self._wait_for_page_stability(page, params.load_dom, params.network_idle)
except Exception as e: # pragma: no cover
log.error(f"Error waiting for selector {params.wait_selector}: {e}")
try:
first_response = page.goto(url, referer=referer)
self._wait_for_page_stability(page, params.load_dom, params.network_idle)
page.wait_for_timeout(params.wait)
if not first_response:
raise RuntimeError(f"Failed to get response for {url}")
response = ResponseFactory.from_playwright_response(
page,
first_response,
final_response[0],
params.selector_config,
meta={"proxy": proxy},
xhr_captured=xhr_captured,
)
return response
if params.page_action:
try:
_ = params.page_action(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_action: {e}")
except Exception as e:
page_info.mark_error()
if attempt < self._config.retries - 1:
if is_proxy_error(e):
log.warning(
f"Proxy '{proxy}' failed (attempt {attempt + 1}) | Retrying in {self._config.retry_delay}s..."
)
if params.wait_selector:
try:
waiter: Locator = page.locator(params.wait_selector)
waiter.first.wait_for(state=params.wait_selector_state)
self._wait_for_page_stability(page, params.load_dom, params.network_idle)
except Exception as e: # pragma: no cover
log.error(f"Error waiting for selector {params.wait_selector}: {e}")
page.wait_for_timeout(params.wait)
response = ResponseFactory.from_playwright_response(
page,
first_response,
final_response[0],
params.selector_config,
meta={"proxy": proxy},
xhr_captured=xhr_captured,
)
return response
except Exception as e:
page_info.mark_error()
if attempt < self._config.retries - 1:
if is_proxy_error(e):
log.warning(
f"Proxy '{proxy}' failed (attempt {attempt + 1}) | Retrying in {self._config.retry_delay}s..."
)
else:
log.warning(
f"Attempt {attempt + 1} failed: {e}. Retrying in {self._config.retry_delay}s..."
)
time_sleep(self._config.retry_delay)
else:
log.warning(
f"Attempt {attempt + 1} failed: {e}. Retrying in {self._config.retry_delay}s..."
)
time_sleep(self._config.retry_delay)
else:
log.error(f"Failed after {self._config.retries} attempts: {e}")
raise
log.error(f"Failed after {self._config.retries} attempts: {e}")
raise
finally:
page.remove_listener("response", handler)
raise RuntimeError("Request failed") # pragma: no cover
@@ -333,69 +333,69 @@ class AsyncDynamicSession(AsyncSession, DynamicSessionMixin):
final_response: List = [None]
xhr_captured: List = []
page = page_info.page
page.on(
"response",
self._create_response_handler(
page_info,
final_response,
xhr_pattern=self._config.capture_xhr,
xhr_container=xhr_captured,
),
handler = self._create_response_handler(
page_info,
final_response,
xhr_pattern=self._config.capture_xhr,
xhr_container=xhr_captured,
)
if params.page_setup:
try:
await params.page_setup(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_setup: {e}")
page.on("response", handler)
try:
first_response = await page.goto(url, referer=referer)
await self._wait_for_page_stability(page, params.load_dom, params.network_idle)
if not first_response:
raise RuntimeError(f"Failed to get response for {url}")
if params.page_action:
if params.page_setup:
try:
_ = await params.page_action(page)
await params.page_setup(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_action: {e}")
log.error(f"Error executing page_setup: {e}")
if params.wait_selector:
try:
waiter: AsyncLocator = page.locator(params.wait_selector)
await waiter.first.wait_for(state=params.wait_selector_state)
await self._wait_for_page_stability(page, params.load_dom, params.network_idle)
except Exception as e: # pragma: no cover
log.error(f"Error waiting for selector {params.wait_selector}: {e}")
try:
first_response = await page.goto(url, referer=referer)
await self._wait_for_page_stability(page, params.load_dom, params.network_idle)
await page.wait_for_timeout(params.wait)
if not first_response:
raise RuntimeError(f"Failed to get response for {url}")
response = await ResponseFactory.from_async_playwright_response(
page,
first_response,
final_response[0],
params.selector_config,
meta={"proxy": proxy},
xhr_captured=xhr_captured,
)
return response
if params.page_action:
try:
_ = await params.page_action(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_action: {e}")
except Exception as e:
page_info.mark_error()
if attempt < self._config.retries - 1:
if is_proxy_error(e):
log.warning(
f"Proxy '{proxy}' failed (attempt {attempt + 1}) | Retrying in {self._config.retry_delay}s..."
)
if params.wait_selector:
try:
waiter: AsyncLocator = page.locator(params.wait_selector)
await waiter.first.wait_for(state=params.wait_selector_state)
await self._wait_for_page_stability(page, params.load_dom, params.network_idle)
except Exception as e: # pragma: no cover
log.error(f"Error waiting for selector {params.wait_selector}: {e}")
await page.wait_for_timeout(params.wait)
response = await ResponseFactory.from_async_playwright_response(
page,
first_response,
final_response[0],
params.selector_config,
meta={"proxy": proxy},
xhr_captured=xhr_captured,
)
return response
except Exception as e:
page_info.mark_error()
if attempt < self._config.retries - 1:
if is_proxy_error(e):
log.warning(
f"Proxy '{proxy}' failed (attempt {attempt + 1}) | Retrying in {self._config.retry_delay}s..."
)
else:
log.warning(
f"Attempt {attempt + 1} failed: {e}. Retrying in {self._config.retry_delay}s..."
)
await asyncio_sleep(self._config.retry_delay)
else:
log.warning(
f"Attempt {attempt + 1} failed: {e}. Retrying in {self._config.retry_delay}s..."
)
await asyncio_sleep(self._config.retry_delay)
else:
log.error(f"Failed after {self._config.retries} attempts: {e}")
raise
log.error(f"Failed after {self._config.retries} attempts: {e}")
raise
finally:
page.remove_listener("response", handler)
raise RuntimeError("Request failed") # pragma: no cover
scrapling/engines/_browsers/_page.py
+29 -8
View File
@@ -24,6 +24,11 @@ class PageInfo(Generic[PageType]):
self.state = "busy"
self.url = url
def mark_ready(self):
"""Mark the page as ready to be reused by the next request"""
self.state = "ready"
self.url = ""
def mark_error(self):
"""Mark the page as having an error"""
self.state = "error"
@@ -55,21 +60,42 @@ class PagePool:
def add_page(self, page: AsyncPage) -> PageInfo[AsyncPage]: ...
def add_page(self, page: SyncPage | AsyncPage) -> PageInfo[SyncPage] | PageInfo[AsyncPage]:
"""Add a new page to the pool"""
"""Add a new page to the pool, marked busy for the request that created it"""
with self._lock:
if len(self.pages) >= self.max_pages:
raise RuntimeError(f"Maximum page limit ({self.max_pages}) reached")
if isinstance(page, AsyncPage):
page_info: PageInfo[SyncPage] | PageInfo[AsyncPage] = cast(
PageInfo[AsyncPage], PageInfo(page, "ready", "")
PageInfo[AsyncPage], PageInfo(page, "busy", "")
)
else:
page_info = cast(PageInfo[SyncPage], PageInfo(page, "ready", ""))
page_info = cast(PageInfo[SyncPage], PageInfo(page, "busy", ""))
self.pages.append(page_info)
return page_info
def get_ready_page(self) -> Optional[PageInfo[SyncPage] | PageInfo[AsyncPage]]:
"""Take the first ready page out of the pool's free pages, marking it busy, or return None"""
with self._lock:
for page_info in self.pages:
if page_info.state == "ready":
page_info.mark_busy()
return page_info
return None
def remove_page(self, page_info: PageInfo[SyncPage] | PageInfo[AsyncPage]):
"""Forget a page, whether it's still in the pool or not"""
with self._lock:
if page_info in self.pages:
self.pages.remove(page_info)
def clear(self) -> List[PageInfo[SyncPage] | PageInfo[AsyncPage]]:
"""Forget every page and return them so the caller can close them"""
with self._lock:
pages, self.pages = self.pages, []
return pages
@property
def pages_count(self) -> int:
"""Get the total number of pages"""
@@ -80,8 +106,3 @@ class PagePool:
"""Get the number of busy pages"""
with self._lock:
return sum(1 for p in self.pages if p.state == "busy")
def cleanup_error_pages(self):
"""Remove pages in error state"""
with self._lock:
self.pages = [p for p in self.pages if p.state != "error"]
scrapling/engines/_browsers/_stealth.py
+116 -116
View File
@@ -238,75 +238,75 @@ class StealthySession(SyncSession, StealthySessionMixin):
final_response: List = [None]
xhr_captured: List = []
page = page_info.page
page.on(
"response",
self._create_response_handler(
page_info,
final_response,
xhr_pattern=self._config.capture_xhr,
xhr_container=xhr_captured,
),
handler = self._create_response_handler(
page_info,
final_response,
xhr_pattern=self._config.capture_xhr,
xhr_container=xhr_captured,
)
if params.page_setup:
try:
params.page_setup(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_setup: {e}")
page.on("response", handler)
try:
first_response = page.goto(url, referer=referer)
self._wait_for_page_stability(page, params.load_dom, params.network_idle)
if params.page_setup:
try:
params.page_setup(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_setup: {e}")
if not first_response:
raise RuntimeError(f"Failed to get response for {url}")
if params.solve_cloudflare:
self._cloudflare_solver(page)
# Make sure the page is fully loaded after the captcha
try:
first_response = page.goto(url, referer=referer)
self._wait_for_page_stability(page, params.load_dom, params.network_idle)
if params.page_action:
try:
_ = params.page_action(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_action: {e}")
if not first_response:
raise RuntimeError(f"Failed to get response for {url}")
if params.wait_selector:
try:
waiter: Locator = page.locator(params.wait_selector)
waiter.first.wait_for(state=params.wait_selector_state)
if params.solve_cloudflare:
self._cloudflare_solver(page)
# Make sure the page is fully loaded after the captcha
self._wait_for_page_stability(page, params.load_dom, params.network_idle)
except Exception as e: # pragma: no cover
log.error(f"Error waiting for selector {params.wait_selector}: {e}")
page.wait_for_timeout(params.wait)
if params.page_action:
try:
_ = params.page_action(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_action: {e}")
response = ResponseFactory.from_playwright_response(
page,
first_response,
final_response[0],
params.selector_config,
meta={"proxy": proxy},
xhr_captured=xhr_captured,
)
return response
if params.wait_selector:
try:
waiter: Locator = page.locator(params.wait_selector)
waiter.first.wait_for(state=params.wait_selector_state)
self._wait_for_page_stability(page, params.load_dom, params.network_idle)
except Exception as e: # pragma: no cover
log.error(f"Error waiting for selector {params.wait_selector}: {e}")
except Exception as e:
page_info.mark_error()
if attempt < self._config.retries - 1:
if is_proxy_error(e):
log.warning(
f"Proxy '{proxy}' failed (attempt {attempt + 1}) | Retrying in {self._config.retry_delay}s..."
)
page.wait_for_timeout(params.wait)
response = ResponseFactory.from_playwright_response(
page,
first_response,
final_response[0],
params.selector_config,
meta={"proxy": proxy},
xhr_captured=xhr_captured,
)
return response
except Exception as e:
page_info.mark_error()
if attempt < self._config.retries - 1:
if is_proxy_error(e):
log.warning(
f"Proxy '{proxy}' failed (attempt {attempt + 1}) | Retrying in {self._config.retry_delay}s..."
)
else:
log.warning(
f"Attempt {attempt + 1} failed: {e}. Retrying in {self._config.retry_delay}s..."
)
time_sleep(self._config.retry_delay)
else:
log.warning(
f"Attempt {attempt + 1} failed: {e}. Retrying in {self._config.retry_delay}s..."
)
time_sleep(self._config.retry_delay)
else:
log.error(f"Failed after {self._config.retries} attempts: {e}")
raise
log.error(f"Failed after {self._config.retries} attempts: {e}")
raise
finally:
page.remove_listener("response", handler)
raise RuntimeError("Request failed") # pragma: no cover
@@ -526,74 +526,74 @@ class AsyncStealthySession(AsyncSession, StealthySessionMixin):
final_response: List = [None]
xhr_captured: List = []
page = page_info.page
page.on(
"response",
self._create_response_handler(
page_info,
final_response,
xhr_pattern=self._config.capture_xhr,
xhr_container=xhr_captured,
),
handler = self._create_response_handler(
page_info,
final_response,
xhr_pattern=self._config.capture_xhr,
xhr_container=xhr_captured,
)
if params.page_setup:
try:
await params.page_setup(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_setup: {e}")
page.on("response", handler)
try:
first_response = await page.goto(url, referer=referer)
await self._wait_for_page_stability(page, params.load_dom, params.network_idle)
if params.page_setup:
try:
await params.page_setup(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_setup: {e}")
if not first_response:
raise RuntimeError(f"Failed to get response for {url}")
if params.solve_cloudflare:
await self._cloudflare_solver(page)
# Make sure the page is fully loaded after the captcha
try:
first_response = await page.goto(url, referer=referer)
await self._wait_for_page_stability(page, params.load_dom, params.network_idle)
if params.page_action:
try:
_ = await params.page_action(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_action: {e}")
if not first_response:
raise RuntimeError(f"Failed to get response for {url}")
if params.wait_selector:
try:
waiter: AsyncLocator = page.locator(params.wait_selector)
await waiter.first.wait_for(state=params.wait_selector_state)
if params.solve_cloudflare:
await self._cloudflare_solver(page)
# Make sure the page is fully loaded after the captcha
await self._wait_for_page_stability(page, params.load_dom, params.network_idle)
except Exception as e: # pragma: no cover
log.error(f"Error waiting for selector {params.wait_selector}: {e}")
await page.wait_for_timeout(params.wait)
if params.page_action:
try:
_ = await params.page_action(page)
except Exception as e: # pragma: no cover
log.error(f"Error executing page_action: {e}")
response = await ResponseFactory.from_async_playwright_response(
page,
first_response,
final_response[0],
params.selector_config,
meta={"proxy": proxy},
xhr_captured=xhr_captured,
)
return response
if params.wait_selector:
try:
waiter: AsyncLocator = page.locator(params.wait_selector)
await waiter.first.wait_for(state=params.wait_selector_state)
await self._wait_for_page_stability(page, params.load_dom, params.network_idle)
except Exception as e: # pragma: no cover
log.error(f"Error waiting for selector {params.wait_selector}: {e}")
except Exception as e:
page_info.mark_error()
if attempt < self._config.retries - 1:
if is_proxy_error(e):
log.warning(
f"Proxy '{proxy}' failed (attempt {attempt + 1}) | Retrying in {self._config.retry_delay}s..."
)
await page.wait_for_timeout(params.wait)
response = await ResponseFactory.from_async_playwright_response(
page,
first_response,
final_response[0],
params.selector_config,
meta={"proxy": proxy},
xhr_captured=xhr_captured,
)
return response
except Exception as e:
page_info.mark_error()
if attempt < self._config.retries - 1:
if is_proxy_error(e):
log.warning(
f"Proxy '{proxy}' failed (attempt {attempt + 1}) | Retrying in {self._config.retry_delay}s..."
)
else:
log.warning(
f"Attempt {attempt + 1} failed: {e}. Retrying in {self._config.retry_delay}s..."
)
await asyncio_sleep(self._config.retry_delay)
else:
log.warning(
f"Attempt {attempt + 1} failed: {e}. Retrying in {self._config.retry_delay}s..."
)
await asyncio_sleep(self._config.retry_delay)
else:
log.error(f"Failed after {self._config.retries} attempts: {e}")
raise
log.error(f"Failed after {self._config.retries} attempts: {e}")
raise
finally:
page.remove_listener("response", handler)
raise RuntimeError("Request failed") # pragma: no cover
tests/fetchers/async/test_dynamic_session.py
+60 -6
View File
@@ -52,21 +52,74 @@ class TestAsyncDynamicSession:
async def test_page_pool_management(self, urls):
"""Test page pool creation and reuse"""
async with AsyncDynamicSession() as session:
# The first request creates a page
# The first request creates a page, and it stays open afterwards
response = await session.fetch(urls["basic"])
assert response.status == 200
assert session.page_pool.pages_count == 0
# The second request should reuse the page
assert session.page_pool.pages_count == 1
page = session.page_pool.pages[0].page
assert session.page_pool.pages[0].state == "ready"
# The second request reuses the same tab
response = await session.fetch(urls["html"])
assert response.status == 200
assert session.page_pool.pages_count == 0
assert session.page_pool.pages_count == 1
assert session.page_pool.pages[0].page is page
# Check pool stats
stats = session.get_pool_stats()
assert stats["total_pages"] == 0
assert stats["total_pages"] == 1
assert stats["busy_pages"] == 0
assert stats["max_pages"] == 1
async def test_close_pages(self, urls):
"""Closing the tabs empties the pool, and the next request opens a fresh one"""
async with AsyncDynamicSession() as session:
await session.fetch(urls["basic"])
page = session.page_pool.pages[0].page
await session.close_pages()
assert session.page_pool.pages_count == 0
assert page.is_closed()
response = await session.fetch(urls["html"])
assert response.status == 200
assert session.page_pool.pages_count == 1
assert session.page_pool.pages[0].page is not page
async def test_per_request_headers_do_not_leak_into_the_next_request(self, httpbin):
"""A reused tab gets its settings reset, so headers from the previous request are gone"""
async with AsyncDynamicSession() as session:
response = await session.fetch(f"{httpbin.url}/headers", extra_headers={"X-Scrapling-First": "1"})
assert "X-Scrapling-First" in response.get_all_text()
response = await session.fetch(f"{httpbin.url}/headers", extra_headers={"X-Scrapling-Second": "2"})
assert "X-Scrapling-Second" in response.get_all_text()
assert "X-Scrapling-First" not in response.get_all_text(), "the new headers must replace the old ones"
response = await session.fetch(f"{httpbin.url}/headers")
assert "X-Scrapling-Second" not in response.get_all_text()
assert "X-Scrapling-First" not in response.get_all_text()
async def test_resource_blocking_does_not_leak_into_the_next_request(self, httpbin):
"""Routes registered for one request are removed before the tab is reused"""
js = """async () => {
const img = document.createElement('img');
const done = new Promise(r => { img.onload = () => r('loaded'); img.onerror = () => r('blocked'); });
img.src = '/image/png?' + Math.random();
document.body.appendChild(img);
return await done;
}"""
results = []
async def probe(page):
results.append(await page.evaluate(js))
async with AsyncDynamicSession() as session:
await session.fetch(f"{httpbin.url}/html", disable_resources=True, page_action=probe)
await session.fetch(f"{httpbin.url}/html", page_action=probe)
assert results == ["blocked", "loaded"]
assert session.page_pool.pages_count == 1
async def test_dynamic_session_with_options(self, urls):
"""Test AsyncDynamicSession with various options"""
async with AsyncDynamicSession(
@@ -83,3 +136,4 @@ class TestAsyncDynamicSession:
# Test with invalid URL
with pytest.raises(Exception):
await session.fetch("invalid://url")
assert session.page_pool.pages_count == 0, "errored tabs are closed and evicted"
tests/fetchers/async/test_stealth_session.py
+59 -5
View File
@@ -53,21 +53,74 @@ class TestAsyncStealthySession:
async def test_page_pool_management(self, urls):
"""Test page pool creation and reuse"""
async with AsyncStealthySession() as session:
# The first request creates a page
# The first request creates a page, and it stays open afterwards
response = await session.fetch(urls["basic"])
assert response.status == 200
assert session.page_pool.pages_count == 0
assert session.page_pool.pages_count == 1
page = session.page_pool.pages[0].page
assert session.page_pool.pages[0].state == "ready"
# The second request should reuse the page
# The second request reuses the same tab
response = await session.fetch(urls["html"])
assert response.status == 200
assert session.page_pool.pages_count == 0
assert session.page_pool.pages_count == 1
assert session.page_pool.pages[0].page is page
# Check pool stats
stats = session.get_pool_stats()
assert stats["total_pages"] == 0
assert stats["total_pages"] == 1
assert stats["busy_pages"] == 0
assert stats["max_pages"] == 1
async def test_close_pages(self, urls):
"""Closing the tabs empties the pool, and the next request opens a fresh one"""
async with AsyncStealthySession() as session:
await session.fetch(urls["basic"])
page = session.page_pool.pages[0].page
await session.close_pages()
assert session.page_pool.pages_count == 0
assert page.is_closed()
response = await session.fetch(urls["html"])
assert response.status == 200
assert session.page_pool.pages_count == 1
assert session.page_pool.pages[0].page is not page
async def test_per_request_headers_do_not_leak_into_the_next_request(self, httpbin):
"""A reused tab gets its settings reset, so headers from the previous request are gone"""
async with AsyncStealthySession() as session:
response = await session.fetch(f"{httpbin.url}/headers", extra_headers={"X-Scrapling-First": "1"})
assert "X-Scrapling-First" in response.get_all_text()
response = await session.fetch(f"{httpbin.url}/headers", extra_headers={"X-Scrapling-Second": "2"})
assert "X-Scrapling-Second" in response.get_all_text()
assert "X-Scrapling-First" not in response.get_all_text(), "the new headers must replace the old ones"
response = await session.fetch(f"{httpbin.url}/headers")
assert "X-Scrapling-Second" not in response.get_all_text()
assert "X-Scrapling-First" not in response.get_all_text()
async def test_resource_blocking_does_not_leak_into_the_next_request(self, httpbin):
"""Routes registered for one request are removed before the tab is reused"""
js = """async () => {
const img = document.createElement('img');
const done = new Promise(r => { img.onload = () => r('loaded'); img.onerror = () => r('blocked'); });
img.src = '/image/png?' + Math.random();
document.body.appendChild(img);
return await done;
}"""
results = []
async def probe(page):
results.append(await page.evaluate(js))
async with AsyncStealthySession() as session:
await session.fetch(f"{httpbin.url}/html", disable_resources=True, page_action=probe)
await session.fetch(f"{httpbin.url}/html", page_action=probe)
assert results == ["blocked", "loaded"]
assert session.page_pool.pages_count == 1
async def test_stealthy_session_with_options(self, urls):
"""Test AsyncStealthySession with various options"""
async with AsyncStealthySession(
@@ -84,3 +137,4 @@ class TestAsyncStealthySession:
# Test with invalid URL
with pytest.raises(Exception):
await session.fetch("invalid://url")
assert session.page_pool.pages_count == 0, "errored tabs are closed and evicted"
tests/fetchers/sync/test_stealth_session.py
+6
View File
@@ -92,3 +92,9 @@ class TestStealthySession:
result = StealthySession._detect_cloudflare(page_content)
assert result is None
assert session.fetch(self.status_200).status == 200
assert session.page_pool.pages_count == 1
page = session.page_pool.pages[0].page
assert session.fetch(self.status_200).status == 200
assert session.page_pool.pages[0].page is page, "sync sessions reuse their single tab"
session.close_pages()
assert session.page_pool.pages_count == 0
tests/fetchers/test_page_reuse.py
+129
View File
@@ -0,0 +1,129 @@
"""Tests for the tab reuse lifecycle of the browser sessions, with mocked Playwright objects."""
from unittest.mock import AsyncMock, Mock
import pytest
from scrapling.engines._browsers._base import AsyncSession, SyncSession
def _sync_page() -> Mock:
page = Mock()
page.is_closed.return_value = False
return page
def _async_page() -> Mock:
page = Mock()
page.is_closed.return_value = False
page.set_extra_http_headers = AsyncMock()
page.unroute_all = AsyncMock()
page.route = AsyncMock()
page.close = AsyncMock()
return page
def _sync_session(max_pages: int = 1) -> SyncSession:
session = SyncSession(max_pages=max_pages)
session.context = Mock()
session.context.new_page.side_effect = lambda: _sync_page()
return session
def _async_session(max_pages: int = 1) -> AsyncSession:
session = AsyncSession(max_pages=max_pages)
session.context = Mock()
session.context.new_page = AsyncMock(side_effect=lambda: _async_page())
return session
class TestSyncTabReuse:
def test_second_request_reuses_the_tab_and_resets_its_settings(self):
session = _sync_session()
with session._page_generator(1000, {"x-test": "1"}, True) as first:
first_page = first.page
assert session.page_pool.pages_count == 1
assert first.state == "ready"
with session._page_generator(2000, None, False) as second:
assert second.page is first_page
assert second.state == "busy"
assert session.context.new_page.call_count == 1
first_page.set_default_timeout.assert_called_with(2000)
first_page.set_extra_http_headers.assert_called_with({})
assert first_page.unroute_all.call_count == 2
first_page.route.assert_called_once()
def test_errored_tab_is_closed_and_evicted(self):
session = _sync_session()
with session._page_generator(1000, None, False) as page_info:
page_info.mark_error()
page_info.page.close.assert_called_once()
assert session.page_pool.pages_count == 0
with session._page_generator(1000, None, False):
pass
assert session.context.new_page.call_count == 2
def test_tab_closed_by_the_browser_is_evicted(self):
session = _sync_session()
with session._page_generator(1000, None, False) as page_info:
page_info.page.is_closed.return_value = True
assert session.page_pool.pages_count == 0
def test_close_pages_closes_every_tab_and_the_next_request_opens_a_new_one(self):
session = _sync_session()
with session._page_generator(1000, None, False) as page_info:
pass
session.close_pages()
page_info.page.close.assert_called_once()
assert session.page_pool.pages_count == 0
with session._page_generator(1000, None, False) as fresh:
assert fresh.page is not page_info.page
assert session.page_pool.pages_count == 1
class TestAsyncTabReuse:
@pytest.mark.asyncio
async def test_second_request_reuses_the_tab_and_resets_its_settings(self):
session = _async_session()
async with session._page_generator(1000, {"x-test": "1"}, True) as first:
first_page = first.page
assert session.page_pool.pages_count == 1
assert first.state == "ready"
async with session._page_generator(2000, None, False) as second:
assert second.page is first_page
assert second.state == "busy"
assert session.context.new_page.await_count == 1
first_page.set_default_timeout.assert_called_with(2000)
first_page.set_extra_http_headers.assert_awaited_with({})
assert first_page.unroute_all.await_count == 2
first_page.route.assert_awaited_once()
@pytest.mark.asyncio
async def test_errored_tab_is_closed_and_evicted(self):
session = _async_session()
async with session._page_generator(1000, None, False) as page_info:
page_info.mark_error()
page_info.page.close.assert_awaited_once()
assert session.page_pool.pages_count == 0
@pytest.mark.asyncio
async def test_pool_waits_for_a_ready_tab_at_capacity(self):
session = _async_session(max_pages=1)
session._max_wait_for_page = 1
async with session._page_generator(1000, None, False):
with pytest.raises(TimeoutError, match="No pages finished"):
async with session._page_generator(1000, None, False):
pass
@pytest.mark.asyncio
async def test_close_pages_closes_every_tab(self):
session = _async_session(max_pages=2)
async with session._page_generator(1000, None, False) as page_info:
pass
await session.close_pages()
page_info.page.close.assert_awaited_once()
assert session.page_pool.pages_count == 0
tests/fetchers/test_pages.py
+32 -16
View File
@@ -27,6 +27,10 @@ class TestPageInfo:
page_info.mark_error()
assert page_info.state == "error"
page_info.mark_ready()
assert page_info.state == "ready"
assert page_info.url == ""
def test_page_info_equality(self):
"""Test PageInfo equality comparison"""
mock_page1 = Mock()
@@ -70,8 +74,9 @@ class TestPagePool:
assert isinstance(page_info, PageInfo)
assert page_info.page == mock_page
assert page_info.state == "ready"
assert page_info.state == "busy", "a new page is busy for the request that opened it"
assert pool.pages_count == 1
assert pool.busy_count == 1
def test_add_page_limit_exceeded(self):
"""Test adding page when limit exceeded"""
@@ -88,28 +93,39 @@ class TestPagePool:
pool = PagePool(max_pages=1)
page_info = pool.add_page(Mock())
assert pool.pages_count == 1
pool.pages.remove(page_info)
pool.remove_page(page_info)
assert pool.pages_count == 0
pool.add_page(Mock())
assert pool.pages_count == 1
def test_cleanup_error_pages(self):
"""Test cleaning up error pages"""
def test_get_ready_page_skips_busy_and_error_pages(self):
pool = PagePool(max_pages=3)
busy = pool.add_page(Mock())
errored = pool.add_page(Mock())
errored.mark_error()
assert pool.get_ready_page() is None
# Add pages
page1 = pool.add_page(Mock())
_ = pool.add_page(Mock())
page3 = pool.add_page(Mock())
busy.mark_ready()
taken = pool.get_ready_page()
assert taken is busy
assert taken.state == "busy", "the returned page is reserved for the caller"
assert pool.get_ready_page() is None
# Mark some as error
page1.mark_error()
page3.mark_error()
def test_remove_page_ignores_unknown_pages(self):
pool = PagePool(max_pages=2)
page_info = pool.add_page(Mock())
pool.remove_page(page_info)
pool.remove_page(page_info)
assert pool.pages_count == 0
assert pool.pages_count == 3
def test_clear_returns_every_page_and_empties_the_pool(self):
pool = PagePool(max_pages=3)
pages = [pool.add_page(Mock()) for _ in range(3)]
pages[1].mark_ready()
pool.cleanup_error_pages()
cleared = pool.clear()
assert pool.pages_count == 1 # Only 2 should remain
assert cleared == pages
assert pool.pages_count == 0
pool.add_page(Mock())
assert pool.pages_count == 1