Networking API

The networking layer provides HttpClient (sync) and AsyncHttpClient (async) — both backed by the same HttpClientConfig and sharing the same politeness, retry, and resilience policies.

HttpClientConfig

Configuration for HttpClient behavior.

This config is expected to grow as policy modules are implemented.

Ethical note on robots.txt

respect_robots_txt is disabled by default to avoid breaking callers that crawl their own infrastructure or operate under explicit agreements. If you are crawling third-party public websites, you are strongly encouraged to enable it:

.. code-block:: python

HttpClientConfig(respect_robots_txt=True)

Respecting robots.txt is the long-established community norm for web crawlers, codified as an IETF Proposed Standard in RFC 9309 (2022). Academic and legal literature on web data collection treats compliance as a baseline ethical expectation. EU data-protection authorities have indicated that ignoring robots.txt can undermine the legitimate interest legal basis required for scraping personal data under GDPR.

Authentication patterns

+---------------------------------+--------------------------------------------------+--------------------+ | Mechanism | Config | Backends | +=================================+==================================================+====================+ | HTTP Basic Auth | auth=("user", "pass") | all | +---------------------------------+--------------------------------------------------+--------------------+ | HTTP Digest Auth | auth=HTTPDigestAuth("user", "pass") | requests only | +---------------------------------+--------------------------------------------------+--------------------+ | Bearer token / API key (header) | default_headers={"Authorization": "Bearer …"}| all | +---------------------------------+--------------------------------------------------+--------------------+ | API key in query string | default_params={"api_key": "…"} | all | +---------------------------------+--------------------------------------------------+--------------------+ | HMAC signing / OAuth tokens | Custom requests.auth.AuthBase via auth | requests only | +---------------------------------+--------------------------------------------------+--------------------+

curl-cffi only supports Basic Auth (auth as a tuple). Passing an AuthBase subclass with backend="curl-cffi" raises ValueError at client construction time.

Backend selection

The default backend ("requests") is sufficient for most targets. Use "curl-cffi" for Cloudflare-protected sites — it impersonates a real browser's TLS ClientHello (JA3/JA4), bypassing L1 (JS challenge) and L2 (TLS fingerprint) without a browser process:

.. code-block:: python

HttpClientConfig(backend="curl-cffi", impersonate="chrome136")

impersonate is required when backend="curl-cffi"; construction raises ValueError otherwise. Valid values are the members of curl_cffi.requests.BrowserType (e.g. "chrome136", "firefox147", "safari184"). Requires pip install ladon-crawl[cffi].

When curl-cffi is installed, HttpClientConfig checks impersonate against BrowserType at construction time. An unrecognised value emits a UserWarning but does not raise — forward-compatible strings (fingerprints added in a newer curl-cffi release) should not be blocked. When curl-cffi is not installed the check is skipped; an invalid value will raise ValueError at client construction time (CurlHttpClient / AsyncCurlHttpClient or the factory helpers).

Source code in src/ladon/networking/config.py

