From 1d15349e07d91c34cc0cd7f26076bdf150882fcd Mon Sep 17 00:00:00 2001
From: Abdullah <52079299+AbdullahY36@users.noreply.github.com>
Date: Fri, 3 Apr 2026 15:08:33 +0200
Subject: [PATCH 1/8] feat(deps): add protego for robots.txt parsing and fix
 pyright type error in static.py

---
 pyproject.toml              | 1 +
 scrapling/engines/static.py | 1 +
 2 files changed, 2 insertions(+)

diff --git a/pyproject.toml b/pyproject.toml
index bb69194..3aee4f1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -67,6 +67,7 @@ dependencies = [
     "tld>=0.13.2",
     "w3lib>=2.4.1",
     "typing_extensions",
+    "protego>=0.4.0",
 ]
 
 [project.optional-dependencies]
diff --git a/scrapling/engines/static.py b/scrapling/engines/static.py
index 1f4b09b..a962ccf 100644
--- a/scrapling/engines/static.py
+++ b/scrapling/engines/static.py
@@ -250,6 +250,7 @@ class _SyncSessionLogic(_ConfigurationLogic):
                 request_args = self._merge_request_args(stealth=stealth, proxy=proxy, **kwargs)
                 try:
                     response = session.request(method, **request_args)
+                    assert response is not None
                     result = ResponseFactory.from_http_request(response, selector_config, meta={"proxy": proxy})
                     return result
                 except CurlError as e:  # pragma: no cover

From 0bbe62fc7f7011e28dc3d535d36d32944bd30d9a Mon Sep 17 00:00:00 2001
From: Abdullah <52079299+AbdullahY36@users.noreply.github.com>
Date: Fri, 3 Apr 2026 15:08:33 +0200
Subject: [PATCH 2/8] feat(spiders): implement RobotsTxtManager with concurrent
 fetch deduplication

---
 scrapling/spiders/robotstxt.py | 169 +++++++++++++++++++++++++++++++++
 1 file changed, 169 insertions(+)
 create mode 100644 scrapling/spiders/robotstxt.py

diff --git a/scrapling/spiders/robotstxt.py b/scrapling/spiders/robotstxt.py
new file mode 100644
index 0000000..4e8612f
--- /dev/null
+++ b/scrapling/spiders/robotstxt.py
@@ -0,0 +1,169 @@
+from asyncio import Event
+from urllib.parse import urlparse
+
+from protego import Protego
+
+from scrapling.core._types import Dict, Optional, Callable, Awaitable
+from scrapling.core.utils import log
+
+
+class RobotsTxtManager:
+    """Manages fetching, parsing, and caching of robots.txt files.
+
+    Accepts a fetch callable ``(url: str, sid: str) -> Awaitable[Response]``
+    so it stays decoupled from any specific session or transport layer.
+
+    All public methods accept only ``(url, sid)`` — domain and scheme are
+    derived internally from the URL so callers don't pass redundant data.
+
+    Handles all standard robots.txt directives including:
+    - User-agent specific rules
+    - Allow/Disallow directives (including wildcards and $ anchors)
+    - Crawl-delay directives
+
+    Deduplicates concurrent robots.txt fetches for the same domain — if multiple
+    requests for the same domain arrive before the first fetch completes, they
+    all wait for that single fetch instead of triggering redundant requests.
+    """
+
+    def __init__(self, fetch_fn: Callable[[str, str], Awaitable]):
+        self._fetch_fn = fetch_fn
+        self._cache: Dict[tuple[str, str], Protego] = {}
+        self._inflight: Dict[tuple[str, str], Event] = {}
+
+    async def _get_parser(self, url: str, sid: str) -> Protego:
+        parsed = urlparse(url)
+        domain = parsed.netloc
+        scheme = parsed.scheme or "https"
+        cache_key = (domain, sid)
+
+        # Return cached parser if available
+        if cache_key in self._cache:
+            return self._cache[cache_key]
+
+        # If a fetch is already in-flight for this domain, wait for it to complete
+        if cache_key in self._inflight:
+            await self._inflight[cache_key].wait()
+            return self._cache[cache_key]
+
+        # Mark fetch as in-flight to deduplicate concurrent requests
+        event = Event()
+        self._inflight[cache_key] = event
+
+        try:
+            robots_url = f"{scheme}://{domain}/robots.txt"
+            content = ""
+            try:
+                response = await self._fetch_fn(robots_url, sid)
+                if response.status == 200:
+                    content = response.body.decode(response.encoding, errors="replace")
+            except Exception as e:
+                log.warning(f"Failed to fetch robots.txt for {domain}: {e}")
+
+            try:
+                parser = Protego.parse(content)
+            except Exception as e:
+                log.warning(f"Failed to parse robots.txt for {domain}: {e}")
+                parser = Protego.parse("")
+
+            self._cache[cache_key] = parser
+        finally:
+            event.set()
+            del self._inflight[cache_key]
+
+        return parser
+
+    async def can_fetch(self, url: str, sid: str) -> bool:
+        """Check if a URL can be fetched according to the domain's robots.txt.
+
+        Handles:
+        - User-agent specific rules (e.g., User-agent: SpinarakBot)
+        - Wildcard user-agent rules (User-agent: *)
+        - Allow/Disallow directives with wildcards (e.g., /*.pdf$)
+        - Allow directives that override Disallow (e.g., Allow: /admin/public-docs/)
+
+        Uses the wildcard user-agent (*) which matches standard robots.txt directives
+        that apply to all bots. This is the conservative approach — if a URL is
+        disallowed for all bots, we respect that.
+
+        Args:
+            url: The full URL to check
+            sid: Session ID for fetching robots.txt
+
+        Returns:
+            True if the URL can be fetched, False otherwise
+        """
+        parser = await self._get_parser(url, sid)
+        return parser.can_fetch(url, "*")
+
+    async def get_crawl_delay(self, url: str, sid: str) -> Optional[float]:
+        """Get the crawl delay for this crawler.
+
+        Uses the wildcard user-agent (*) to get the general crawl delay
+        that applies to all bots.
+
+        Args:
+            url: Any URL on the domain to check
+            sid: Session ID for fetching robots.txt
+
+        Returns:
+            The crawl delay in seconds, or None if not specified
+        """
+        parser = await self._get_parser(url, sid)
+        delay = parser.crawl_delay("*")
+        return float(delay) if delay is not None else None
+
+    async def get_request_rate(self, url: str, sid: str) -> Optional[tuple[int, int]]:
+        """Get the request rate for this crawler.
+
+        Uses the wildcard user-agent (*) to get the general request rate
+        that applies to all bots.
+
+        Args:
+            url: Any URL on the domain to check
+            sid: Session ID for fetching robots.txt
+
+        Returns:
+            A tuple of (requests, seconds) if specified, or None if not specified
+        """
+        parser = await self._get_parser(url, sid)
+        rate = parser.request_rate("*")
+        if rate is not None:
+            return (rate.requests, rate.seconds)
+        return None
+
+    async def _get_delay_directives(self, url: str, sid: str) -> tuple[Optional[float], Optional[tuple[int, int]]]:
+        """Return both crawl-delay and request-rate in a single parser lookup.
+
+        Args:
+            url: Any URL on the domain to check
+            sid: Session ID for fetching robots.txt
+
+        Returns:
+            A tuple of (crawl_delay, request_rate) where crawl_delay is in seconds
+            or None, and request_rate is (requests, seconds) or None.
+        """
+        parser = await self._get_parser(url, sid)
+        c_delay = parser.crawl_delay("*")
+        rate = parser.request_rate("*")
+        return (
+            float(c_delay) if c_delay is not None else None,
+            (rate.requests, rate.seconds) if rate is not None else None,
+        )
+
+    def clear_cache(self, domain: Optional[str] = None, sid: Optional[str] = None) -> None:
+        """Clear the robots.txt cache.
+
+        Args:
+            domain: If specified, only clear cache for this domain
+            sid: If specified, only clear cache for this session ID
+                 If both are None, clears the entire cache
+        """
+        if domain is None and sid is None:
+            self._cache.clear()
+        else:
+            keys_to_remove = [
+                key for key in self._cache if (domain is None or key[0] == domain) and (sid is None or key[1] == sid)
+            ]
+            for key in keys_to_remove:
+                del self._cache[key]

