Scrapy Zyte API

https://scrapy-zyte-api.readthedocs.io
Admin
Scrapy Zyte API is a Scrapy plugin that enables seamless integration with Zyte API for web scraping,...
Tokens:36,501
Snippets:414
Trust Score:8.3
Update:2 months ago
Show doc for...
Tokens:
Raw
### Install scrapy-zyte-api

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/setup.rst.txt

Commands to install the base package and optional extras for specific features.

```shell
pip install scrapy-zyte-api
```

```shell
pip install scrapy-zyte-api[provider]
```

```shell
pip install scrapy-zyte-api[x402]
```

```shell
pip install scrapy-zyte-api[provider,x402]
```

--------------------------------

### Manually configure Scrapy components

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/setup.rst.txt

Manual setup of downloader handlers, middlewares, and the request fingerprinter.

```python
DOWNLOAD_HANDLERS = {
    "http": "scrapy_zyte_api.ScrapyZyteAPIDownloadHandler",
    "https": "scrapy_zyte_api.ScrapyZyteAPIDownloadHandler",
}
DOWNLOADER_MIDDLEWARES = {
    "scrapy_zyte_api.ScrapyZyteAPIDownloaderMiddleware": 633,
}
SPIDER_MIDDLEWARES = {
    "scrapy_zyte_api.ScrapyZyteAPISpiderMiddleware": 100,
    "scrapy_zyte_api.ScrapyZyteAPIRefererSpiderMiddleware": 1000,
}
REQUEST_FINGERPRINTER_CLASS = "scrapy_zyte_api.ScrapyZyteAPIRequestFingerprinter"
TWISTED_REACTOR = "twisted.internet.asyncioreactor.AsyncioSelectorReactor"
```

--------------------------------

### Engine Lifecycle Management

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Handles engine start and spider closure events for session tasks.

```python
    async def _handle_engine_start(self):
        assert self._crawler.engine
        self._download_async = getattr(self._crawler.engine, "download_async", None)
        self._download = None if self._download_async else self._crawler.engine.download

    def _handle_spider_closed(self):
        self._closing = True
        for task in list(self._init_tasks):
            task.cancel()
```

--------------------------------

### Example Zyte API Log Output

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/reference/settings.rst.txt

Example of the JSON object logged when ZYTE_API_LOG_REQUESTS is enabled.

```text
Sending Zyte API extract request: {"url": "https://example.com", "httpResponseBody": true}
```

--------------------------------

### Asynchronous Session Parameter Initialization

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

An example of implementing the params method as a coroutine to perform asynchronous downloads during session initialization.

```python
async def params(self, request: Request) -> Dict[str, Any]:
    bootstrap_request = Request(
        "https://example.com/api/get-session",
        meta={
            "zyte_api_session_enabled": False,
            "zyte_api": {
                "httpResponseBody": True,
                "responseCookies": True,
            },
        },
    )
    response = await self.crawler.engine.download_async(bootstrap_request)
    return {
        "url": "https://example.com/new-session",
        "httpResponseBody": True,
        "requestCookies": response.raw_api_response["responseCookies"],
    }
```

--------------------------------

### Define session location dictionary

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/usage/session.rst.txt

Example dictionary structure for the location parameter used in session initialization.

```python
{
    "addressCountry": "US",
    "addressRegion": "NY",
    "postalCode": "10001",
    "streetAddress": "3 Penn Plz",
}
```

--------------------------------

### init_session(self, session_id, request, download)

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Initializes a session using a chain of requests. This method should be overridden in a subclass of SessionConfig to perform custom login or session setup flows.

```APIDOC
## init_session(self, session_id, request, download)

### Description
Override this method to initialize a session using a chain of requests. The method receives a session ID, the original request, and a download function that returns a scrapy.http.Response.

### Parameters
- **session_id** (str) - The unique identifier for the session.
- **request** (scrapy.http.Request) - The original request triggering the session initialization.
- **download** (callable) - An asynchronous function that accepts a scrapy.http.Request and returns a scrapy.http.Response.

### Usage Example
```python
from scrapy import Request
from scrapy_zyte_api import SessionConfig

class MySessionConfig(SessionConfig):
    async def init_session(self, session_id, request, download):
        r1 = await download(
            Request(
                "https://example.com/login",
                meta={"zyte_api": {"browserHtml": True}},
            )
        )
        token = r1.css("input[name=csrf]::attr(value)").get()
        if not token:
            return False
        r2 = await download(
            Request(
                "https://example.com/submit",
                meta={
                    "zyte_api": {
                        "browserHtml": True,
                        "requestBody": f"csrf={token}",
                    }
                },
            )
        )
        return "Welcome" in r2.text
```
```

--------------------------------

### Example Scrapy Request to Zyte API Parameters

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/reference/request.rst.txt

Demonstrates how a sample Scrapy request is translated into Zyte API parameters.