@dataclass(frozen=True)
class HttpClientConfig:
    """Configuration for HttpClient behavior.

    This config is expected to grow as policy modules are implemented.

    Ethical note on robots.txt
    --------------------------
    ``respect_robots_txt`` is disabled by default to avoid breaking callers
    that crawl their own infrastructure or operate under explicit agreements.
    **If you are crawling third-party public websites, you are strongly
    encouraged to enable it:**

    .. code-block:: python

        HttpClientConfig(respect_robots_txt=True)

    Respecting robots.txt is the long-established community norm for web
    crawlers, codified as an IETF Proposed Standard in RFC 9309 (2022).
    Academic and legal literature on web data collection treats compliance
    as a baseline ethical expectation.  EU data-protection authorities have
    indicated that ignoring robots.txt can undermine the *legitimate interest*
    legal basis required for scraping personal data under GDPR.

    Authentication patterns
    -----------------------
    +---------------------------------+--------------------------------------------------+--------------------+
    | Mechanism                       | Config                                           | Backends           |
    +=================================+==================================================+====================+
    | HTTP Basic Auth                 | ``auth=("user", "pass")``                        | all                |
    +---------------------------------+--------------------------------------------------+--------------------+
    | HTTP Digest Auth                | ``auth=HTTPDigestAuth("user", "pass")``          | requests only      |
    +---------------------------------+--------------------------------------------------+--------------------+
    | Bearer token / API key (header) | ``default_headers={"Authorization": "Bearer …"}``| all                |
    +---------------------------------+--------------------------------------------------+--------------------+
    | API key in query string         | ``default_params={"api_key": "…"}``              | all                |
    +---------------------------------+--------------------------------------------------+--------------------+
    | HMAC signing / OAuth tokens     | Custom ``requests.auth.AuthBase`` via ``auth``   | requests only      |
    +---------------------------------+--------------------------------------------------+--------------------+

    ``curl-cffi`` only supports Basic Auth (``auth`` as a tuple).  Passing an
    ``AuthBase`` subclass with ``backend="curl-cffi"`` raises ``ValueError``
    at client construction time.

    Backend selection
    -----------------
    The default backend (``"requests"``) is sufficient for most targets.
    Use ``"curl-cffi"`` for Cloudflare-protected sites — it impersonates a
    real browser's TLS ``ClientHello`` (JA3/JA4), bypassing L1 (JS challenge)
    and L2 (TLS fingerprint) without a browser process:

    .. code-block:: python

        HttpClientConfig(backend="curl-cffi", impersonate="chrome136")

    ``impersonate`` is required when ``backend="curl-cffi"``; construction
    raises ``ValueError`` otherwise.  Valid values are the members of
    ``curl_cffi.requests.BrowserType`` (e.g. ``"chrome136"``,
    ``"firefox147"``, ``"safari184"``).  Requires
    ``pip install ladon-crawl[cffi]``.

    When curl-cffi is installed, ``HttpClientConfig`` checks ``impersonate``
    against ``BrowserType`` at construction time.  An unrecognised value
    emits a ``UserWarning`` but does **not** raise — forward-compatible
    strings (fingerprints added in a newer curl-cffi release) should not
    be blocked.  When curl-cffi is not installed the check is skipped; an
    invalid value will raise ``ValueError`` at client construction time
    (``CurlHttpClient`` / ``AsyncCurlHttpClient`` or the factory helpers).
    """

    user_agent: str | None = None
    default_headers: Mapping[str, str] = field(default_factory=_default_headers)
    retries: int = 0
    verify_tls: bool = True
    connect_timeout_seconds: float | None = None
    read_timeout_seconds: float | None = None
    backoff_base_seconds: float = 0.0
    timeout_seconds: float = 30.0
    min_request_interval_seconds: float = 0.0
    # Threshold counts *call sequences*, not individual HTTP attempts.
    # With retries=2 and threshold=3, the circuit opens after 3 fully-exhausted
    # sequences (up to 9 individual HTTP failures).  See CircuitBreaker docstring.
    circuit_breaker_failure_threshold: int | None = None
    circuit_breaker_recovery_seconds: float = 60.0
    # Disabled by default; enable for any public-web crawl — see class docstring.
    respect_robots_txt: bool = False
    # HTTP status codes that trigger automatic retry with Retry-After respect.
    # Only GET/HEAD are auto-retried; POST/etc. receive the response as-is.
    retry_on_status: frozenset[int] = frozenset({429, 503})
    max_retry_after_seconds: float = 300.0
    # When True, applies full jitter to exponential backoff: sleep duration is
    # drawn uniformly from [0, cap] instead of always sleeping cap.  Reduces
    # thundering-herd when multiple crawlers restart simultaneously.
    backoff_jitter: bool = False
    # Proxy map passed verbatim to requests.Session.proxies.  Follows the
    # requests convention: {"http": "http://host:port", "https": "http://host:port"}.
    # Accepted schemes: http, https, socks4, socks4h, socks5, socks5h.
    # SOCKS proxies require requests[socks].
    proxies: Mapping[str, str] | None = None
    # Proxy rotation strategy.  Mutually exclusive with proxies.
    # HttpClient calls next_proxy() before each request attempt and
    # mark_failure() when a transport error or rate-limit response occurs.
    proxy_pool: "ProxyPool | None" = None
    # HTTP authentication passed verbatim to requests.Session.auth.
    # Use a (username, password) tuple for Basic Auth, or an AuthBase subclass
    # (HTTPDigestAuth, custom HMAC/OAuth token injectors) for other schemes.
    # Bearer tokens and static API keys belong in default_headers instead.
    auth: tuple[str, str] | AuthBase | None = None
    # Default query parameters merged into every request.  Follows the same
    # override contract as default_headers: per-request params take precedence
    # on key collision.  Useful for API keys passed as query string parameters.
    default_params: Mapping[str, str] | None = None
    # HTTP backend — "requests" (default) or "curl-cffi".  See class docstring.
    backend: Literal["requests", "curl-cffi"] = "requests"
    # Browser fingerprint for curl-cffi impersonation.  See class docstring.
    impersonate: str | None = None

    def __post_init__(self) -> None:
        if self.retries < 0:
            raise ValueError("retries must be >= 0")
        if self.backoff_base_seconds < 0:
            raise ValueError("backoff_base_seconds must be >= 0")
        if self.min_request_interval_seconds < 0:
            raise ValueError("min_request_interval_seconds must be >= 0")
        if (
            self.circuit_breaker_failure_threshold is not None
            and self.circuit_breaker_failure_threshold <= 0
        ):
            raise ValueError(
                "circuit_breaker_failure_threshold must be > 0 when provided"
            )
        if self.circuit_breaker_recovery_seconds <= 0:
            raise ValueError("circuit_breaker_recovery_seconds must be > 0")
        if self.max_retry_after_seconds <= 0:
            raise ValueError("max_retry_after_seconds must be > 0")
        if not all(100 <= s <= 599 for s in self.retry_on_status):
            raise ValueError(
                "retry_on_status must contain only valid HTTP status codes (100-599)"
            )

        has_connect_timeout = self.connect_timeout_seconds is not None
        has_read_timeout = self.read_timeout_seconds is not None
        if has_connect_timeout != has_read_timeout:
            raise ValueError(
                "connect_timeout_seconds and read_timeout_seconds "
                "must be set together"
            )
        if self.timeout_seconds <= 0:
            raise ValueError("timeout_seconds must be > 0")
        if (
            self.connect_timeout_seconds is not None
            and self.connect_timeout_seconds <= 0
        ):
            raise ValueError(
                "connect_timeout_seconds must be > 0 when provided"
            )
        if (
            self.read_timeout_seconds is not None
            and self.read_timeout_seconds <= 0
        ):
            raise ValueError("read_timeout_seconds must be > 0 when provided")

        # Freeze copied mappings to avoid post-init mutation side effects.
        object.__setattr__(
            self,
            "default_headers",
            MappingProxyType(dict(self.default_headers)),
        )
        if self.proxies is not None and self.proxy_pool is not None:
            raise ValueError(
                "proxies and proxy_pool are mutually exclusive; set only one"
            )
        if self.proxies is not None:
            validate_proxy(self.proxies)
            object.__setattr__(
                self,
                "proxies",
                MappingProxyType(dict(self.proxies)),
            )
        if self.backend not in {"requests", "curl-cffi"}:
            raise ValueError(
                f"backend must be 'requests' or 'curl-cffi', got {self.backend!r}"
            )
        if self.backend == "curl-cffi" and self.impersonate is None:
            raise ValueError(
                "impersonate is required when backend='curl-cffi'; "
                "e.g. HttpClientConfig(backend='curl-cffi', impersonate='chrome136')"
            )
        if self.backend == "curl-cffi" and self.impersonate is not None:
            valid = _get_cffi_valid_impersonate()
            if valid is not None and self.impersonate not in valid:
                warnings.warn(
                    f"Unknown impersonate target {self.impersonate!r}; "
                    "it will be rejected at client construction time. "
                    'Run `python -c "from curl_cffi.requests import '
                    'BrowserType; print([b.value for b in BrowserType])"`'
                    " for valid values.",
                    UserWarning,
                    stacklevel=3,
                )
        if isinstance(self.auth, tuple) and len(self.auth) != 2:
            raise ValueError("auth tuple must be (username, password)")
        if self.default_params is not None:
            object.__setattr__(
                self,
                "default_params",
                MappingProxyType(dict(self.default_params)),
            )