From 5c40c6a8539282fcf6573f82f9152099c41120fa Mon Sep 17 00:00:00 2001
From: Abdullah <52079299+AbdullahY36@users.noreply.github.com>
Date: Fri, 3 Apr 2026 15:08:33 +0200
Subject: [PATCH 3/8] feat(spiders): integrate robots.txt compliance into the
 crawl engine

---
 scrapling/spiders/engine.py | 84 ++++++++++++++++++++++++++++++++++---
 scrapling/spiders/result.py |  2 +
 scrapling/spiders/spider.py |  3 ++
 3 files changed, 83 insertions(+), 6 deletions(-)

diff --git a/scrapling/spiders/engine.py b/scrapling/spiders/engine.py
index d77f838..cf1715e 100644
--- a/scrapling/spiders/engine.py
+++ b/scrapling/spiders/engine.py
@@ -10,6 +10,7 @@ from scrapling.core.utils import log
 from scrapling.spiders.request import Request
 from scrapling.spiders.scheduler import Scheduler
 from scrapling.spiders.session import SessionManager
+from scrapling.spiders.robotstxt import RobotsTxtManager
 from scrapling.spiders.result import CrawlStats, ItemList
 from scrapling.spiders.checkpoint import CheckpointManager, CheckpointData
 from scrapling.core._types import Dict, Union, Optional, TYPE_CHECKING, Any, AsyncGenerator
@@ -41,8 +42,18 @@ class CrawlerEngine:
         )
         self.stats = CrawlStats()
 
+        if self.spider.robots_txt_obey:
+
+            async def _fetch_robots(url: str, sid: str):
+                return await self.session_manager.fetch(Request(url, sid=sid))
+
+            self._robots_manager: Optional[RobotsTxtManager] = RobotsTxtManager(_fetch_robots)
+        else:
+            self._robots_manager = None
+
         self._global_limiter = CapacityLimiter(spider.concurrent_requests)
         self._domain_limiters: dict[str, CapacityLimiter] = {}
+        self._domain_delays: dict[str, float] = {}
         self._allowed_domains: set[str] = spider.allowed_domains or set()
 
         self._active_tasks: int = 0
@@ -68,13 +79,58 @@ class CrawlerEngine:
                 return True
         return False
 
+    async def _get_domain_delay(self, request: Request) -> float:
+        """Resolve the effective download delay for a domain.
+
+        Takes the max of the spider's configured delay and any robots.txt
+        directives (Crawl-delay / Request-rate). Result is cached per domain.
+        Also pre-creates a per-domain concurrency limiter of 1 when robots.txt
+        enforces any delay, before the caller acquires it via _rate_limiter().
+        """
+        robots_manager = self._robots_manager
+        if robots_manager is None:
+            return self.spider.download_delay
+
+        domain = request.domain
+
+        # Return cached delay if available
+        if domain in self._domain_delays:
+            return self._domain_delays[domain]
+
+        # Fetch both robots.txt directives in a single parser lookup
+        c_delay, r_rate = await robots_manager._get_delay_directives(request.url, request.sid)
+
+        delay = self.spider.download_delay
+        robots_enforced_delay = False
+
+        if r_rate:
+            req_count, period = r_rate
+            if req_count > 0:
+                delay = max(delay, period / req_count)
+                robots_enforced_delay = True
+
+        if c_delay is not None:
+            delay = max(delay, c_delay)
+            robots_enforced_delay = True
+
+        self._domain_delays[domain] = delay
+
+        # Enforce 1 concurrent request for this domain when robots.txt adds a delay
+        if robots_enforced_delay and delay > 0 and domain not in self._domain_limiters:
+            if self.spider.concurrent_requests_per_domain:
+                log.warning(
+                    f"robots.txt for {domain} enforces a delay, overriding"
+                    f" concurrent_requests_per_domain={self.spider.concurrent_requests_per_domain} with 1"
+                )
+            self._domain_limiters[domain] = CapacityLimiter(1)
+
+        return delay
+
     def _rate_limiter(self, domain: str) -> CapacityLimiter:
         """Get or create a per-domain concurrency limiter if enabled, otherwise use the global limiter."""
         if self.spider.concurrent_requests_per_domain:
-            if domain not in self._domain_limiters:
-                self._domain_limiters[domain] = CapacityLimiter(self.spider.concurrent_requests_per_domain)
-            return self._domain_limiters[domain]
-        return self._global_limiter
+            self._domain_limiters.setdefault(domain, CapacityLimiter(self.spider.concurrent_requests_per_domain))
+        return self._domain_limiters.get(domain, self._global_limiter)
 
     def _normalize_request(self, request: Request) -> None:
         """Normalize request fields before enqueueing.
@@ -87,9 +143,21 @@ class CrawlerEngine:
 
     async def _process_request(self, request: Request) -> None:
         """Download and process a single request."""
+        if self._robots_manager:
+            can_fetch = await self._robots_manager.can_fetch(request.url, request.sid)
+            if not can_fetch:
+                self.stats.robots_disallowed_count += 1
+                log.debug(f"Request disallowed by robots.txt: {request.url}")
+                return
+            # Must be called before _rate_limiter: may create CapacityLimiter(1) in _domain_limiters
+            # when robots.txt enforces a delay, which _rate_limiter then picks up.
+            delay = await self._get_domain_delay(request)
+        else:
+            delay = self.spider.download_delay
+
         async with self._rate_limiter(request.domain):
-            if self.spider.download_delay:
-                await anyio.sleep(self.spider.download_delay)
+            if delay:
+                await anyio.sleep(delay)
 
             if request._session_kwargs.get("proxy"):
                 self.stats.proxies.append(request._session_kwargs["proxy"])
@@ -227,15 +295,19 @@ class CrawlerEngine:
         self._pause_requested = False
         self._force_stop = False
         self.stats = CrawlStats(start_time=anyio.current_time())
+        self._domain_limiters.clear()
+        self._domain_delays.clear()
 
         # Check for existing checkpoint
         resuming = (await self._restore_from_checkpoint()) if self._checkpoint_system_enabled else False
         self._last_checkpoint_time = anyio.current_time()
 
         async with self.session_manager:
+            # Set stats from spider configuration
             self.stats.concurrent_requests = self.spider.concurrent_requests
             self.stats.concurrent_requests_per_domain = self.spider.concurrent_requests_per_domain
             self.stats.download_delay = self.spider.download_delay
+
             await self.spider.on_start(resuming=resuming)
 
             try:
diff --git a/scrapling/spiders/result.py b/scrapling/spiders/result.py
index 08a7658..b374152 100644
--- a/scrapling/spiders/result.py
+++ b/scrapling/spiders/result.py
@@ -47,6 +47,7 @@ class CrawlStats:
     concurrent_requests_per_domain: int = 0
     failed_requests_count: int = 0
     offsite_requests_count: int = 0
+    robots_disallowed_count: int = 0
     response_bytes: int = 0
     items_scraped: int = 0
     items_dropped: int = 0
@@ -95,6 +96,7 @@ class CrawlStats:
             "sessions_requests_count": self.sessions_requests_count,
             "failed_requests_count": self.failed_requests_count,
             "offsite_requests_count": self.offsite_requests_count,
+            "robots_disallowed_count": self.robots_disallowed_count,
             "blocked_requests_count": self.blocked_requests_count,
             "response_status_count": self.response_status_count,
             "response_bytes": self.response_bytes,
diff --git a/scrapling/spiders/spider.py b/scrapling/spiders/spider.py
index 4f38912..6aaa24f 100644
--- a/scrapling/spiders/spider.py
+++ b/scrapling/spiders/spider.py
@@ -72,6 +72,9 @@ class Spider(ABC):
     start_urls: list[str] = []
     allowed_domains: Set[str] = set()
 
+    # Robots.txt compliance
+    robots_txt_obey: bool = False
+
     # Concurrency settings
     concurrent_requests: int = 4
     concurrent_requests_per_domain: int = 0

From 132f33c84611c10d8c2fb2082fb63266bf1cc809 Mon Sep 17 00:00:00 2001
From: Abdullah <52079299+AbdullahY36@users.noreply.github.com>
Date: Fri, 3 Apr 2026 15:08:34 +0200
Subject: [PATCH 4/8] test(spiders): add comprehensive test suite for
 robots.txt compliance

---
 tests/spiders/test_engine.py    |   2 +
 tests/spiders/test_robotstxt.py | 615 ++++++++++++++++++++++++++++++++
 2 files changed, 617 insertions(+)
 create mode 100644 tests/spiders/test_robotstxt.py

diff --git a/tests/spiders/test_engine.py b/tests/spiders/test_engine.py
index b7bfd0f..e362036 100644
--- a/tests/spiders/test_engine.py
+++ b/tests/spiders/test_engine.py
@@ -83,6 +83,7 @@ class MockSpider:
         is_blocked_fn=None,
         on_scraped_item_fn=None,
         retry_blocked_request_fn=None,
+        robots_txt_obey: bool = False,
     ):
         self.concurrent_requests = concurrent_requests
         self.concurrent_requests_per_domain = concurrent_requests_per_domain
@@ -93,6 +94,7 @@ class MockSpider:
         self.fp_include_headers = fp_include_headers
         self.fp_keep_fragments = fp_keep_fragments
         self.name = "test_spider"
+        self.robots_txt_obey = robots_txt_obey
 
         # Tracking lists
         self.on_start_calls: list[dict] = []
diff --git a/tests/spiders/test_robotstxt.py b/tests/spiders/test_robotstxt.py
new file mode 100644
index 0000000..5a447c0
--- /dev/null
+++ b/tests/spiders/test_robotstxt.py
@@ -0,0 +1,615 @@
+"""Tests for RobotsTxtManager."""
+
+import asyncio
+
+import pytest
+
+from scrapling.spiders.robotstxt import RobotsTxtManager
+from scrapling.core._types import List, Optional
+
+
+# ---------------------------------------------------------------------------
+# Fixtures and helpers
+# ---------------------------------------------------------------------------
+
+
+class MockResponse:
+    """Minimal response stub matching the shape _get_parser expects."""
+
+    def __init__(self, status: int = 200, body: bytes = b"", encoding: str = "utf-8"):
+        self.status = status
+        self.body = body
+        self.encoding = encoding
+
+
+def make_fetch_fn(status: int = 200, content: str = "", encoding: str = "utf-8"):
+    """Return an async fetch callable that returns a fixed response.
+
+    Attaches a `.calls` list so tests can assert how many times it was invoked
+    and with which arguments.
+    """
+    calls: List[tuple] = []
+
+    async def _fetch(url: str, sid: str) -> MockResponse:
+        calls.append((url, sid))
+        return MockResponse(status=status, body=content.encode(encoding), encoding=encoding)
+
+    _fetch.calls = calls  # type: ignore[attr-defined]
+    return _fetch
+
+
+# ---------------------------------------------------------------------------
+# Shared robots.txt fixtures
+# ---------------------------------------------------------------------------
+
+ROBOTS_BASIC = """\
+User-agent: *
+Disallow: /admin/
+Crawl-delay: 2
+"""
+
+ROBOTS_WITH_RATE = """\
+User-agent: *
+Request-rate: 1/10
+Disallow: /private/
+"""
+
+ROBOTS_WITH_SITEMAP = """\
+User-agent: *
+Disallow:
+
+Sitemap: https://example.com/sitemap.xml
+Sitemap: https://example.com/sitemap2.xml
+"""
+
+ROBOTS_ALLOW_OVERRIDE = """\
+User-agent: *
+Disallow: /secret/
+Allow: /secret/public.html
+"""
+
+ROBOTS_DISALLOW_ALL = """\
+User-agent: *
+Disallow: /
+"""
+
+
+# ---------------------------------------------------------------------------
+# Tests: can_fetch
+# ---------------------------------------------------------------------------
+
+
+class TestCanFetch:
+    @pytest.mark.asyncio
+    async def test_allowed_url_returns_true(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC))
+
+        assert await mgr.can_fetch("https://example.com/products", "s1") is True
+
+    @pytest.mark.asyncio
+    async def test_disallowed_url_returns_false(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC))
+
+        assert await mgr.can_fetch("https://example.com/admin/", "s1") is False
+
+    @pytest.mark.asyncio
+    async def test_disallowed_subpath_returns_false(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC))
+
+        assert await mgr.can_fetch("https://example.com/admin/users", "s1") is False
+
+    @pytest.mark.asyncio
+    async def test_root_url_is_allowed(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC))
+
+        assert await mgr.can_fetch("https://example.com/", "s1") is True
+
+    @pytest.mark.asyncio
+    async def test_allow_directive_overrides_disallow(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_ALLOW_OVERRIDE))
+
+        assert await mgr.can_fetch("https://example.com/secret/public.html", "s1") is True
+        assert await mgr.can_fetch("https://example.com/secret/private.html", "s1") is False
+
+    @pytest.mark.asyncio
+    async def test_disallow_all_blocks_every_path(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_DISALLOW_ALL))
+
+        assert await mgr.can_fetch("https://example.com/", "s1") is False
+        assert await mgr.can_fetch("https://example.com/page", "s1") is False
+        assert await mgr.can_fetch("https://example.com/a/b/c", "s1") is False
+
+    @pytest.mark.asyncio
+    async def test_empty_robots_allows_everything(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=""))
+
+        assert await mgr.can_fetch("https://example.com/anything", "s1") is True
+        assert await mgr.can_fetch("https://example.com/admin/secret", "s1") is True
+
+    @pytest.mark.asyncio
+    async def test_non_200_response_allows_everything(self):
+        for status in [403, 404, 500, 503]:
+            mgr = RobotsTxtManager(make_fetch_fn(status=status))
+            result = await mgr.can_fetch("https://example.com/page", "s1")
+            assert result is True, f"Expected True for HTTP {status}"
+
+    @pytest.mark.asyncio
+    async def test_fetch_error_allows_everything(self):
+        async def failing_fetch(url: str, sid: str) -> MockResponse:
+            raise ConnectionError("network failure")
+
+        mgr = RobotsTxtManager(failing_fetch)
+
+        assert await mgr.can_fetch("https://example.com/page", "s1") is True
+
+    @pytest.mark.asyncio
+    async def test_wildcard_path_pattern(self):
+        content = "User-agent: *\nDisallow: /*.pdf$"
+        mgr = RobotsTxtManager(make_fetch_fn(content=content))
+
+        assert await mgr.can_fetch("https://example.com/report.pdf", "s1") is False
+        assert await mgr.can_fetch("https://example.com/report.html", "s1") is True
+
+    @pytest.mark.asyncio
+    async def test_returns_bool(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC))
+        result = await mgr.can_fetch("https://example.com/", "s1")
+        assert isinstance(result, bool)
+
+
+# ---------------------------------------------------------------------------
+# Tests: get_crawl_delay
+# ---------------------------------------------------------------------------
+
+
+class TestGetCrawlDelay:
+    @pytest.mark.asyncio
+    async def test_returns_float_when_set(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC))
+
+        delay = await mgr.get_crawl_delay("https://example.com/", "s1")
+
+        assert delay == 2.0
+        assert isinstance(delay, float)
+
+    @pytest.mark.asyncio
+    async def test_returns_none_when_not_set(self):
+        content = "User-agent: *\nDisallow: /admin/"
+        mgr = RobotsTxtManager(make_fetch_fn(content=content))
+
+        assert await mgr.get_crawl_delay("https://example.com/", "s1") is None
+
+    @pytest.mark.asyncio
+    async def test_returns_none_for_empty_robots(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=""))
+
+        assert await mgr.get_crawl_delay("https://example.com/", "s1") is None
+
+    @pytest.mark.asyncio
+    async def test_returns_none_on_fetch_error(self):
+        async def failing_fetch(url: str, sid: str) -> MockResponse:
+            raise ConnectionError("network failure")
+
+        mgr = RobotsTxtManager(failing_fetch)
+
+        assert await mgr.get_crawl_delay("https://example.com/", "s1") is None
+
+    @pytest.mark.asyncio
+    async def test_returns_none_for_non_200_response(self):
+        mgr = RobotsTxtManager(make_fetch_fn(status=404))
+
+        assert await mgr.get_crawl_delay("https://example.com/", "s1") is None
+
+    @pytest.mark.asyncio
+    async def test_fractional_delay(self):
+        content = "User-agent: *\nCrawl-delay: 0.5"
+        mgr = RobotsTxtManager(make_fetch_fn(content=content))
+
+        delay = await mgr.get_crawl_delay("https://example.com/", "s1")
+
+        assert delay == 0.5
+
+    @pytest.mark.asyncio
+    async def test_url_path_does_not_affect_result(self):
+        """Any URL on the same domain should return the same delay."""
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC))
+
+        d1 = await mgr.get_crawl_delay("https://example.com/", "s1")
+        d2 = await mgr.get_crawl_delay("https://example.com/deep/path/page.html", "s1")
+
+        assert d1 == d2
+
+
+# ---------------------------------------------------------------------------
+# Tests: get_request_rate
+# ---------------------------------------------------------------------------
+
+
+class TestGetRequestRate:
+    @pytest.mark.asyncio
+    async def test_returns_tuple_when_set(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_WITH_RATE))
+
+        rate = await mgr.get_request_rate("https://example.com/", "s1")
+
+        assert rate is not None
+        assert isinstance(rate, tuple)
+        assert len(rate) == 2
+
+    @pytest.mark.asyncio
+    async def test_tuple_contains_integers(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_WITH_RATE))
+
+        rate = await mgr.get_request_rate("https://example.com/", "s1")
+
+        assert rate is not None
+        requests, seconds = rate
+        assert isinstance(requests, int)
+        assert isinstance(seconds, int)
+
+    @pytest.mark.asyncio
+    async def test_returns_none_when_not_set(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC))
+
+        assert await mgr.get_request_rate("https://example.com/", "s1") is None
+
+    @pytest.mark.asyncio
+    async def test_returns_none_for_empty_robots(self):
+        mgr = RobotsTxtManager(make_fetch_fn(content=""))
+
+        assert await mgr.get_request_rate("https://example.com/", "s1") is None
+
+    @pytest.mark.asyncio
+    async def test_returns_none_on_fetch_error(self):
+        async def failing_fetch(url: str, sid: str) -> MockResponse:
+            raise ConnectionError("network failure")
+
+        mgr = RobotsTxtManager(failing_fetch)
+
+        assert await mgr.get_request_rate("https://example.com/", "s1") is None
+
+    @pytest.mark.asyncio
+    async def test_returns_none_for_non_200_response(self):
+        mgr = RobotsTxtManager(make_fetch_fn(status=404))
+
+        assert await mgr.get_request_rate("https://example.com/", "s1") is None
+
+
+# ---------------------------------------------------------------------------
+# Tests: get_sitemaps
+# ---------------------------------------------------------------------------
+
+
+# ---------------------------------------------------------------------------
+# Tests: caching behaviour
+# ---------------------------------------------------------------------------
+
+
+class TestCachingBehaviour:
+    @pytest.mark.asyncio
+    async def test_second_call_same_domain_uses_cache(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://example.com/page1", "s1")
+        await mgr.can_fetch("https://example.com/page2", "s1")
+
+        assert len(fetch_fn.calls) == 1
+
+    @pytest.mark.asyncio
+    async def test_all_methods_share_cache(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://example.com/", "s1")
+        await mgr.get_crawl_delay("https://example.com/", "s1")
+        await mgr.get_request_rate("https://example.com/", "s1")
+
+        assert len(fetch_fn.calls) == 1
+
+    @pytest.mark.asyncio
+    async def test_different_sids_use_separate_cache_entries(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://example.com/", "s1")
+        await mgr.can_fetch("https://example.com/", "s2")
+
+        assert len(fetch_fn.calls) == 2
+
+    @pytest.mark.asyncio
+    async def test_different_domains_use_separate_cache_entries(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://example.com/", "s1")
+        await mgr.can_fetch("https://other.com/", "s1")
+
+        assert len(fetch_fn.calls) == 2
+
+    @pytest.mark.asyncio
+    async def test_cache_keyed_by_domain_not_path(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://example.com/a/b/c", "s1")
+        await mgr.can_fetch("https://example.com/x/y/z", "s1")
+        await mgr.can_fetch("https://example.com/admin/", "s1")
+
+        assert len(fetch_fn.calls) == 1
+
+    @pytest.mark.asyncio
+    async def test_sid_is_passed_to_fetch_fn(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://example.com/", "my_session")
+
+        _, received_sid = fetch_fn.calls[0]
+        assert received_sid == "my_session"
+
+
+# ---------------------------------------------------------------------------
+# Tests: robots.txt URL construction
+# ---------------------------------------------------------------------------
+
+
+class TestRobotsTxtUrlConstruction:
+    @pytest.mark.asyncio
+    async def test_http_scheme_preserved(self):
+        fetch_fn = make_fetch_fn(content="")
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("http://example.com/page", "s1")
+
+        fetched_url, _ = fetch_fn.calls[0]
+        assert fetched_url == "http://example.com/robots.txt"
+
+    @pytest.mark.asyncio
+    async def test_https_scheme_preserved(self):
+        fetch_fn = make_fetch_fn(content="")
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://example.com/page", "s1")
+
+        fetched_url, _ = fetch_fn.calls[0]
+        assert fetched_url == "https://example.com/robots.txt"
+
+    @pytest.mark.asyncio
+    async def test_fetched_at_domain_root_regardless_of_request_path(self):
+        fetch_fn = make_fetch_fn(content="")
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://example.com/deep/nested/path/page.html", "s1")
+
+        fetched_url, _ = fetch_fn.calls[0]
+        assert fetched_url == "https://example.com/robots.txt"
+
+    @pytest.mark.asyncio
+    async def test_port_included_in_url(self):
+        fetch_fn = make_fetch_fn(content="")
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("http://example.com:8080/page", "s1")
+
+        fetched_url, _ = fetch_fn.calls[0]
+        assert fetched_url == "http://example.com:8080/robots.txt"
+
+    @pytest.mark.asyncio
+    async def test_different_ports_treated_as_different_domains(self):
+        fetch_fn = make_fetch_fn(content="")
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("http://example.com:8000/page", "s1")
+        await mgr.can_fetch("http://example.com:9000/page", "s1")
+
+        assert len(fetch_fn.calls) == 2
+        urls = [call[0] for call in fetch_fn.calls]
+        assert "http://example.com:8000/robots.txt" in urls
+        assert "http://example.com:9000/robots.txt" in urls
+
+
+# ---------------------------------------------------------------------------
+# Tests: encoding
+# ---------------------------------------------------------------------------
+
+
+class TestEncoding:
+    @pytest.mark.asyncio
+    async def test_non_utf8_body_decoded_with_response_encoding(self):
+        content = "User-agent: *\nDisallow: /admin/\nCrawl-delay: 3"
+        body = content.encode("latin-1")
+
+        async def fetch_fn(url: str, sid: str) -> MockResponse:
+            return MockResponse(status=200, body=body, encoding="latin-1")
+
+        mgr = RobotsTxtManager(fetch_fn)
+        delay = await mgr.get_crawl_delay("https://example.com/", "s1")
+
+        assert delay == 3.0
+
+    @pytest.mark.asyncio
+    async def test_bytes_body_decoded_correctly(self):
+        content = "User-agent: *\nDisallow: /private/"
+        body = content.encode("utf-8")
+
+        async def fetch_fn(url: str, sid: str) -> MockResponse:
+            return MockResponse(status=200, body=body, encoding="utf-8")
+
+        mgr = RobotsTxtManager(fetch_fn)
+
+        assert await mgr.can_fetch("https://example.com/private/", "s1") is False
+        assert await mgr.can_fetch("https://example.com/public/", "s1") is True
+
+
+# ---------------------------------------------------------------------------
+# Tests: clear_cache
+# ---------------------------------------------------------------------------
+
+
+class TestClearCache:
+    @pytest.mark.asyncio
+    async def test_clear_all_forces_refetch(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://example.com/", "s1")
+        mgr.clear_cache()
+        await mgr.can_fetch("https://example.com/", "s1")
+
+        assert len(fetch_fn.calls) == 2
+
+    @pytest.mark.asyncio
+    async def test_clear_by_domain_only_invalidates_that_domain(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://example.com/", "s1")
+        await mgr.can_fetch("https://other.com/", "s1")
+        assert len(fetch_fn.calls) == 2
+
+        mgr.clear_cache(domain="example.com")
+
+        await mgr.can_fetch("https://example.com/", "s1")  # refetched
+        await mgr.can_fetch("https://other.com/", "s1")    # still cached
+
+        assert len(fetch_fn.calls) == 3
+
+    @pytest.mark.asyncio
+    async def test_clear_by_sid_only_invalidates_that_sid(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://example.com/", "s1")
+        await mgr.can_fetch("https://example.com/", "s2")
+        assert len(fetch_fn.calls) == 2
+
+        mgr.clear_cache(sid="s1")
+
+        await mgr.can_fetch("https://example.com/", "s1")  # refetched
+        await mgr.can_fetch("https://example.com/", "s2")  # still cached
+
+        assert len(fetch_fn.calls) == 3
+
+    @pytest.mark.asyncio
+    async def test_clear_by_domain_and_sid_targets_exact_entry(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://example.com/", "s1")
+        await mgr.can_fetch("https://example.com/", "s2")
+        assert len(fetch_fn.calls) == 2
+
+        mgr.clear_cache(domain="example.com", sid="s1")
+
+        await mgr.can_fetch("https://example.com/", "s1")  # refetched
+        await mgr.can_fetch("https://example.com/", "s2")  # still cached
+
+        assert len(fetch_fn.calls) == 3
+
+    def test_clear_nonexistent_domain_does_not_raise(self):
+        mgr = RobotsTxtManager(make_fetch_fn())
+        mgr.clear_cache(domain="nevervisited.com")  # should not raise
+
+    def test_clear_empty_cache_does_not_raise(self):
+        mgr = RobotsTxtManager(make_fetch_fn())
+        mgr.clear_cache()  # should not raise
+
+    @pytest.mark.asyncio
+    async def test_clear_all_empties_cache_completely(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.can_fetch("https://a.com/", "s1")
+        await mgr.can_fetch("https://b.com/", "s1")
+        await mgr.can_fetch("https://c.com/", "s1")
+        assert len(fetch_fn.calls) == 3
+
+        mgr.clear_cache()
+
+        await mgr.can_fetch("https://a.com/", "s1")
+        await mgr.can_fetch("https://b.com/", "s1")
+        await mgr.can_fetch("https://c.com/", "s1")
+
+        assert len(fetch_fn.calls) == 6
+
+
+# ---------------------------------------------------------------------------
+# Tests: concurrent access (double-checked locking)
+# ---------------------------------------------------------------------------
+
+
+class TestConcurrency:
+    @pytest.mark.asyncio
+    async def test_concurrent_calls_same_domain_same_sid_deduplicated(self):
+        """Multiple concurrent tasks for the same domain+sid trigger only one robots.txt fetch."""
+        fetch_count = 0
+
+        async def slow_fetch(url: str, sid: str) -> MockResponse:
+            nonlocal fetch_count
+            fetch_count += 1
+            await asyncio.sleep(0.02)  # simulate network latency
+            return MockResponse(status=200, body=ROBOTS_BASIC.encode(), encoding="utf-8")
+
+        mgr = RobotsTxtManager(slow_fetch)
+
+        results = await asyncio.gather(*[
+            mgr.can_fetch(f"https://example.com/page{i}", "s1")
+            for i in range(8)
+        ])
+
+        # Concurrent calls for the same domain+sid are deduplicated to a single fetch
+        assert fetch_count == 1
+        assert all(isinstance(r, bool) for r in results)
+
+    @pytest.mark.asyncio
+    async def test_concurrent_calls_different_domains_fetch_independently(self):
+        fetch_count = 0
+
+        async def slow_fetch(url: str, sid: str) -> MockResponse:
+            nonlocal fetch_count
+            fetch_count += 1
+            await asyncio.sleep(0.01)
+            return MockResponse(status=200, body=b"", encoding="utf-8")
+
+        mgr = RobotsTxtManager(slow_fetch)
+
+        await asyncio.gather(
+            mgr.can_fetch("https://alpha.com/", "s1"),
+            mgr.can_fetch("https://beta.com/", "s1"),
+            mgr.can_fetch("https://gamma.com/", "s1"),
+        )
+
+        assert fetch_count == 3
+
+    @pytest.mark.asyncio
+    async def test_concurrent_calls_consistent_results(self):
+        """All concurrent callers should see the same allow/disallow result."""
+        mgr = RobotsTxtManager(make_fetch_fn(content=ROBOTS_BASIC))
+
+        results = await asyncio.gather(*[
+            mgr.can_fetch("https://example.com/admin/", "s1")
+            for _ in range(6)
+        ])
+
+        assert all(r is False for r in results)
+
+    @pytest.mark.asyncio
+    async def test_different_sids_concurrent_fetch_independently(self):
+        fetch_count = 0
+
+        async def slow_fetch(url: str, sid: str) -> MockResponse:
+            nonlocal fetch_count
+            fetch_count += 1
+            await asyncio.sleep(0.01)
+            return MockResponse(status=200, body=b"", encoding="utf-8")
+
+        mgr = RobotsTxtManager(slow_fetch)
+
+        await asyncio.gather(
+            mgr.can_fetch("https://example.com/", "s1"),
+            mgr.can_fetch("https://example.com/", "s2"),
+            mgr.can_fetch("https://example.com/", "s3"),
+        )
+
+        assert fetch_count == 3

From 07129ce4b1645f73df25943eb88c215c05e1f14d Mon Sep 17 00:00:00 2001
From: Abdullah <52079299+AbdullahY36@users.noreply.github.com>
Date: Fri, 3 Apr 2026 17:51:00 +0200
Subject: [PATCH 5/8] feat(deps): move protego to fetchers optional dependency

protego is only used by the spider framework for robots.txt compliance.

Moving it from core dependencies to the optional 'fetchers' group reduces
the dependency footprint for users who don't need the spider framework.

for pyproject.toml file
---
 pyproject.toml | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/pyproject.toml b/pyproject.toml
index 41fb14c..4333d7e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -66,8 +66,7 @@ dependencies = [
     "orjson>=3.11.8",
     "tld>=0.13.2",
     "w3lib>=2.4.1",
-    "typing_extensions",
-    "protego>=0.4.0",
+    "typing_extensions"
 ]
 
 [project.optional-dependencies]
@@ -79,7 +78,8 @@ fetchers = [
     "browserforge>=1.2.4",
     "apify-fingerprint-datapoints>=0.12.0",
     "msgspec>=0.20.0",
-    "anyio>=4.12.1"
+    "anyio>=4.12.1",
+    "protego>=0.4.0",
 ]
 ai = [
     "mcp>=1.26.0",

From e2b293f41c289a496195b547df89482a23a44739 Mon Sep 17 00:00:00 2001
From: Abdullah <52079299+AbdullahY36@users.noreply.github.com>
Date: Sat, 4 Apr 2026 03:00:15 +0200
Subject: [PATCH 6/8] refactor(spiders): simplify robots.txt cache to
 domain-only key
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

robots.txt is a domain-level document and does not vary by session.

Keying the cache by (domain, sid) was both wasteful and incorrect —
it caused redundant fetches when the same domain was accessed by different sessions.

- Cache is now keyed by domain string only; all sessions share one entry
- Removed asyncio.Event inflight-deduplication mechanism (superseded by the prefetch approach added in the next commit)
- clear_cache() loses the `sid` parameter (breaking change); clearing a domain now evicts the single shared entry for all sessions
- Updated tests to reflect shared-cache semantics

Files: scrapling/spiders/robotstxt.py, tests/spiders/test_robotstxt.py
---
 scrapling/spiders/robotstxt.py  | 108 +++++++++++++-------------
 tests/spiders/test_robotstxt.py | 131 ++++++++++++++++++--------------
 2 files changed, 128 insertions(+), 111 deletions(-)

diff --git a/scrapling/spiders/robotstxt.py b/scrapling/spiders/robotstxt.py
index 4e8612f..c64c66e 100644
--- a/scrapling/spiders/robotstxt.py
+++ b/scrapling/spiders/robotstxt.py
@@ -1,6 +1,6 @@
-from asyncio import Event
 from urllib.parse import urlparse
 
+from anyio import create_task_group
 from protego import Protego
 
 from scrapling.core._types import Dict, Optional, Callable, Awaitable
@@ -21,56 +21,40 @@ class RobotsTxtManager:
     - Allow/Disallow directives (including wildcards and $ anchors)
     - Crawl-delay directives
 
-    Deduplicates concurrent robots.txt fetches for the same domain — if multiple
-    requests for the same domain arrive before the first fetch completes, they
-    all wait for that single fetch instead of triggering redundant requests.
+    robots.txt is a domain-level document and does not vary by session, so the
+    cache is keyed by domain only. The ``sid`` parameter on public methods
+    controls which session is used for the initial fetch if the domain is not
+    yet cached, but all sessions share the same parsed result afterwards.
     """
 
     def __init__(self, fetch_fn: Callable[[str, str], Awaitable]):
         self._fetch_fn = fetch_fn