```APIDOC
## Example Request Mapping

A Scrapy request like this:

```python
Request(
    method="POST",
    url="https://httpbin.org/anything",
    headers={\"Content-Type\": \"application/json\"},
    body=b'{\"foo\": \"bar\"}',
    cookies={\"a\": \"b\"},
)
```

Results in the following Zyte API parameters:

```json
{
    "customHttpRequestHeaders": [
        {
            "name": "Content-Type",
            "value": "application/json"
        }
    ],
    "experimental": {
        "requestCookies": [
            {
                "name": "a",
                "value": "b",
                "domain": ""
            }
        ],
        "responseCookies": true
    },
    "httpResponseBody": true,
    "httpResponseHeaders": true,
    "httpRequestBody": "eyJmb28iOiAiYmFyIn0=",
    "httpRequestMethod": "POST",
    "url": "https://httpbin.org/anything"
}
```
```

--------------------------------

### Explicitly Setting httpResponseBody

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/reference/request.rst.txt

Example of how to explicitly set `httpResponseBody` to `True` for binary resources.

```APIDOC
## Explicitly Setting httpResponseBody

For binary resources like images or PDFs, explicitly set `httpResponseBody` to `True`:

```python
Request(
    url="https://toscrape.com/img/zyte.png",
    meta={
        "zyte_api_automap": {"httpResponseBody": True},
    },
)
```
```

--------------------------------

### Example Scrapy request for mapping

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/reference/request.html

A standard Scrapy request demonstrating method, headers, body, and cookies that will be mapped to Zyte API parameters.

```python
Request(
    method="POST",
    url="https://httpbin.org/anything",
    headers={"Content-Type": "application/json"},
    body=b'{"foo": "bar"}',
    cookies={"a": "b"},
)
```

--------------------------------

### Explicitly Setting httpResponseHeaders

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/reference/request.rst.txt

Example of how to explicitly set `httpResponseHeaders` to `True` to access response headers.

```APIDOC
## Explicitly Setting httpResponseHeaders

To access response headers, explicitly set `httpResponseHeaders` to `True`:

```python
Request(
    url="https://toscrape.com/",
    meta={
        "zyte_api_automap": {"httpResponseHeaders": True},
    },
)
```
```

--------------------------------

### Get next working session

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Determines whether to create a new session or retrieve one from the queue based on pending initial sessions.

```python
async def _next(self, request) -> str:
    """Return the ID of the next working session in the session pool
    rotation.

    *request* is needed to determine the URL to use for request
    initialization.
    """
    pool = self.get_pool(request)
    if self._pending_initial_sessions[pool] >= 1:
        self._pending_initial_sessions[pool] -= 1
        session_id = await self._create_session(request, pool)
    else:
        session_id = await self._next_from_queue(request, pool)
    return session_id
```

--------------------------------

### NO_CALLBACK Fallback

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Fallback definition for NO_CALLBACK when not available in the Scrapy installation.

```python
try:
    from scrapy.http.request import NO_CALLBACK
except ImportError:

    def NO_CALLBACK(response):  # type: ignore[misc]
        pass  # pragma: no cover
```

--------------------------------

### Configure Ethereum private key via environment variables

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/setup.rst.txt

Set the ZYTE_API_ETH_KEY environment variable to authorize payments using the x402 protocol.

```shell
> set ZYTE_API_ETH_KEY=YOUR_ETH_PRIVATE_KEY
```

```shell
$ export ZYTE_API_ETH_KEY=YOUR_ETH_PRIVATE_KEY
```

--------------------------------

### Initialize a session with a chain of requests

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Override init_session to perform multiple requests, such as logging in, before proceeding with the session.

```python
from scrapy import Request
from scrapy_zyte_api import SessionConfig


class MySessionConfig(SessionConfig):
    async def init_session(self, session_id, request, download):
        r1 = await download(
            Request(
                "https://example.com/login",
                meta={"zyte_api": {"browserHtml": True}},
            )
        )
        token = r1.css("input[name=csrf]::attr(value)").get()
        if not token:
            return False
        r2 = await download(
            Request(
                "https://example.com/submit",
                meta={
                    "zyte_api": {
                        "browserHtml": True,
                        "requestBody": f"csrf={token}",
                    }
                },
            )
        )
        return "Welcome" in r2.text
```

--------------------------------

### Session Initialization and Configuration

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Initializes session caches, retry policies, and error handling based on crawler settings.

```python
        self._session_config_cache: WeakKeyDictionary[Request, SessionConfig] = (
            WeakKeyDictionary()
        )
        self._session_config_map: dict[type[SessionConfig], SessionConfig] = {}

        self._fatal_error_handler = FatalErrorHandler(crawler)

        self._stats_per_pool: bool = settings.getbool("ZYTE_API_SESSION_STATS_PER_POOL")

        session_retry_policy = settings["ZYTE_API_SESSION_RETRY_POLICY"]
        if session_retry_policy is None:
            retry_policy = settings.get(
                "ZYTE_API_RETRY_POLICY", "zyte_api.zyte_api_retrying"
            )
            loaded_retry_policy = load_object(retry_policy)
            session_retry_policy = _SESSION_RETRY_POLICIES.get(loaded_retry_policy)
            if session_retry_policy is None:
                session_retry_policy = retry_policy
                logger.warning(
                    "ZYTE_API_RETRY_POLICY is set to a custom value "
                    f"({retry_policy!r}), which will also be used for "
                    "plugin-managed session requests. Session retry policies "
                    "must not retry 520 or 521 responses; if yours does, "
                    "plugin-managed sessions may not work as expected. Set "
                    "ZYTE_API_SESSION_RETRY_POLICY explicitly to silence this "
                    "warning. See the ZYTE_API_SESSION_RETRY_POLICY setting "
                    "reference for details.",
                )
        self._session_retry_policy = session_retry_policy
```

--------------------------------

### Initialize session data storage

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Create a dictionary in the class constructor to store session-specific data.

```python
def __init__(self, crawler):
    super().__init__(crawler)
    self.session_data = {}
```

--------------------------------

### Resulting Zyte API parameters

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/reference/request.html

The JSON representation of the Zyte API parameters generated from the example Scrapy request.

```json
{
    "customHttpRequestHeaders": [
        {
            "name": "Content-Type",
            "value": "application/json"
        }
    ],
    "experimental": {
        "requestCookies": [
            {
                "name": "a",
                "value": "b",
                "domain": ""
            }
        ],
        "responseCookies": true
    },
    "httpResponseBody": true,
    "httpResponseHeaders": true,
    "httpRequestBody": "eyJmb28iOiAiYmFyIn0=",
    "httpRequestMethod": "POST",
    "url": "https://httpbin.org/anything"
}
```

--------------------------------

### Enable Cookie Sessions Globally

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/usage/session.rst.txt

Configure settings.py to enable session management and switch to cookie-based session mode globally.

```python
ZYTE_API_SESSION_ENABLED = True
ZYTE_API_SESSION_COOKIE_MODE = True
```

--------------------------------

### Zyte API Redirect Debug Log Format

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/usage/redirect.rst.txt

Example of the debug message logged by scrapy-zyte-api when a URL change is detected.

```text
Redirecting to <200 https://final-url.example> from <GET https://original-url.example>
```

--------------------------------

### SessionConfig.init_session

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/usage/session.html

Initializes a session using a chain of requests.

```APIDOC
## init_session(session_id: str, request: Request, download: Callable[[Request], Awaitable[Response]]) -> bool

### Description
Initializes the session identified by session_id. Override this method to perform custom initialization logic, such as logging in or navigating through multiple pages.

### Parameters
- **session_id** (str) - The ID of the session to initialize.
- **request** (Request) - The request that triggered the initialization.
- **download** (Callable) - A function to execute HTTP requests within the session.
```

--------------------------------

### Configure ZYTE_API_SESSION_PARAMS

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/reference/settings.rst.txt

Defines session initialization parameters including browser options and actions in settings.py.

```python
ZYTE_API_SESSION_PARAMS = {
    "browserHtml": True,
    "actions": [
        {
            "action": "setLocation",
            "address": {"postalCode": "10001"},
        }
    ],
}
```

--------------------------------

### params(request: Request) -> dict[str, Any] | Awaitable[dict[str, Any]]

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Returns the Zyte API request parameters used to initialize a session for a given request.

```APIDOC
## params(request: Request)

### Description
Returns the Zyte API request parameters to use to initialize a session for the provided request. The default implementation relies on settings and request metadata.

### Parameters
- **request** (Request) - Required - The Scrapy request object.

### Returns
- **dict[str, Any] | Awaitable[dict[str, Any]]** - A dictionary of Zyte API parameters or an awaitable returning such a dictionary.
```

--------------------------------

### Define Session Parameters

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Defines the parameters for session initialization, optionally using location data from the request.

```python
def params(self, request: Request) -> Dict[str, Any]:
    if location := self.location(request):
        return {
            "url": "https://example.com/new-session/for-country",
            "httpResponseBody": True,
            "httpRequestMethod": "POST",
            "httpRequestText": location["addressCountry"],
        }
    return {
        "url": "https://example.com/new-session",
        "httpResponseBody": True,
    }
```

--------------------------------

### Handle SessionConfig RulesRegistry imports

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Fallback implementation for the RulesRegistry when web-poet is not available.

```python
try:
    from web_poet import RulesRegistry
except ImportError:

    class SessionConfigRulesRegistry:
        def session_config_cls(self, request: Request) -> type[SessionConfig]:
            return SessionConfig
```

--------------------------------

### Session Configuration Classes and Exceptions

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Defines custom exceptions and TypedDict configurations for session management.

```python
class PoolError(ValueError):
    pass


class TooManyBadSessionInits(RuntimeError):
    pass


class PoolConfig(TypedDict):
    id: str
    delay: NotRequired[float]
    randomize_delay: NotRequired[bool]
    size: NotRequired[int]


class PoolOptions(TypedDict):
    delay: NotRequired[float]
    randomize_delay: NotRequired[bool]
    size: NotRequired[int]


QueueSession = tuple[str, float]  # (session_id, next_use_timestamp)


class _SessionInitParamError(Exception):
    """Raised by SessionConfig.init_session() when params() fails, so that
    _SessionManager._init_session() can distinguish the stat to increment."""

    def __init__(self, message: str | None = None):
        super().__init__()
        self.message = message


class _SessionInitCheckError(Exception):
    """Raised by SessionConfig.init_session() when check() raises an
    unexpected exception, carrying the response for the log message."""

    def __init__(self, response):
        super().__init__()
        self.response = response
```

--------------------------------

### location(request: Request) -> dict[str, str]

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Returns the address dictionary used for location-based session initialization for a given request.

```APIDOC
## location(request: Request)

### Description
Returns the address dictionary to use for location-based session initialization for the provided request. This method can be overridden to provide fallback locations.

### Parameters
- **request** (Request) - Required - The Scrapy request object to determine the location for.

### Returns
- **dict[str, str]** - A dictionary containing address fields (e.g., addressCountry, addressRegion, postalCode).
```

--------------------------------

### Define Page Input Classes

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_page_inputs.html

Classes for handling browser actions, geolocation, screenshots, and network capture responses.

```python
from base64 import b64decode
from typing import Any

import attrs

from ._annotations import _ActionResult




[docs]
@attrs.define
class Actions:
    """A page input that specifies browser actions and contains their results.

    The actions must be :ref:`specified with an annotation
    <browser-actions>` using :func:`~scrapy_zyte_api.actions`.
    """

    #: Results of actions.
    results: list[_ActionResult] | None




[docs]
@attrs.define
class Geolocation:
    """A page input that forces a given geolocation for all other page inputs.

    The target geolocation must be :ref:`specified with an annotation
    <geolocation>`.
    """




[docs]
@attrs.define
class Screenshot:
    """A container for holding the screenshot of a webpage."""

    #: Body.
    body: bytes

    @classmethod
    def from_base64(cls, body):
        return cls(body=b64decode(body.encode()))




[docs]
@attrs.define
class CapturedResponse:
    """A network response captured during browser page rendering.

    Part of :class:`NetworkCapture`.
    """

    #: Response URL.
    url: str

    #: HTTP status code.
    status: int

    #: Response headers.
    headers: dict[str, str]

    #: Response body. ``None`` if ``httpResponseBody`` was not set to ``True``
    #: on the matching filter.
    body: bytes | None

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "CapturedResponse":
        body_b64: str | None = data.get("httpResponseBody")
        return cls(
            url=data["url"],
            status=data["statusCode"],
            headers=data.get("headers", {}),
            body=b64decode(body_b64) if body_b64 is not None else None,
        )




[docs]
@attrs.define
class NetworkCapture:
    """A page input that specifies network capture filters and contains captured responses.

    The filters must be :ref:`specified with an annotation
    <network-capture>` using :func:`~scrapy_zyte_api.network_capture`.
    """

    #: Captured responses.
    results: list[CapturedResponse]
```

--------------------------------

### Configuring browser actions

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/usage/scrapy-poet.rst.txt

Defines browser actions using the Actions dependency and the actions function, and validates the results.

```python
from typing import Annotated

from scrapy_zyte_api import Actions, actions


@attrs.define
class MyPageObject(BasePage):
    product: Product
    actions: Annotated[
        Actions,
        actions(
            [
                {
                    "action": "click",
                    "selector": {"type": "css", "value": "button#openDescription"},
                    "delay": 0,
                    "button": "left",
                    "onError": "return",
                },
                {"action": "waitForTimeout", "timeout": 5, "onError": "return"},
            ]
        ),
    ]
```

```python
def validate_input(self):
    for action_result in self.actions.result:
        if action_result["status"] != "success":
            return Product(is_valid=False)
    return None
```

--------------------------------

### Requesting page inputs in page objects

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/usage/scrapy-poet.html

Define page objects that accept supported inputs like BrowserResponse and Product via dependency injection.

```python
@attrs.define
class ProductPage(BasePage):
    response: BrowserResponse
    product: Product


class ZyteApiSpider(scrapy.Spider):
    ...

    def parse_page(self, response: DummyResponse, page: ProductPage):
        ...
```

--------------------------------

### Configure Ethereum private key in settings.py

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/setup.rst.txt

Define the ZYTE_API_ETH_KEY variable directly within the Scrapy settings module.

```python
ZYTE_API_ETH_KEY = "YOUR_ETH_PRIVATE_KEY"
```

--------------------------------

### Configure session pool

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Return a dictionary to define custom session pool parameters based on the request URL.

```python
def pool(self, request):
    if "ecommerce.example" in urlparse_cached(request).netloc:
        return {
            "id": "ecommerce.example",
            "delay": 2.0,
            "size": 16,
        }
    return super().pool(request)
```

--------------------------------

### Enable scrapy-zyte-api via Addon

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/setup.rst.txt

Configuration for Scrapy 2.10+ to enable the integration using the Addon system.

```python
ADDONS = {
    "scrapy_zyte_api.Addon": 500,
}
```

--------------------------------

### Configure Scrapy Addons

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/changes.html

Use the ADDONS setting to enable scrapy-poet and scrapy-zyte-api, ensuring proper middleware integration for versions 0.26.0 and higher.

```python
ADDONS = {
    "scrapy_poet.Addon": 300,
    "scrapy_zyte_api.Addon": 500,
}
```

--------------------------------

### Create a new session

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Initializes a new session ID and adds it to the specified pool, handling retries and delays.

```python
async def _create_session(self, request: Request, pool: str) -> str:
    async with self._fatal_error_handler:
        while True:
            if self._closing:
                raise IgnoreRequest
            session_id = str(uuid4())
            session_init_succeeded = await self._init_session(
                session_id, request, pool
            )
            if session_init_succeeded:
                self._pools[pool].add(session_id)
                self._bad_inits[pool] = 0
                break
            if session_init_succeeded is None:
                continue
            self._bad_inits[pool] += 1
            if self._bad_inits[pool] >= self._max_bad_inits[pool]:
                raise TooManyBadSessionInits
        pool_config = self._pool_configs[pool]
        delay = pool_config["delay"]
        sleep_delay = next_use_delay = delay
        if pool_config["randomize_delay"]:
            next_use_delay *= random.uniform(0.5, 1.5)  # noqa: S311
            sleep_delay *= random.uniform(0.5, 1.5)  # noqa: S311
        await sleep(sleep_delay)
        next_use = time.time() + next_use_delay
        self._queues[pool].append((session_id, next_use))
        return session_id
```

--------------------------------

### Configure API key in settings.py

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/setup.rst.txt

Setting the API key directly in the Scrapy settings module.

```python
ZYTE_API_KEY = "YOUR_API_KEY"
```

--------------------------------

### Annotation Helper Functions

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_annotations.html

Functions to convert configuration iterables into hashable tuples for use as annotations.

```python
[docs]
def actions(value: Iterable[Action]) -> tuple[Any, ...]:
    """Convert an iterable of :class:`~scrapy_zyte_api.Action` dicts into a hashable value."""
    # both lists and dicts are not hashable and we need dep types to be hashable
    return tuple(make_hashable(action) for action in value)


class NetworkCaptureFilter(TypedDict, total=False):
    """A filter for :func:`~scrapy_zyte_api.network_capture`.

    See :ref:`network-capture`.
    """

    filterType: str
    value: str
    matchType: str
    httpResponseBody: bool


[docs]
def network_capture(filters: Iterable[NetworkCaptureFilter]) -> tuple[Any, ...]:
    """Convert an iterable of :class:`~scrapy_zyte_api.NetworkCaptureFilter` dicts into a hashable value."""
    return tuple(make_hashable(f) for f in filters)




[docs]
def custom_attrs(
    input: dict[str, Any],  # noqa: A002
    options: dict[str, Any] | None = None,
) -> tuple[frozenset[Any], frozenset[Any] | None]:
    input_wrapped = make_hashable(input)
    options_wrapped = make_hashable(options) if options else None
    return input_wrapped, options_wrapped
```

--------------------------------

### Configure Default Zyte API Parameters

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/usage/scrapy-poet.rst.txt

Set default extraction options for Zyte API using the ZYTE_API_PROVIDER_PARAMS setting.

```python
ZYTE_API_PROVIDER_PARAMS = {
    "productOptions": {"extractFrom": "httpResponseBody"},
    "productNavigationOptions": {"extractFrom": "httpResponseBody"},
}
```

--------------------------------

### Provide a fallback location configuration

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/usage/session.html

Override the location method to provide default address fields when the standard configuration is missing.

```python
def location(self, request: Request) -> Dict[str, str]:
    fallback = {"addressCountry": "US", "addressRegion": "NY", "postalCode": "10001"}
    return super().location(request) or fallback
```

--------------------------------

### Initialize _SessionManager

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

The constructor initializes session pools, configuration settings, and signal connections for managing Zyte API sessions.

```python
class _SessionManager:
    def __init__(self, crawler: Crawler):
        self._crawler = crawler
        self._closing = False
        crawler.signals.connect(
            self._handle_engine_start, signal=signals.engine_started
        )
        crawler.signals.connect(
            self._handle_spider_closed, signal=signals.spider_closed
        )

        settings = crawler.settings

        self._default_pool_delay = settings.getfloat(
            "ZYTE_API_SESSION_DELAY", settings.getfloat("DOWNLOAD_DELAY")
        )
        self._randomize_delay = settings.getbool(
            "ZYTE_API_SESSION_RANDOMIZE_DELAY",
            settings.getbool("RANDOMIZE_DOWNLOAD_DELAY"),
        )
        self._default_pool_size = settings.getint("ZYTE_API_SESSION_POOL_SIZE", 8)
        self._pending_initial_sessions: dict[str, int] = {}
        self._pool_configs = settings.getdict("ZYTE_API_SESSION_POOLS")
        pool_sizes = settings.getdict("ZYTE_API_SESSION_POOL_SIZES")
        if pool_sizes:
            logger.warning(
                "ZYTE_API_SESSION_POOL_SIZES is deprecated, use "
                "ZYTE_API_SESSION_POOLS instead"
            )
            for pool_id, pool_size in pool_sizes.items():
                self._pool_configs.setdefault(pool_id, {}).setdefault("size", pool_size)
        for pool, config in self._pool_configs.items():
            if "size" in config:
                self._pending_initial_sessions[pool] = config["size"]

        self._max_check_failures = settings.getint(
            "ZYTE_API_SESSION_MAX_CHECK_FAILURES", 1
        )
        self._check_failures: dict[str, int] = defaultdict(int)

        self._max_errors = settings.getint("ZYTE_API_SESSION_MAX_ERRORS", 1)
        self._errors: dict[str, int] = defaultdict(int)

        max_bad_inits = settings.getint("ZYTE_API_SESSION_MAX_BAD_INITS", 8)
        self._max_bad_inits: dict[str, int] = defaultdict(lambda: max_bad_inits)
        max_bad_inits_per_pool = settings.getdict(
            "ZYTE_API_SESSION_MAX_BAD_INITS_PER_POOL"
        )
        for pool, pool_max_bad_inits in max_bad_inits_per_pool.items():
            self._max_bad_inits[pool] = pool_max_bad_inits
        self._bad_inits: dict[str, int] = defaultdict(int)

        # Transparent mode, needed to determine whether to set the session
        # using ``zyte_api`` or ``zyte_api_automap``.
        self._transparent_mode: bool = settings.getbool(
            "ZYTE_API_TRANSPARENT_MODE", False
        )

        # Each pool contains the IDs of sessions that have not expired yet.
        #
        # While the initial sessions of a pool have not all been started, for
        # every request needing a session, a new session is initialized and
        # then added to the pool.
        #
        # Once a pool is full, sessions are picked from the pool queue, which
        # should contain all pool sessions that have been initialized.
        #
        # As soon as a session expires, it is removed from its pool, and a task
        # to initialize that new session is started.
        self._pools: dict[str, set[str]] = defaultdict(set)

        # Maps session_id → list of cookies for cookie sessions. Only
        # populated for sessions that use cookie-based management.
        self._cookie_jar: dict[str, list] = {}
        self._pool_cache: WeakKeyDictionary[Request, str] = WeakKeyDictionary()

        # The queue is a rotating list of session IDs to use.
        #
        # The way to use the queue is to get a session ID with popleft(), and
        # put it back to the end of the queue with append().
        #
        # The queue may contain session IDs from expired sessions. If the
        # popped session ID cannot be found in the pool, then it should be
        # discarded instead of being put back in the queue.
        #
        # When a new session ID is added to the pool, it is still not added to
        # the queue until the session is actually initialized, when it is
        # appended to the queue.
        #
        # If the queue is empty, sleep and try again. Sessions from the pool
        # will be appended to the queue as they are initialized and ready to
        # use.
        self._queues: dict[str, deque[QueueSession]] = defaultdict(deque)
        self._queue_max_attempts = settings.getint(
            "ZYTE_API_SESSION_QUEUE_MAX_ATTEMPTS", 60
        )
        self._queue_wait_time = settings.getfloat(
            "ZYTE_API_SESSION_QUEUE_WAIT_TIME", 1.0
        )
        self._creation_retry_delay = settings.getfloat(
            "ZYTE_API_SESSION_CREATION_RETRY_DELAY", 60.0
        )

        # Contains the on-going tasks to create new sessions.
        #
        # Keeping a reference to those tasks until they are done is necessary
        # to prevent garbage collection to remove the tasks.
        self._init_tasks: set[Task] = set()
```

--------------------------------

### Explicitly enable binary response body

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/reference/request.html

Use this configuration when requesting binary resources like images or PDFs to ensure the response body is returned.

```python
Request(
    url="https://toscrape.com/img/zyte.png",
    meta={
        "zyte_api_automap": {"httpResponseBody": True},
    },
)
```

--------------------------------

### Configure ZYTE_API_SESSION_LOCATION

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/reference/settings.rst.txt

Sets the location parameters for session initialization in settings.py.

```python
ZYTE_API_SESSION_LOCATION = {"postalCode": "10001"}
```

--------------------------------

### Performing type checks with ZyteAPIMixin

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/reference/response.rst.txt

Illustrates importing the mixin class used for isinstance checks across all Zyte API response types.

```python
from scrapy_zyte_api.responses import ZyteAPIMixin
```

--------------------------------

### Enable transparent mode

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/setup.rst.txt

Setting to route all requests through the Zyte API.

```python
ZYTE_API_TRANSPARENT_MODE = True
```

--------------------------------

### Configure scrapy-poet provider

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/setup.rst.txt

Adding the Zyte API provider to the SCRAPY_POET_PROVIDERS setting.

```python
SCRAPY_POET_PROVIDERS = {
    "scrapy_zyte_api.providers.ZyteApiProvider": 1100,
}
```

--------------------------------

### Annotating item dependencies

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/usage/scrapy-poet.rst.txt

Uses ExtractFrom annotation to specify extraction options for item dependencies.

```python
from typing import Annotated

from scrapy_zyte_api import ExtractFrom


@attrs.define
class MyPageObject(BasePage):
    product: Annotated[Product, ExtractFrom.httpResponseBody]
```

--------------------------------

### Initialize Zyte API session

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Internal method to initialize a session, handle download requests, and manage session-related errors and cookie state.

```python
    async def _init_session(
        self, session_id: str, request: Request, pool: str
    ) -> bool | None:
        assert self._crawler.engine
        session_config = self._get_session_config(request)

        async def download(init_request: Request) -> Response:
            meta = {**init_request.meta}
            meta[SESSION_INIT_META_KEY] = True
            meta.setdefault("dont_merge_cookies", True)
            meta.setdefault("zyte_api_retry_policy", self._session_retry_policy)
            if "zyte_api" in meta:
                zyte_api = {**meta["zyte_api"]}
                if "session" not in zyte_api:
                    zyte_api["session"] = {"id": session_id}
                elif zyte_api["session"] is None:
                    del zyte_api["session"]
                meta["zyte_api"] = zyte_api
            if init_request.callback is None:
                init_request = init_request.replace(meta=meta, callback=NO_CALLBACK)
            else:
                init_request = init_request.replace(meta=meta)
            if self._download_async is not None:  # Scrapy >= 2.14
                return await self._download_async(init_request)
            assert self._download
            if not _DOWNLOAD_NEEDS_SPIDER:
                return await deferred_to_future(self._download(init_request))
            return await deferred_to_future(
                self._download(init_request, spider=self._crawler.spider)  # type: ignore[call-arg]
            )

        cookies_mode = session_config.cookie_mode(request)
        _init_responses: list = []

        async def _tracking_download(init_request: Request) -> Response:
            response = await download(init_request)
            _init_responses.append(response)
            return response

        try:
            result = await session_config.init_session(
                session_id, request, _tracking_download
            )
        except _SessionInitParamError as e:
            self._inc_stat("init/param-error", pool)
            if e.message:
                logger.error(e.message)
            else:
                logger.exception(
                    f"Unexpected exception raised while obtaining session "
                    f"initialization parameters for request {request}.",
                    exc_info=e.__cause__,
                )
            return False
        except _SessionInitCheckError as e:
            self._inc_stat("init/check-error", pool)
            logger.exception(
                f"Unexpected exception raised while checking session "
                f"validity on response {e.response}.",
                exc_info=e.__cause__,
            )
            return False
        except RequestError as e:
            error_type = e.parsed.type
            if error_type == "/problem/over-session-limit":
                self._inc_stat("init/over-limit", pool)
                logger.warning(
                    f"Session initialization failed because the active session "
                    f"limit has been reached for request {request}. Will retry "
                    f"after {self._creation_retry_delay} seconds."
                )
                await sleep(self._creation_retry_delay)
                return None
            if error_type == "/problem/session-creation-error":
                self._inc_stat("init/server-error", pool)
                logger.warning(
                    f"Session initialization failed due to a server error for "
                    f"request {request}. Will retry after "
                    f"{self._creation_retry_delay} seconds."
                )
                await sleep(self._creation_retry_delay)
                return None
            self._inc_stat("init/failed", pool)
            return False
        except CloseSpider:
            raise
        except Exception:
            self._inc_stat("init/failed", pool)
            return False
        outcome = "passed" if result else "failed"
        self._inc_stat(f"init/check-{outcome}", pool)
        if cookies_mode and result and _init_responses:
            self._cookie_jar[session_id] = _init_responses[-1].raw_api_response.get(
                "responseCookies", []
            )
        return result
```

--------------------------------

### Implement a Session Checker with Crawler Access

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/reference/settings.rst.txt

Demonstrates using the from_crawler class method to access Scrapy settings within a custom session checker.

```python
from scrapy import Request
from scrapy.http.response import Response


class MySessionChecker:

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler)

    def __init__(self, crawler):
        location = crawler.settings["ZYTE_API_SESSION_LOCATION"]
        self.postal_code = location["postalCode"]

    def check(self, response: Response, request: Request) -> bool:
        return response.css(".postal_code::text").get() == self.postal_code


ZYTE_API_SESSION_CHECKER = MySessionChecker
```

--------------------------------

### Handling screenshot responses and decoding raw data

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/reference/response.rst.txt

Shows how to handle responses containing screenshots and decode base64 encoded data from the raw API response.

```python
def parse(self, response):
    print(response.url)
    # "https://quotes.toscrape.com/"
    print(response.status)
    # 200
    print(response.headers)
    # {}
    print(response.text)
    # ""
    print(response.body)
    # b""
    print(response.raw_api_response)
    # {
    #     "url": "https://quotes.toscrape.com/",
    #     "statusCode": 200,
    #     "screenshot": "iVBORw0KGgoAAAANSUh…",
    # }
    from base64 import b64decode

    print(b64decode(response.raw_api_response["screenshot"]))
    # b'\x89PNG\r\n\x1a\n\x00\x00\x00\r…'
```

--------------------------------

### Set ZYTE_API_KEY environment variable

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/setup.rst.txt

Commands to set the API key as an environment variable for different operating systems.

```shell
> set ZYTE_API_KEY=YOUR_API_KEY
```

```shell
$ export ZYTE_API_KEY=YOUR_API_KEY
```

--------------------------------

### Define Actions Input

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/reference/inputs.html

Represents browser actions and their results, requiring an annotation via actions().

```python
class scrapy_zyte_api.Actions(results : list[_ActionResult] | None)
```

--------------------------------

### Override location method for session initialization

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Provides a fallback location dictionary when the default implementation returns an empty dictionary.

```python
def location(self, request: Request) -> Dict[str, str]:
    fallback = {"addressCountry": "US", "addressRegion": "NY", "postalCode": "10001"}
    return super().location(request) or fallback
```

```python
def location(self, request: Request) -> Dict[str, str]:
    fallback = {}
    if postal_code := w3lib.url.url_query_parameter(request.url, "postalCode"):
        fallback["postalCode"] = postal_code
    return super().location(request) or fallback
```

--------------------------------

### SessionConfig.location

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/usage/session.html

Returns the address dictionary for location-based session initialization.

```APIDOC
## location(request: Request) -> dict[str, str]

### Description
Returns the address dictionary to use for location-based session initialization for a given request.

### Parameters
- **request** (Request) - The request for which to determine the location.
```

--------------------------------

### Session Configuration Retrieval

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Retrieves or builds the session configuration for a given request.

```python
    def _get_session_config(self, request: Request) -> SessionConfig:
        try:
            return self._session_config_cache[request]
        except KeyError:
            cls = session_config_registry.session_config_cls(request)
            if cls not in self._session_config_map:
                self._session_config_map[cls] = _build_from_crawler(cls, self._crawler)
            self._session_config_cache[request] = self._session_config_map[cls]
            return self._session_config_map[cls]
```

--------------------------------

### SessionConfig Implementation

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_modules/scrapy_zyte_api/_session.html

Main configuration class for managing Scrapy Zyte API sessions.

```python
class SessionConfig:
    """Default session configuration for :ref:`scrapy-zyte-api sessions
    <session>`."""

    #: List of address fields to use when available, and their order, when
    #: :ref:`creating a pool ID for a request <session-pools>` based on the
    #: content of the :reqmeta:`zyte_api_session_location` metadata key. See
    #: :meth:`pool`.
    ADDRESS_FIELDS: list[str] = [
        "addressCountry",
        "addressRegion",
        "postalCode",
        "streetAddress",
    ]

    @classmethod
    def from_crawler(cls, crawler):
        return cls(crawler)

    def __init__(self, crawler):
        self.crawler = crawler

        settings = crawler.settings
        self._setting_location = settings.getdict("ZYTE_API_SESSION_LOCATION")
        self._setting_params = settings.getdict("ZYTE_API_SESSION_PARAMS")

        checker_cls = settings.get("ZYTE_API_SESSION_CHECKER", None)
        if checker_cls:
            self._checker = _build_from_crawler(load_object(checker_cls), crawler)
        else:
            self._checker = None
        self._enabled = crawler.settings.getbool("ZYTE_API_SESSION_ENABLED", False)
        self._cookie_mode = settings.getbool("ZYTE_API_SESSION_COOKIE_MODE")
        self._init_action_failure_invalidates = settings.getbool(
            "ZYTE_API_SESSION_INIT_ACTION_FAILURE_INVALIDATES_SESSION", True
        )
        self._pool_counters = defaultdict(int)
        self._param_pools: defaultdict[str, dict[str, int]] = defaultdict(dict)
```

--------------------------------

### Requesting page inputs in page objects and callbacks

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/_sources/usage/scrapy-poet.rst.txt

Demonstrates requesting supported page inputs within a Page Object class and directly in a spider callback.

```python
@attrs.define
class ProductPage(BasePage):
    response: BrowserResponse
    product: Product


class ZyteApiSpider(scrapy.Spider):
    ...

    def parse_page(self, response: DummyResponse, page: ProductPage):
        ...
```

```python
class ZyteApiSpider(scrapy.Spider):
    ...

    def parse_page(self,
                   response: DummyResponse,
                   browser_response: BrowserResponse,
                   product: Product,
                   ):
        ...
```

--------------------------------

### Configure Scrapy-Zyte-API Settings

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/changes.html

Required settings for integrating the downloader middleware and request fingerprinter in Scrapy 2.7+.

```python
DOWNLOADER_MIDDLEWARES = {
    "scrapy_zyte_api.ScrapyZyteAPIDownloaderMiddleware": 633,
}
# only applicable for Scrapy 2.7+
REQUEST_FINGERPRINTER_CLASS = "scrapy_zyte_api.ScrapyZyteAPIRequestFingerprinter"
```

--------------------------------

### Enable browser rendering via Zyte API

Source: https://scrapy-zyte-api.readthedocs.io/en/latest/usage/manual.html

Use the browserHtml parameter within the zyte_api meta dictionary to enable browser rendering for a request.

```python
import scrapy


class SampleQuotesSpider(scrapy.Spider):
    name = "sample_quotes"

    async def start(self):
        yield scrapy.Request(
            url="https://quotes.toscrape.com/",
            meta={
                "zyte_api": {
                    "browserHtml": True,
                }
            },
        )

    def parse(self, response):
        print(response.text)
        # "<html>…</html>"
```
Tokens:
Raw