HttpClient

Bases: SyncPolicyBase

Core HTTP client interface (sync).

All outbound HTTP in Ladon must go through this client to ensure consistent politeness, resilience, and observability. Methods return a Result that contains either a value or an error plus request metadata.

Thread safety

HttpClient is not thread-safe. It is designed for the single-threaded, single-run crawler model. Do not share an instance across threads without external locking.

Source code in src/ladon/networking/client.py

class HttpClient(SyncPolicyBase):
    """Core HTTP client interface (sync).

    All outbound HTTP in Ladon must go through this client to ensure consistent
    politeness, resilience, and observability. Methods return a Result that
    contains either a value or an error plus request metadata.

    Thread safety
    -------------
    ``HttpClient`` is **not** thread-safe.  It is designed for the
    single-threaded, single-run crawler model.  Do not share an instance
    across threads without external locking.
    """

    def __init__(self, config: HttpClientConfig) -> None:
        """Create a new HttpClient.

        Args:
            config: Configuration for timeouts, headers, and policy settings.
        """
        super().__init__(config)
        self._session = requests.Session()
        if self._config.user_agent:
            self._session.headers["User-Agent"] = self._config.user_agent
        self._session.headers.update(self._config.default_headers)
        if self._config.proxies is not None:
            self._session.proxies.update(self._config.proxies)
        if self._config.auth is not None:
            self._session.auth = self._config.auth
        self._robots_cache = (
            RobotsCache(
                self._session,
                self._config.user_agent or "*",
                fetch_timeout=self._config.timeout_seconds,
                verify_tls=self._config.verify_tls,
            )
            if self._config.respect_robots_txt
            else None
        )

    def close(self) -> None:
        """Close the underlying session and release pooled connections."""
        self._session.close()

    @property
    def _proxies(self) -> MutableMapping[str, str]:
        return self._session.proxies  # type: ignore[no-any-return]

    def _is_transport_exception(self, exc: Exception) -> bool:
        """Return True for any requests library transport exception."""
        return isinstance(exc, requests.exceptions.RequestException)

    def _is_retryable_exception(self, method: str, exc: Exception) -> bool:
        """Return True for retryable transport errors."""
        if method.upper() not in {"GET", "HEAD"}:
            return False
        return isinstance(
            exc,
            (requests.exceptions.Timeout, requests.exceptions.ConnectionError),
        )

    def _handle_request_exception(
        self,
        method: str,
        request_url: str,
        e: Exception,
        context: Mapping[str, Any] | None,
        attempts: int,
        timeout: float | tuple[float, float],
    ) -> Result[Any, Exception]:
        """Map requests exceptions to Ladon errors."""
        response = getattr(e, "response", None)
        meta = self._build_meta(
            method,
            request_url,
            response,
            context,
            attempts,
            timeout,
            final_error=type(e).__name__,
        )

        if isinstance(e, requests.exceptions.Timeout):
            return Err(RequestTimeoutError(str(e)), meta=meta)

        if isinstance(e, requests.exceptions.ConnectionError):
            return Err(TransientNetworkError(str(e)), meta=meta)

        return Err(HttpClientError(str(e)), meta=meta)

    def get(
        self,
        url: str,
        *,
        headers: Mapping[str, str] | None = None,
        params: Mapping[str, str] | None = None,
        timeout: float | None = None,
        allow_redirects: bool = True,
        context: Mapping[str, Any] | None = None,
    ) -> Result[bytes, Exception]:
        """Perform an HTTP GET request.

        Args:
            url: Absolute URL to request.
            headers: Optional per-request headers merged with defaults.
            params: Optional query parameters.
            timeout: Override timeout in seconds for this request.
            allow_redirects: Whether redirects should be followed.
            context: Optional caller context for logging/tracing.

        Returns:
            Result containing response bytes on success, or an error on failure.
        """
        resolved_timeout = self._get_timeout(timeout)
        return self._request(
            method="GET",
            url=url,
            context=context,
            timeout=resolved_timeout,
            request_fn=lambda: self._session.get(
                url,
                headers=headers,
                params=self._merge_params(params),
                timeout=resolved_timeout,
                allow_redirects=allow_redirects,
                verify=self._config.verify_tls,
            ),
            value_builder=self._content_value,
        )

    def head(
        self,
        url: str,
        *,
        headers: Mapping[str, str] | None = None,
        params: Mapping[str, str] | None = None,
        timeout: float | None = None,
        allow_redirects: bool = True,
        context: Mapping[str, Any] | None = None,
    ) -> Result[Mapping[str, Any], Exception]:
        """Perform an HTTP HEAD request.

        Args:
            url: Absolute URL to request.
            headers: Optional per-request headers merged with defaults.
            params: Optional query parameters.
            timeout: Override timeout in seconds for this request.
            allow_redirects: Whether redirects should be followed.
            context: Optional caller context for logging/tracing.

        Returns:
            Result containing response metadata on success, or an error on
            failure.
        """
        resolved_timeout = self._get_timeout(timeout)
        return self._request(
            method="HEAD",
            url=url,
            context=context,
            timeout=resolved_timeout,
            request_fn=lambda: self._session.head(
                url,
                headers=headers,
                params=self._merge_params(params),
                timeout=resolved_timeout,
                allow_redirects=allow_redirects,
                verify=self._config.verify_tls,
            ),
            value_builder=self._headers_value,
        )

    def post(
        self,
        url: str,
        *,
        headers: Mapping[str, str] | None = None,
        params: Mapping[str, str] | None = None,
        data: Any | None = None,
        json: Any | None = None,
        timeout: float | None = None,
        allow_redirects: bool = True,
        context: Mapping[str, Any] | None = None,
    ) -> Result[bytes, Exception]:
        """Perform an HTTP POST request.

        Args:
            url: Absolute URL to request.
            headers: Optional per-request headers merged with defaults.
            params: Optional query parameters.
            data: Optional form/body payload.
            json: Optional JSON payload (mutually exclusive with data).
            timeout: Override timeout in seconds for this request.
            allow_redirects: Whether redirects should be followed.
            context: Optional caller context for logging/tracing.

        Returns:
            Result containing response bytes on success, or an error on failure.
        """
        resolved_timeout = self._get_timeout(timeout)
        return self._request(
            method="POST",
            url=url,
            context=context,
            timeout=resolved_timeout,
            request_fn=lambda: self._session.post(
                url,
                headers=headers,
                params=self._merge_params(params),
                data=data,
                json=json,
                timeout=resolved_timeout,
                allow_redirects=allow_redirects,
                verify=self._config.verify_tls,
            ),
            value_builder=self._content_value,
        )

    def download(
        self,
        url: str,
        *,
        headers: Mapping[str, str] | None = None,
        params: Mapping[str, str] | None = None,
        timeout: float | None = None,
        allow_redirects: bool = True,
        context: Mapping[str, Any] | None = None,
    ) -> Result[requests.Response, Exception]:
        """Stream a download request.

        Args:
            url: Absolute URL to request.
            headers: Optional per-request headers merged with defaults.
            params: Optional query parameters.
            timeout: Override timeout in seconds for this request.
            allow_redirects: Whether redirects should be followed.
            context: Optional caller context for logging/tracing.

        Returns:
            Result containing a stream/handle or download descriptor on success,
            or an error on failure.
        """
        resolved_timeout = self._get_timeout(timeout)
        return self._request(
            method="GET",
            url=url,
            context=context,
            timeout=resolved_timeout,
            request_fn=lambda: self._session.get(
                url,
                headers=headers,
                params=self._merge_params(params),
                timeout=resolved_timeout,
                allow_redirects=allow_redirects,
                stream=True,
                verify=self._config.verify_tls,
            ),
            value_builder=self._response_value,
        )