-        self._cache: Dict[tuple[str, str], Protego] = {}
-        self._inflight: Dict[tuple[str, str], Event] = {}
+        self._cache: Dict[str, Protego] = {}
 
     async def _get_parser(self, url: str, sid: str) -> Protego:
         parsed = urlparse(url)
         domain = parsed.netloc
+
+        if domain in self._cache:
+            return self._cache[domain]
+
         scheme = parsed.scheme or "https"
-        cache_key = (domain, sid)
-
-        # Return cached parser if available
-        if cache_key in self._cache:
-            return self._cache[cache_key]
-
-        # If a fetch is already in-flight for this domain, wait for it to complete
-        if cache_key in self._inflight:
-            await self._inflight[cache_key].wait()
-            return self._cache[cache_key]
-
-        # Mark fetch as in-flight to deduplicate concurrent requests
-        event = Event()
-        self._inflight[cache_key] = event
+        robots_url = f"{scheme}://{domain}/robots.txt"
+        content = ""
+        try:
+            response = await self._fetch_fn(robots_url, sid)
+            if response.status == 200:
+                content = response.body.decode(response.encoding, errors="replace")
+        except Exception as e:
+            log.warning(f"Failed to fetch robots.txt for {domain}: {e}")
 
         try:
-            robots_url = f"{scheme}://{domain}/robots.txt"
-            content = ""
-            try:
-                response = await self._fetch_fn(robots_url, sid)
-                if response.status == 200:
-                    content = response.body.decode(response.encoding, errors="replace")
-            except Exception as e:
-                log.warning(f"Failed to fetch robots.txt for {domain}: {e}")
-
-            try:
-                parser = Protego.parse(content)
-            except Exception as e:
-                log.warning(f"Failed to parse robots.txt for {domain}: {e}")
-                parser = Protego.parse("")
-
-            self._cache[cache_key] = parser
-        finally:
-            event.set()
-            del self._inflight[cache_key]
+            parser = Protego.parse(content)
+        except Exception as e:
+            log.warning(f"Failed to parse robots.txt for {domain}: {e}")
+            parser = Protego.parse("")
 