close()

Close the underlying session and release pooled connections.

Source code in src/ladon/networking/client.py

def close(self) -> None:
    """Close the underlying session and release pooled connections."""
    self._session.close()

get(url, *, headers=None, params=None, timeout=None, allow_redirects=True, context=None)

Perform an HTTP GET request.

Parameters:

Name	Type	Description	Default
url	str	Absolute URL to request.	required
headers	Mapping[str, str] | None	Optional per-request headers merged with defaults.	None
params	Mapping[str, str] | None	Optional query parameters.	None
timeout	float | None	Override timeout in seconds for this request.	None
allow_redirects	bool	Whether redirects should be followed.	True
context	Mapping[str, Any] | None	Optional caller context for logging/tracing.	None

Returns:

Type	Description
Result[bytes, Exception]	Result containing response bytes on success, or an error on failure.

Source code in src/ladon/networking/client.py

def get(
    self,
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
    timeout: float | None = None,
    allow_redirects: bool = True,
    context: Mapping[str, Any] | None = None,
) -> Result[bytes, Exception]:
    """Perform an HTTP GET request.

    Args:
        url: Absolute URL to request.
        headers: Optional per-request headers merged with defaults.
        params: Optional query parameters.
        timeout: Override timeout in seconds for this request.
        allow_redirects: Whether redirects should be followed.
        context: Optional caller context for logging/tracing.

    Returns:
        Result containing response bytes on success, or an error on failure.
    """
    resolved_timeout = self._get_timeout(timeout)
    return self._request(
        method="GET",
        url=url,
        context=context,
        timeout=resolved_timeout,
        request_fn=lambda: self._session.get(
            url,
            headers=headers,
            params=self._merge_params(params),
            timeout=resolved_timeout,
            allow_redirects=allow_redirects,
            verify=self._config.verify_tls,
        ),
        value_builder=self._content_value,
    )

head(url, *, headers=None, params=None, timeout=None, allow_redirects=True, context=None)

Perform an HTTP HEAD request.

Parameters:

Name	Type	Description	Default
url	str	Absolute URL to request.	required
headers	Mapping[str, str] | None	Optional per-request headers merged with defaults.	None
params	Mapping[str, str] | None	Optional query parameters.	None
timeout	float | None	Override timeout in seconds for this request.	None
allow_redirects	bool	Whether redirects should be followed.	True
context	Mapping[str, Any] | None	Optional caller context for logging/tracing.	None

Returns:

Type	Description
Result[Mapping[str, Any], Exception]	Result containing response metadata on success, or an error on
Result[Mapping[str, Any], Exception]	failure.

Source code in src/ladon/networking/client.py

def head(
    self,
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
    timeout: float | None = None,
    allow_redirects: bool = True,
    context: Mapping[str, Any] | None = None,
) -> Result[Mapping[str, Any], Exception]:
    """Perform an HTTP HEAD request.

    Args:
        url: Absolute URL to request.
        headers: Optional per-request headers merged with defaults.
        params: Optional query parameters.
        timeout: Override timeout in seconds for this request.
        allow_redirects: Whether redirects should be followed.
        context: Optional caller context for logging/tracing.

    Returns:
        Result containing response metadata on success, or an error on
        failure.
    """
    resolved_timeout = self._get_timeout(timeout)
    return self._request(
        method="HEAD",
        url=url,
        context=context,
        timeout=resolved_timeout,
        request_fn=lambda: self._session.head(
            url,
            headers=headers,
            params=self._merge_params(params),
            timeout=resolved_timeout,
            allow_redirects=allow_redirects,
            verify=self._config.verify_tls,
        ),
        value_builder=self._headers_value,
    )

post(url, *, headers=None, params=None, data=None, json=None, timeout=None, allow_redirects=True, context=None)

Perform an HTTP POST request.

Parameters:

Name	Type	Description	Default
url	str	Absolute URL to request.	required
headers	Mapping[str, str] | None	Optional per-request headers merged with defaults.	None
params	Mapping[str, str] | None	Optional query parameters.	None
data	Any | None	Optional form/body payload.	None
json	Any | None	Optional JSON payload (mutually exclusive with data).	None
timeout	float | None	Override timeout in seconds for this request.	None
allow_redirects	bool	Whether redirects should be followed.	True
context	Mapping[str, Any] | None	Optional caller context for logging/tracing.	None

Returns:

Type	Description
Result[bytes, Exception]	Result containing response bytes on success, or an error on failure.

Source code in src/ladon/networking/client.py

def post(
    self,
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
    data: Any | None = None,
    json: Any | None = None,
    timeout: float | None = None,
    allow_redirects: bool = True,
    context: Mapping[str, Any] | None = None,
) -> Result[bytes, Exception]:
    """Perform an HTTP POST request.

    Args:
        url: Absolute URL to request.
        headers: Optional per-request headers merged with defaults.
        params: Optional query parameters.
        data: Optional form/body payload.
        json: Optional JSON payload (mutually exclusive with data).
        timeout: Override timeout in seconds for this request.
        allow_redirects: Whether redirects should be followed.
        context: Optional caller context for logging/tracing.

    Returns:
        Result containing response bytes on success, or an error on failure.
    """
    resolved_timeout = self._get_timeout(timeout)
    return self._request(
        method="POST",
        url=url,
        context=context,
        timeout=resolved_timeout,
        request_fn=lambda: self._session.post(
            url,
            headers=headers,
            params=self._merge_params(params),
            data=data,
            json=json,
            timeout=resolved_timeout,
            allow_redirects=allow_redirects,
            verify=self._config.verify_tls,
        ),
        value_builder=self._content_value,
    )

download(url, *, headers=None, params=None, timeout=None, allow_redirects=True, context=None)

Stream a download request.

Parameters:

Name	Type	Description	Default
url	str	Absolute URL to request.	required
headers	Mapping[str, str] | None	Optional per-request headers merged with defaults.	None
params	Mapping[str, str] | None	Optional query parameters.	None
timeout	float | None	Override timeout in seconds for this request.	None
allow_redirects	bool	Whether redirects should be followed.	True
context	Mapping[str, Any] | None	Optional caller context for logging/tracing.	None

Returns:

Type	Description
Result[Response, Exception]	Result containing a stream/handle or download descriptor on success,
Result[Response, Exception]	or an error on failure.

Source code in src/ladon/networking/client.py

def download(
    self,
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
    timeout: float | None = None,
    allow_redirects: bool = True,
    context: Mapping[str, Any] | None = None,
) -> Result[requests.Response, Exception]:
    """Stream a download request.

    Args:
        url: Absolute URL to request.
        headers: Optional per-request headers merged with defaults.
        params: Optional query parameters.
        timeout: Override timeout in seconds for this request.
        allow_redirects: Whether redirects should be followed.
        context: Optional caller context for logging/tracing.

    Returns:
        Result containing a stream/handle or download descriptor on success,
        or an error on failure.
    """
    resolved_timeout = self._get_timeout(timeout)
    return self._request(
        method="GET",
        url=url,
        context=context,
        timeout=resolved_timeout,
        request_fn=lambda: self._session.get(
            url,
            headers=headers,
            params=self._merge_params(params),
            timeout=resolved_timeout,
            allow_redirects=allow_redirects,
            stream=True,
            verify=self._config.verify_tls,
        ),
        value_builder=self._response_value,
    )

AsyncHttpClient

AsyncHttpClient is the async counterpart to HttpClient. It uses httpx as its backend and implements the same retry, backoff, circuit breaker, proxy rotation, and auth policies. Use it with async_run_crawl() and AsyncCrawlPlugin.

respect_robots_txt is not yet supported

Passing respect_robots_txt=True to HttpClientConfig raises NotImplementedError at AsyncHttpClient construction time. The default (False) is safe for all current use cases.

Bases: AsyncPolicyBase

Core async HTTP client (httpx backend).

All outbound async HTTP in Ladon must go through this client to ensure consistent politeness, resilience, and observability. Methods return a Result that contains either a value or an error plus request metadata.

This is the only Ladon module that imports httpx. Adapters must not import httpx directly.

Robots.txt enforcement is not yet supported; passing respect_robots_txt=True raises NotImplementedError at construction time.

Auth is limited to (username, password) tuples (HTTP Basic Auth). Passing a requests.auth.AuthBase subclass raises NotImplementedError at construction time; use an httpx.Auth subclass for other schemes.

Concurrent safety

Safe for concurrent use within a single asyncio event loop. Do not share an instance across threads.

Source code in src/ladon/networking/async_client.py