+        self._cache[domain] = parser
         return parser
 
     async def can_fetch(self, url: str, sid: str) -> bool:
@@ -88,7 +72,7 @@ class RobotsTxtManager:
 
         Args:
             url: The full URL to check
-            sid: Session ID for fetching robots.txt
+            sid: Session ID for fetching robots.txt if not yet cached
 
         Returns:
             True if the URL can be fetched, False otherwise
@@ -104,7 +88,7 @@ class RobotsTxtManager:
 
         Args:
             url: Any URL on the domain to check
-            sid: Session ID for fetching robots.txt
+            sid: Session ID for fetching robots.txt if not yet cached
 
         Returns:
             The crawl delay in seconds, or None if not specified
@@ -121,7 +105,7 @@ class RobotsTxtManager:
 
         Args:
             url: Any URL on the domain to check
-            sid: Session ID for fetching robots.txt
+            sid: Session ID for fetching robots.txt if not yet cached
 
         Returns:
             A tuple of (requests, seconds) if specified, or None if not specified
@@ -137,7 +121,7 @@ class RobotsTxtManager:
 
         Args:
             url: Any URL on the domain to check
-            sid: Session ID for fetching robots.txt
+            sid: Session ID for fetching robots.txt if not yet cached
 
         Returns:
             A tuple of (crawl_delay, request_rate) where crawl_delay is in seconds