class AsyncHttpClient(AsyncPolicyBase):
    """Core async HTTP client (httpx backend).

    All outbound async HTTP in Ladon must go through this client to ensure
    consistent politeness, resilience, and observability. Methods return a
    ``Result`` that contains either a value or an error plus request metadata.

    This is the only Ladon module that imports ``httpx``. Adapters must not
    import ``httpx`` directly.

    Robots.txt enforcement is not yet supported; passing
    ``respect_robots_txt=True`` raises ``NotImplementedError`` at
    construction time.

    Auth is limited to ``(username, password)`` tuples (HTTP Basic Auth).
    Passing a ``requests.auth.AuthBase`` subclass raises ``NotImplementedError``
    at construction time; use an ``httpx.Auth`` subclass for other schemes.

    Concurrent safety
    -----------------
    Safe for concurrent use within a single asyncio event loop. Do not share
    an instance across threads.
    """

    def __init__(self, config: HttpClientConfig) -> None:
        if config.respect_robots_txt:
            raise NotImplementedError(
                "respect_robots_txt is not supported by AsyncHttpClient; "
                "async robots enforcement is deferred to a future release"
            )
        if config.auth is not None and not isinstance(config.auth, tuple):
            raise NotImplementedError(
                "AsyncHttpClient only supports (username, password) tuple auth; "
                "for other schemes use an httpx.Auth subclass directly"
            )
        super().__init__(config)

        headers: dict[str, str] = {}
        if config.user_agent:
            headers["User-Agent"] = config.user_agent
        headers.update(config.default_headers)
        self._base_headers = headers
        self._auth: tuple[str, str] | None = (
            config.auth if isinstance(config.auth, tuple) else None
        )

        mounts = (
            self._to_httpx_proxies(config.proxies)
            if config.proxies is not None
            else None
        )
        self._http = httpx.AsyncClient(
            headers=headers,
            mounts=mounts,
            auth=self._auth,
            verify=config.verify_tls,
        )

    async def aclose(self) -> None:
        """Close the underlying httpx client and release connections."""
        await self._http.aclose()

    # ------------------------------------------------------------------
    # httpx API delta converters — the blast-radius boundary
    # ------------------------------------------------------------------

    @staticmethod
    def _to_httpx_proxies(
        proxies: Mapping[str, str],
    ) -> dict[str, httpx.AsyncHTTPTransport]:
        """Convert a requests-style proxy dict to an httpx mounts dict.

        requests: ``{"https": "http://proxy:8080"}``
        httpx:    ``{"https://": AsyncHTTPTransport(proxy=Proxy("http://proxy:8080"))}``

        A missed conversion means proxies silently do not apply.
        """
        return {
            (k if k.endswith("://") else k + "://"): httpx.AsyncHTTPTransport(
                proxy=httpx.Proxy(v)
            )
            for k, v in proxies.items()
        }

    def _to_httpx_timeout(self, override: float | None) -> httpx.Timeout:
        """Build an ``httpx.Timeout`` from config, with optional scalar override."""
        if override is not None:
            if override <= 0:
                raise ValueError("timeout override must be > 0 when provided")
            return httpx.Timeout(override)
        if (
            self._config.connect_timeout_seconds is not None
            and self._config.read_timeout_seconds is not None
        ):
            return httpx.Timeout(
                connect=self._config.connect_timeout_seconds,
                read=self._config.read_timeout_seconds,
                write=None,
                pool=None,
            )
        return httpx.Timeout(self._config.timeout_seconds)

    def _build_meta(
        self,
        method: str,
        request_url: str,
        response: httpx.Response | None,
        context: Mapping[str, Any] | None,
        attempts: int,
        timeout: Any,
        final_error: str | None = None,
    ) -> dict[str, Any]:
        meta: dict[str, Any] = {}
        meta["method"] = method
        meta["url"] = request_url
        meta["attempts"] = attempts
        meta["timeout_s"] = timeout
        if context:
            context_dict = dict(context)
            meta["context"] = context_dict
            for key, value in context_dict.items():
                meta.setdefault(key, value)
        if response is not None:
            meta["status_code"] = response.status_code
            final_url = str(response.url)  # httpx.URL → str
            meta["final_url"] = final_url
            if final_url != request_url:
                meta["redirected"] = True
            meta["reason"] = (
                response.reason_phrase
            )  # httpx: reason_phrase not reason
            try:
                meta["elapsed_s"] = response.elapsed.total_seconds()
            except AttributeError:
                pass
        if final_error is not None:
            meta["final_error"] = final_error
        return meta

    # ------------------------------------------------------------------
    # Abstract method implementations
    # ------------------------------------------------------------------

    def _is_transport_exception(self, exc: Exception) -> bool:
        """Return True for any httpx transport exception."""
        return isinstance(exc, httpx.RequestError)

    def _is_retryable_exception(self, method: str, exc: Exception) -> bool:
        if method.upper() not in {"GET", "HEAD"}:
            return False
        return isinstance(exc, (httpx.TimeoutException, httpx.ConnectError))

    def _client_for_proxy(
        self, proxy: Mapping[str, str] | None
    ) -> httpx.AsyncClient:
        """Return the httpx client to use for this proxy.

        When proxy pool rotation is active a fresh client is created per
        attempt; the caller is responsible for closing it (``client is not
        self._http`` is the sentinel).  Without a pool the shared
        ``self._http`` is returned for connection-pool efficiency.
        """
        if self._config.proxy_pool is None:
            return self._http
        mounts = self._to_httpx_proxies(proxy) if proxy is not None else None
        return httpx.AsyncClient(
            headers=self._base_headers,
            mounts=mounts,
            auth=self._auth,
            verify=self._config.verify_tls,
        )

    async def _execute_attempt(
        self,
        request_fn: Any,
        proxy: Mapping[str, str] | None,
    ) -> Any:
        client = self._client_for_proxy(proxy)
        try:
            return await request_fn(client)
        finally:
            if client is not self._http:
                await client.aclose()

    def _handle_request_exception(
        self,
        method: str,
        request_url: str,
        e: Exception,
        context: Mapping[str, Any] | None,
        attempts: int,
        timeout: Any,
    ) -> Result[Any, Exception]:
        """Map httpx exceptions to Ladon errors."""
        meta = self._build_meta(
            method,
            request_url,
            None,
            context,
            attempts,
            timeout,
            final_error=type(e).__name__,
        )
        if isinstance(e, httpx.TimeoutException):
            return Err(RequestTimeoutError(str(e)), meta=meta)
        if isinstance(
            e,
            (
                httpx.ConnectError,
                httpx.NetworkError,
                httpx.RemoteProtocolError,
                httpx.LocalProtocolError,
            ),
        ):
            return Err(TransientNetworkError(str(e)), meta=meta)
        return Err(HttpClientError(str(e)), meta=meta)

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    async def get(
        self,
        url: str,
        *,
        headers: Mapping[str, str] | None = None,
        params: Mapping[str, str] | None = None,
        timeout: float | None = None,
        allow_redirects: bool = True,
        context: Mapping[str, Any] | None = None,
    ) -> Result[bytes, Exception]:
        """Perform an async HTTP GET request.

        Returns:
            Result containing response bytes on success, or an error on failure.
        """
        timeout_obj = self._to_httpx_timeout(timeout)
        merged_params = self._merge_params(params)

        async def _fn(client: httpx.AsyncClient) -> httpx.Response:
            return await client.get(
                url,
                headers=headers,
                params=merged_params,
                timeout=timeout_obj,
                follow_redirects=allow_redirects,
            )

        return await self._request(
            method="GET",
            url=url,
            context=context,
            timeout=timeout_obj,
            request_fn=_fn,
            value_builder=lambda r: r.content,
        )

    async def head(
        self,
        url: str,
        *,
        headers: Mapping[str, str] | None = None,
        params: Mapping[str, str] | None = None,
        timeout: float | None = None,
        allow_redirects: bool = True,
        context: Mapping[str, Any] | None = None,
    ) -> Result[Mapping[str, Any], Exception]:
        """Perform an async HTTP HEAD request.

        Returns:
            Result containing response headers on success, or an error on failure.
        """
        timeout_obj = self._to_httpx_timeout(timeout)
        merged_params = self._merge_params(params)

        async def _fn(client: httpx.AsyncClient) -> httpx.Response:
            return await client.head(
                url,
                headers=headers,
                params=merged_params,
                timeout=timeout_obj,
                follow_redirects=allow_redirects,
            )

        return await self._request(
            method="HEAD",
            url=url,
            context=context,
            timeout=timeout_obj,
            request_fn=_fn,
            value_builder=lambda r: dict(r.headers),
        )

    async def post(
        self,
        url: str,
        *,
        headers: Mapping[str, str] | None = None,
        params: Mapping[str, str] | None = None,
        data: Any | None = None,
        json: Any | None = None,
        timeout: float | None = None,
        allow_redirects: bool = True,
        context: Mapping[str, Any] | None = None,
    ) -> Result[bytes, Exception]:
        """Perform an async HTTP POST request.

        Returns:
            Result containing response bytes on success, or an error on failure.
        """
        timeout_obj = self._to_httpx_timeout(timeout)
        merged_params = self._merge_params(params)

        async def _fn(client: httpx.AsyncClient) -> httpx.Response:
            return await client.post(
                url,
                headers=headers,
                params=merged_params,
                data=data,
                json=json,
                timeout=timeout_obj,
                follow_redirects=allow_redirects,
            )

        return await self._request(
            method="POST",
            url=url,
            context=context,
            timeout=timeout_obj,
            request_fn=_fn,
            value_builder=lambda r: r.content,
        )

    async def download(
        self,
        url: str,
        *,
        headers: Mapping[str, str] | None = None,
        params: Mapping[str, str] | None = None,
        timeout: float | None = None,
        allow_redirects: bool = True,
        context: Mapping[str, Any] | None = None,
    ) -> Result[httpx.Response, Exception]:
        """Perform an async download (GET, returns the full httpx.Response).

        Returns:
            Result containing the ``httpx.Response`` on success, or an error
            on failure.
        """
        timeout_obj = self._to_httpx_timeout(timeout)
        merged_params = self._merge_params(params)

        async def _fn(client: httpx.AsyncClient) -> httpx.Response:
            return await client.get(
                url,
                headers=headers,
                params=merged_params,
                timeout=timeout_obj,
                follow_redirects=allow_redirects,
            )

        return await self._request(
            method="GET",
            url=url,
            context=context,
            timeout=timeout_obj,
            request_fn=_fn,
            value_builder=lambda r: r,
        )