@@ -151,19 +135,35 @@ class RobotsTxtManager:
             (rate.requests, rate.seconds) if rate is not None else None,
         )
 
-    def clear_cache(self, domain: Optional[str] = None, sid: Optional[str] = None) -> None:
-        """Clear the robots.txt cache.
+    async def prefetch(self, urls: list[str], sid: str) -> None:
+        """Pre-warm the robots.txt cache for a list of seed URLs concurrently.
+
+        Callers are responsible for deduplicating URLs by domain before calling
+        this method — passing multiple URLs for the same domain will trigger
+        redundant fetches since no inflight deduplication exists here.
 
         Args:
-            domain: If specified, only clear cache for this domain
-            sid: If specified, only clear cache for this session ID
-                 If both are None, clears the entire cache
+            urls: Seed URLs whose domains should be pre-fetched (one per domain).
+            sid: Session ID to use for the robots.txt fetch requests.
         """
-        if domain is None and sid is None:
+        if not urls:
+            return
+        log.debug(f"Pre-fetching robots.txt for {len(urls)} domain(s)")
+        async with create_task_group() as tg:
+            for url in urls:
+                tg.start_soon(self._get_parser, url, sid)
+
+    def clear_cache(self, domain: Optional[str] = None) -> None:
+        """Clear the robots.txt cache.
+
+        Note: the ``sid`` parameter was removed — the cache is now keyed by
+        domain only, so clearing a domain evicts all sessions at once.
+
+        Args:
+            domain: If specified, only clear cache for this domain.
+                    If None, clears the entire cache.
+        """
+        if domain is None:
             self._cache.clear()
         else:
-            keys_to_remove = [
-                key for key in self._cache if (domain is None or key[0] == domain) and (sid is None or key[1] == sid)
-            ]
-            for key in keys_to_remove:
-                del self._cache[key]
+            self._cache.pop(domain, None)
diff --git a/tests/spiders/test_robotstxt.py b/tests/spiders/test_robotstxt.py
index 5a447c0..efb8328 100644
--- a/tests/spiders/test_robotstxt.py
+++ b/tests/spiders/test_robotstxt.py
@@ -5,7 +5,6 @@ import asyncio
 import pytest
 
 from scrapling.spiders.robotstxt import RobotsTxtManager
-from scrapling.core._types import List, Optional
 
 
 # ---------------------------------------------------------------------------
@@ -28,7 +27,7 @@ def make_fetch_fn(status: int = 200, content: str = "", encoding: str = "utf-8")
     Attaches a `.calls` list so tests can assert how many times it was invoked
     and with which arguments.
     """
-    calls: List[tuple] = []
+    calls: list[tuple] = []
 
     async def _fetch(url: str, sid: str) -> MockResponse:
         calls.append((url, sid))
@@ -275,11 +274,6 @@ class TestGetRequestRate:
         assert await mgr.get_request_rate("https://example.com/", "s1") is None
 
 
-# ---------------------------------------------------------------------------
-# Tests: get_sitemaps
-# ---------------------------------------------------------------------------
-
-
 # ---------------------------------------------------------------------------
 # Tests: caching behaviour
 # ---------------------------------------------------------------------------
@@ -308,14 +302,15 @@ class TestCachingBehaviour:
         assert len(fetch_fn.calls) == 1
 
     @pytest.mark.asyncio
-    async def test_different_sids_use_separate_cache_entries(self):
+    async def test_different_sids_share_cache_entry(self):
+        """robots.txt is domain-level — different sessions share the same cached parser."""
         fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
         mgr = RobotsTxtManager(fetch_fn)
 
         await mgr.can_fetch("https://example.com/", "s1")
         await mgr.can_fetch("https://example.com/", "s2")
 
-        assert len(fetch_fn.calls) == 2
+        assert len(fetch_fn.calls) == 1
 
     @pytest.mark.asyncio
     async def test_different_domains_use_separate_cache_entries(self):