aclose() async

Close the underlying httpx client and release connections.

Source code in src/ladon/networking/async_client.py

async def aclose(self) -> None:
    """Close the underlying httpx client and release connections."""
    await self._http.aclose()

get(url, *, headers=None, params=None, timeout=None, allow_redirects=True, context=None) async

Perform an async HTTP GET request.

Returns:

Type	Description
Result[bytes, Exception]	Result containing response bytes on success, or an error on failure.

Source code in src/ladon/networking/async_client.py

async def get(
    self,
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
    timeout: float | None = None,
    allow_redirects: bool = True,
    context: Mapping[str, Any] | None = None,
) -> Result[bytes, Exception]:
    """Perform an async HTTP GET request.

    Returns:
        Result containing response bytes on success, or an error on failure.
    """
    timeout_obj = self._to_httpx_timeout(timeout)
    merged_params = self._merge_params(params)

    async def _fn(client: httpx.AsyncClient) -> httpx.Response:
        return await client.get(
            url,
            headers=headers,
            params=merged_params,
            timeout=timeout_obj,
            follow_redirects=allow_redirects,
        )

    return await self._request(
        method="GET",
        url=url,
        context=context,
        timeout=timeout_obj,
        request_fn=_fn,
        value_builder=lambda r: r.content,
    )

head(url, *, headers=None, params=None, timeout=None, allow_redirects=True, context=None) async

Perform an async HTTP HEAD request.

Returns:

Type	Description
Result[Mapping[str, Any], Exception]	Result containing response headers on success, or an error on failure.

Source code in src/ladon/networking/async_client.py

async def head(
    self,
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
    timeout: float | None = None,
    allow_redirects: bool = True,
    context: Mapping[str, Any] | None = None,
) -> Result[Mapping[str, Any], Exception]:
    """Perform an async HTTP HEAD request.

    Returns:
        Result containing response headers on success, or an error on failure.
    """
    timeout_obj = self._to_httpx_timeout(timeout)
    merged_params = self._merge_params(params)

    async def _fn(client: httpx.AsyncClient) -> httpx.Response:
        return await client.head(
            url,
            headers=headers,
            params=merged_params,
            timeout=timeout_obj,
            follow_redirects=allow_redirects,
        )

    return await self._request(
        method="HEAD",
        url=url,
        context=context,
        timeout=timeout_obj,
        request_fn=_fn,
        value_builder=lambda r: dict(r.headers),
    )

post(url, *, headers=None, params=None, data=None, json=None, timeout=None, allow_redirects=True, context=None) async

Perform an async HTTP POST request.

Returns:

Type	Description
Result[bytes, Exception]	Result containing response bytes on success, or an error on failure.

Source code in src/ladon/networking/async_client.py

async def post(
    self,
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
    data: Any | None = None,
    json: Any | None = None,
    timeout: float | None = None,
    allow_redirects: bool = True,
    context: Mapping[str, Any] | None = None,
) -> Result[bytes, Exception]:
    """Perform an async HTTP POST request.

    Returns:
        Result containing response bytes on success, or an error on failure.
    """
    timeout_obj = self._to_httpx_timeout(timeout)
    merged_params = self._merge_params(params)

    async def _fn(client: httpx.AsyncClient) -> httpx.Response:
        return await client.post(
            url,
            headers=headers,
            params=merged_params,
            data=data,
            json=json,
            timeout=timeout_obj,
            follow_redirects=allow_redirects,
        )

    return await self._request(
        method="POST",
        url=url,
        context=context,
        timeout=timeout_obj,
        request_fn=_fn,
        value_builder=lambda r: r.content,
    )

download(url, *, headers=None, params=None, timeout=None, allow_redirects=True, context=None) async

Perform an async download (GET, returns the full httpx.Response).

Returns:

Type	Description
Result[Response, Exception]	Result containing the httpx.Response on success, or an error
Result[Response, Exception]	on failure.

Source code in src/ladon/networking/async_client.py