@@ -476,37 +471,21 @@ class TestClearCache:
         assert len(fetch_fn.calls) == 3
 
     @pytest.mark.asyncio
-    async def test_clear_by_sid_only_invalidates_that_sid(self):
+    async def test_clear_by_domain_invalidates_all_sessions(self):
+        """Clearing a domain evicts the single shared cache entry for all sessions."""
         fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
         mgr = RobotsTxtManager(fetch_fn)
 
         await mgr.can_fetch("https://example.com/", "s1")
-        await mgr.can_fetch("https://example.com/", "s2")
+        assert len(fetch_fn.calls) == 1
+
+        mgr.clear_cache(domain="example.com")
+
+        await mgr.can_fetch("https://example.com/", "s1")  # refetched — cache was cleared
+        await mgr.can_fetch("https://example.com/", "s2")  # hits the newly warm cache, no fetch
+
         assert len(fetch_fn.calls) == 2
 
-        mgr.clear_cache(sid="s1")
-
-        await mgr.can_fetch("https://example.com/", "s1")  # refetched
-        await mgr.can_fetch("https://example.com/", "s2")  # still cached
-
-        assert len(fetch_fn.calls) == 3
-
-    @pytest.mark.asyncio
-    async def test_clear_by_domain_and_sid_targets_exact_entry(self):
-        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
-        mgr = RobotsTxtManager(fetch_fn)
-
-        await mgr.can_fetch("https://example.com/", "s1")
-        await mgr.can_fetch("https://example.com/", "s2")
-        assert len(fetch_fn.calls) == 2
-
-        mgr.clear_cache(domain="example.com", sid="s1")
-
-        await mgr.can_fetch("https://example.com/", "s1")  # refetched
-        await mgr.can_fetch("https://example.com/", "s2")  # still cached
-
-        assert len(fetch_fn.calls) == 3
-
     def test_clear_nonexistent_domain_does_not_raise(self):
         mgr = RobotsTxtManager(make_fetch_fn())
         mgr.clear_cache(domain="nevervisited.com")  # should not raise
@@ -535,32 +514,30 @@ class TestClearCache:
 
 
 # ---------------------------------------------------------------------------
-# Tests: concurrent access (double-checked locking)
+# Tests: concurrent access
 # ---------------------------------------------------------------------------
 
 
-class TestConcurrency:
+class TestCacheAndConcurrency:
     @pytest.mark.asyncio
-    async def test_concurrent_calls_same_domain_same_sid_deduplicated(self):
-        """Multiple concurrent tasks for the same domain+sid trigger only one robots.txt fetch."""
+    async def test_cached_domain_not_refetched(self):
+        """Once a domain is cached, subsequent calls return the cached parser without fetching."""
         fetch_count = 0
 
-        async def slow_fetch(url: str, sid: str) -> MockResponse:
+        async def counting_fetch(url: str, sid: str) -> MockResponse:
             nonlocal fetch_count
             fetch_count += 1
-            await asyncio.sleep(0.02)  # simulate network latency
             return MockResponse(status=200, body=ROBOTS_BASIC.encode(), encoding="utf-8")
 
-        mgr = RobotsTxtManager(slow_fetch)
+        mgr = RobotsTxtManager(counting_fetch)
 
-        results = await asyncio.gather(*[
-            mgr.can_fetch(f"https://example.com/page{i}", "s1")
-            for i in range(8)
-        ])
+        # First call fetches and caches
+        await mgr.can_fetch("https://example.com/page1", "s1")
+        # Subsequent calls hit the cache
+        for i in range(7):
+            await mgr.can_fetch(f"https://example.com/page{i + 2}", "s1")
 
-        # Concurrent calls for the same domain+sid are deduplicated to a single fetch
         assert fetch_count == 1
-        assert all(isinstance(r, bool) for r in results)
 
     @pytest.mark.asyncio
     async def test_concurrent_calls_different_domains_fetch_independently(self):
@@ -595,21 +572,61 @@ class TestConcurrency:
         assert all(r is False for r in results)
 
     @pytest.mark.asyncio
-    async def test_different_sids_concurrent_fetch_independently(self):
+    async def test_different_sids_share_cache_after_first_fetch(self):
+        """After the first fetch, all sessions share the cached parser regardless of sid."""
         fetch_count = 0
 
-        async def slow_fetch(url: str, sid: str) -> MockResponse:
+        async def counting_fetch(url: str, sid: str) -> MockResponse:
             nonlocal fetch_count
             fetch_count += 1
-            await asyncio.sleep(0.01)
             return MockResponse(status=200, body=b"", encoding="utf-8")
 
-        mgr = RobotsTxtManager(slow_fetch)
+        mgr = RobotsTxtManager(counting_fetch)
 
-        await asyncio.gather(
-            mgr.can_fetch("https://example.com/", "s1"),
-            mgr.can_fetch("https://example.com/", "s2"),
-            mgr.can_fetch("https://example.com/", "s3"),
-        )
+        # First call fetches and caches
+        await mgr.can_fetch("https://example.com/", "s1")
+        # s2 and s3 hit the cache — no additional fetches
+        await mgr.can_fetch("https://example.com/", "s2")
+        await mgr.can_fetch("https://example.com/", "s3")
 