async def download(
    self,
    url: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: Mapping[str, str] | None = None,
    timeout: float | None = None,
    allow_redirects: bool = True,
    context: Mapping[str, Any] | None = None,
) -> Result[httpx.Response, Exception]:
    """Perform an async download (GET, returns the full httpx.Response).

    Returns:
        Result containing the ``httpx.Response`` on success, or an error
        on failure.
    """
    timeout_obj = self._to_httpx_timeout(timeout)
    merged_params = self._merge_params(params)

    async def _fn(client: httpx.AsyncClient) -> httpx.Response:
        return await client.get(
            url,
            headers=headers,
            params=merged_params,
            timeout=timeout_obj,
            follow_redirects=allow_redirects,
        )

    return await self._request(
        method="GET",
        url=url,
        context=context,
        timeout=timeout_obj,
        request_fn=_fn,
        value_builder=lambda r: r,
    )

Result type

Core data types for the HttpClient interface.

This module defines the Result container and configuration objects used by the networking layer. These types are intentionally small and stable because they anchor adapter contracts and observability metadata.

Result dataclass

Bases: Generic[T, E]

Result type returned by HttpClient operations.

A Result carries either a value or an error, plus a metadata mapping that captures request context (latency, retries, status, etc.).

Source code in src/ladon/networking/types.py

@dataclass(frozen=True)
class Result(Generic[T, E]):
    """Result type returned by HttpClient operations.

    A Result carries either a value or an error, plus a metadata mapping that
    captures request context (latency, retries, status, etc.).
    """

    value: T | None
    error: E | None
    meta: Meta = field(default_factory=_default_meta)

    @property
    def ok(self) -> bool:
        """Return True when the Result contains a value, not an error."""

        return self.error is None

ok property

Return True when the Result contains a value, not an error.

Ok(value, meta=None)

Construct a successful Result with optional metadata.

Source code in src/ladon/networking/types.py

def Ok(value: T, meta: Meta | None = None) -> Result[T, HttpClientError]:
    """Construct a successful Result with optional metadata."""

    return Result(value=value, error=None, meta=meta or {})

Err(error, meta=None)

Construct a failed Result with optional metadata.

Source code in src/ladon/networking/types.py

def Err(
    error: HttpClientError, meta: Meta | None = None
) -> Result[Any, HttpClientError]:
    """Construct a failed Result with optional metadata."""

    return Result(value=None, error=error, meta=meta or {})

Error types

Error types for the core HttpClient interface.

HttpClientError

Bases: Exception

Base exception for HTTP client failures.

Source code in src/ladon/networking/errors.py

class HttpClientError(Exception):
    """Base exception for HTTP client failures."""

CircuitOpenError

Bases: HttpClientError

Raised when the circuit breaker blocks a request.

Attributes:

Name	Type	Description
host		The host (netloc) whose circuit is open.

Source code in src/ladon/networking/errors.py

class CircuitOpenError(HttpClientError):
    """Raised when the circuit breaker blocks a request.

    Attributes:
        host: The host (``netloc``) whose circuit is open.
    """

    def __init__(self, host: str) -> None:
        super().__init__(f"circuit open for {host}")
        self.host = host

RobotsBlockedError

Bases: HttpClientError

Raised when robots.txt disallows a request.

Only raised when HttpClientConfig.respect_robots_txt is True. The disallowed URL is included in error.args[0].

Source code in src/ladon/networking/errors.py

class RobotsBlockedError(HttpClientError):
    """Raised when ``robots.txt`` disallows a request.

    Only raised when ``HttpClientConfig.respect_robots_txt`` is ``True``.
    The disallowed URL is included in ``error.args[0]``.
    """

RequestTimeoutError

Bases: HttpClientError

Raised when a request exceeds a configured timeout.

Source code in src/ladon/networking/errors.py

class RequestTimeoutError(HttpClientError):
    """Raised when a request exceeds a configured timeout."""

TransientNetworkError

Bases: HttpClientError

Raised for connection-level transport failures (e.g. ConnectionError, DNS resolution failure).

Ladon retries these internally; by the time this error reaches the caller all configured retries are exhausted. Do not retry externally on this error — the internal retry budget has already been spent.

Source code in src/ladon/networking/errors.py

class TransientNetworkError(HttpClientError):
    """Raised for connection-level transport failures (e.g. ConnectionError,
    DNS resolution failure).

    Ladon retries these internally; by the time this error reaches the caller
    all configured retries are exhausted.  Do not retry externally on this
    error — the internal retry budget has already been spent.
    """

RateLimitedError

Bases: HttpClientError

Raised when all retries are exhausted due to HTTP-level rate limiting.

Returned when the server responds with a status code in HttpClientConfig.retry_on_status (default: 429, 503) and the retry budget in HttpClientConfig.retries is exhausted.

Attributes:

Name	Type	Description
status_code		The HTTP status code that triggered rate limiting.
retry_after		The Retry-After delay in seconds parsed from the final blocked response, or None if the header was absent or unparseable.

Source code in src/ladon/networking/errors.py

class RateLimitedError(HttpClientError):
    """Raised when all retries are exhausted due to HTTP-level rate limiting.

    Returned when the server responds with a status code in
    ``HttpClientConfig.retry_on_status`` (default: 429, 503) and the retry
    budget in ``HttpClientConfig.retries`` is exhausted.

    Attributes:
        status_code: The HTTP status code that triggered rate limiting.
        retry_after: The ``Retry-After`` delay in seconds parsed from the
            final blocked response, or ``None`` if the header was absent or
            unparseable.
    """

    def __init__(self, status_code: int, retry_after: float | None) -> None:
        msg = f"rate limited by server (HTTP {status_code})"
        if retry_after is not None:
            msg += f"; Retry-After: {retry_after:.1f}s"
        super().__init__(msg)
        self.status_code = status_code
        self.retry_after = retry_after

RetryableHttpError

Bases: TransientNetworkError

Deprecated alias for TransientNetworkError. Removed in v0.1.0.

Use TransientNetworkError instead.

Source code in src/ladon/networking/errors.py

class RetryableHttpError(TransientNetworkError):
    """Deprecated alias for ``TransientNetworkError``. Removed in v0.1.0.

    Use ``TransientNetworkError`` instead.
    """

    def __init__(self, *args: object) -> None:
        import warnings

        warnings.warn(
            "RetryableHttpError is deprecated and will be removed in v0.1.0. "
            "Use TransientNetworkError instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        super().__init__(*args)