-        assert fetch_count == 3
+        assert fetch_count == 1
+
+
+# ---------------------------------------------------------------------------
+# Tests: prefetch
+# ---------------------------------------------------------------------------
+
+
+class TestPrefetch:
+    @pytest.mark.asyncio
+    async def test_prefetch_fetches_all_domains(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.prefetch(["https://a.com/", "https://b.com/", "https://c.com/"], "s1")
+
+        assert len(fetch_fn.calls) == 3
+        fetched = {url for url, _ in fetch_fn.calls}
+        assert fetched == {"https://a.com/robots.txt", "https://b.com/robots.txt", "https://c.com/robots.txt"}
+
+    @pytest.mark.asyncio
+    async def test_prefetch_warms_cache_for_subsequent_calls(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.prefetch(["https://example.com/"], "s1")
+        assert len(fetch_fn.calls) == 1
+
+        # Any subsequent call for the same domain hits the cache
+        await mgr.can_fetch("https://example.com/products", "s1")
+        await mgr.can_fetch("https://example.com/products", "s2")
+        assert len(fetch_fn.calls) == 1
+
+    @pytest.mark.asyncio
+    async def test_prefetch_empty_list_is_noop(self):
+        fetch_fn = make_fetch_fn(content=ROBOTS_BASIC)
+        mgr = RobotsTxtManager(fetch_fn)
+
+        await mgr.prefetch([], "s1")
+
+        assert len(fetch_fn.calls) == 0

From a86e9709ea0442fcf7b1c9b0459ac0716514d8a6 Mon Sep 17 00:00:00 2001
From: Abdullah <52079299+AbdullahY36@users.noreply.github.com>
Date: Sat, 4 Apr 2026 03:00:15 +0200
Subject: [PATCH 7/8] feat(spiders): pre-warm robots.txt cache before crawl
 loop starts
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Previously robots.txt was fetched lazily on the first request per
domain, causing early concurrent requests to each stall waiting for
the same network fetch. The cache is now warmed before the crawl loop
starts, making all subsequent robots.txt lookups a local read.

- RobotsTxtManager gains a prefetch(urls, sid) method that fetches all domains concurrently via a task group
- CrawlerEngine._prefetch_robots_txt() is called after on_start():
  uses allowed_domains if configured, otherwise falls back to unique
  domains extracted from start_urls
- Mid-crawl domain discovery (not covered by prefetch) still fetches
  lazily; two concurrent callbacks on the same new domain can each
  trigger a fetch — accepted tradeoff, documented in _get_domain_delay

Files: scrapling/spiders/robotstxt.py, scrapling/spiders/engine.py, tests/spiders/test_engine.py
---
 scrapling/spiders/engine.py  | 28 +++++++++++-
 tests/spiders/test_engine.py | 86 +++++++++++++++++++++++++++++++++++-
 2 files changed, 112 insertions(+), 2 deletions(-)

diff --git a/scrapling/spiders/engine.py b/scrapling/spiders/engine.py
index cf1715e..163101b 100644
--- a/scrapling/spiders/engine.py
+++ b/scrapling/spiders/engine.py
@@ -97,7 +97,10 @@ class CrawlerEngine:
         if domain in self._domain_delays:
             return self._domain_delays[domain]
 
-        # Fetch both robots.txt directives in a single parser lookup
+        # For domains covered by _prefetch_robots_txt this is a local parser read.
+        # Domains discovered mid-crawl (not in start_urls/allowed_domains) will fetch here.
+        # Two concurrent callbacks hitting the same new domain can each trigger a fetch;
+        # the second write is a no-op in effect (same content), but the extra request is accepted.
         c_delay, r_rate = await robots_manager._get_delay_directives(request.url, request.sid)
 
         delay = self.spider.download_delay
@@ -287,6 +290,27 @@ class CrawlerEngine:
 
         return True
 
+    async def _prefetch_robots_txt(self) -> None:
+        """Pre-warm the robots.txt cache before the crawl loop starts.
+
+        Uses allowed_domains if configured, otherwise falls back to unique domains
+        extracted from start_urls via Request.domain. Both paths use https.
+        """
+        if not self._robots_manager:
+            return
+
+        if self._allowed_domains:
+            domains = self._allowed_domains
+        elif self.spider.start_urls:
+            # Deduplicate by domain so we spawn exactly one task per domain
+            domains = {Request(url).domain for url in self.spider.start_urls}
+        else:
+            return
+
+        seed_urls = [f"https://{domain}/" for domain in domains]
+
+        await self._robots_manager.prefetch(seed_urls, self.session_manager.default_session_id)
+
     async def crawl(self) -> CrawlStats:
         """Run the spider and return CrawlStats."""
         self._running = True
@@ -310,6 +334,8 @@ class CrawlerEngine:
 
             await self.spider.on_start(resuming=resuming)
 
+            await self._prefetch_robots_txt()
+
             try:
                 if not resuming:
                     async for request in self.spider.start_requests():
diff --git a/tests/spiders/test_engine.py b/tests/spiders/test_engine.py
index e362036..e7382f8 100644
--- a/tests/spiders/test_engine.py
+++ b/tests/spiders/test_engine.py
@@ -8,6 +8,7 @@ import pytest
 
 from scrapling.spiders.engine import CrawlerEngine, _dump
 from scrapling.spiders.request import Request
+from scrapling.spiders.robotstxt import RobotsTxtManager
 from scrapling.spiders.session import SessionManager
 from scrapling.spiders.result import CrawlStats, ItemList
 from scrapling.spiders.checkpoint import CheckpointData
@@ -22,10 +23,11 @@ from scrapling.core._types import Any, Dict, Set, AsyncGenerator
 class MockResponse:
     """Minimal Response stand-in."""
 
-    def __init__(self, status: int = 200, body: bytes = b"ok", url: str = "https://example.com"):
+    def __init__(self, status: int = 200, body: bytes = b"ok", url: str = "https://example.com", encoding: str = "utf-8"):
         self.status = status
         self.body = body
         self.url = url
+        self.encoding = encoding
         self.request: Any = None
         self.meta: Dict[str, Any] = {}
 
@@ -84,6 +86,7 @@ class MockSpider:
         on_scraped_item_fn=None,
         retry_blocked_request_fn=None,
         robots_txt_obey: bool = False,
+        start_urls: list[str] | None = None,
     ):
         self.concurrent_requests = concurrent_requests
         self.concurrent_requests_per_domain = concurrent_requests_per_domain
@@ -95,6 +98,7 @@ class MockSpider:
         self.fp_keep_fragments = fp_keep_fragments
         self.name = "test_spider"
         self.robots_txt_obey = robots_txt_obey
+        self.start_urls = start_urls or []
 
         # Tracking lists
         self.on_start_calls: list[dict] = []
@@ -914,3 +918,83 @@ class TestPauseDuringCrawl:
         await engine.crawl()
 
         assert engine.paused is False
+
+
+# ---------------------------------------------------------------------------
+# Tests: _prefetch_robots_txt
+# ---------------------------------------------------------------------------
+
+
+class TestPrefetchRobotsTxt:
+    """_prefetch_robots_txt warms the robots.txt cache before the crawl loop."""
+
+    @staticmethod
+    def _make_counting_fetch():
+        """Return (fetch_fn, calls_list) where calls_list records every (url, sid) pair."""
+        calls: list[tuple[str, str]] = []
+
+        async def _fetch(url: str, sid: str):
+            calls.append((url, sid))
+            return MockResponse(status=200, body=b"", url=url)
+
+        return _fetch, calls
+
+    @pytest.mark.asyncio
+    async def test_prefetch_uses_allowed_domains_when_set(self):
+        fetch_fn, calls = self._make_counting_fetch()
+        spider = MockSpider(allowed_domains={"a.com", "b.com"}, robots_txt_obey=True)
+        engine = _make_engine(spider=spider)
+        engine._robots_manager = RobotsTxtManager(fetch_fn)
+
+        await engine._prefetch_robots_txt()
+
+        fetched_domains = {Request(url).domain for url, _ in calls}
+        assert fetched_domains == {"a.com", "b.com"}
+
+    @pytest.mark.asyncio
+    async def test_prefetch_falls_back_to_start_urls_when_no_allowed_domains(self):
+        fetch_fn, calls = self._make_counting_fetch()
+        spider = MockSpider(robots_txt_obey=True, start_urls=["https://example.com/page1"])
+        engine = _make_engine(spider=spider)
+        engine._robots_manager = RobotsTxtManager(fetch_fn)
+
+        await engine._prefetch_robots_txt()
+
+        assert len(calls) == 1
+        assert calls[0][0] == "https://example.com/robots.txt"
+
+    @pytest.mark.asyncio
+    async def test_prefetch_noop_when_robots_disabled(self):
+        fetch_fn, calls = self._make_counting_fetch()
+        spider = MockSpider(robots_txt_obey=False)
+        engine = _make_engine(spider=spider)
+
+        assert engine._robots_manager is None
+
+        await engine._prefetch_robots_txt()
+
+        assert calls == []
+
+    @pytest.mark.asyncio
+    async def test_prefetch_noop_when_start_urls_empty(self):
+        fetch_fn, calls = self._make_counting_fetch()
+        spider = MockSpider(robots_txt_obey=True, start_urls=[])
+        engine = _make_engine(spider=spider)
+        engine._robots_manager = RobotsTxtManager(fetch_fn)
+
+        await engine._prefetch_robots_txt()
+
+        assert calls == []
+
+    @pytest.mark.asyncio
+    async def test_prefetch_deduplicates_same_domain_in_start_urls(self):
+        fetch_fn, calls = self._make_counting_fetch()
+        spider = MockSpider(robots_txt_obey=True, start_urls=["https://example.com/a", "https://example.com/b"])
+        engine = _make_engine(spider=spider)
+        engine._robots_manager = RobotsTxtManager(fetch_fn)
+
+        await engine._prefetch_robots_txt()
+
+        # set of Request.domain values deduplicates to one task per domain
+        assert len(calls) == 1
+        assert calls[0][0] == "https://example.com/robots.txt"

From a134fdb8cce853f66713e115b81d195ec6e7b0c3 Mon Sep 17 00:00:00 2001
From: Abdullah <52079299+AbdullahY36@users.noreply.github.com>
Date: Sat, 4 Apr 2026 03:10:17 +0200
Subject: [PATCH 8/8] feat(spiders): enable robots.txt compliance by default

robots_txt_obey now defaults to True. Spiders must explicitly opt out
with robots_txt_obey = False rather than opt in, making ethical
crawling the default behaviour.

File: scrapling/spiders/spider.py
---
 scrapling/spiders/spider.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/scrapling/spiders/spider.py b/scrapling/spiders/spider.py
index 6aaa24f..52afcbd 100644
--- a/scrapling/spiders/spider.py
+++ b/scrapling/spiders/spider.py
@@ -73,7 +73,7 @@ class Spider(ABC):
     allowed_domains: Set[str] = set()
 
     # Robots.txt compliance
-    robots_txt_obey: bool = False
+    robots_txt_obey: bool = True
 
     # Concurrency settings
     concurrent_requests: int = 4