Skip to content

intercept

BaseFetchInterception

Base class to wait for a Fetch response matching a URL pattern. Use this to collect and decode a paused fetch response, while keeping the use block clean and returning its own result.

Parameters:

Name Type Description Default
tab Connection

The Tab instance to monitor.

 required
url_pattern str

The URL pattern to match requests and responses.

 required
request_stage RequestStage

The stage of the fetch request to intercept (e.g., request or response).

 required
resource_type ResourceType

The type of resource to intercept (e.g., document, script, etc.).

 required
Source code in zendriver/core/intercept.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class BaseFetchInterception:
    """
    Base class to wait for a Fetch response matching a URL pattern.
    Use this to collect and decode a paused fetch response, while keeping
    the use block clean and returning its own result.

    :param tab: The Tab instance to monitor.
    :param url_pattern: The URL pattern to match requests and responses.
    :param request_stage: The stage of the fetch request to intercept (e.g., request or response).
    :param resource_type: The type of resource to intercept (e.g., document, script, etc.).
    """

    def __init__(
        self,
        tab: Connection,
        url_pattern: str,
        request_stage: RequestStage,
        resource_type: ResourceType,
    ):
        self.tab = tab
        self.url_pattern = url_pattern
        self.request_stage = request_stage
        self.resource_type = resource_type
        self.response_future: asyncio.Future[cdp.fetch.RequestPaused] = asyncio.Future()

    async def _response_handler(self, event: cdp.fetch.RequestPaused) -> None:
        """
        Internal handler for response events.
        :param event: The response event.
        :type event: cdp.fetch.RequestPaused
        """
        self._remove_response_handler()
        self.response_future.set_result(event)

    def _remove_response_handler(self) -> None:
        """
        Remove the response event handler.
        """
        self.tab.remove_handlers(cdp.fetch.RequestPaused, self._response_handler)

    async def __aenter__(self) -> "BaseFetchInterception":
        """
        Enter the context manager, adding request and response handlers.
        """
        await self._setup()
        return self

    async def __aexit__(self, *args: typing.Any) -> None:
        """
        Exit the context manager, removing request and response handlers.
        """
        await self._teardown()

    async def _setup(self) -> None:
        await self.tab.send(
            cdp.fetch.enable(
                [
                    RequestPattern(
                        url_pattern=self.url_pattern,
                        request_stage=self.request_stage,
                        resource_type=self.resource_type,
                    )
                ]
            )
        )
        self.tab.enabled_domains.append(
            cdp.fetch
        )  # trick to avoid another `fetch.enable` call by _register_handlers
        self.tab.add_handler(cdp.fetch.RequestPaused, self._response_handler)

    async def _teardown(self) -> None:
        self._remove_response_handler()
        await self.tab.send(cdp.fetch.disable())

    async def reset(self) -> None:
        """
        Resets the internal state, allowing the interception to be reused.
        """
        self.response_future = asyncio.Future()
        await self._teardown()
        await self._setup()

    @property
    async def request(self) -> cdp.network.Request:
        """
        Get the matched request.
        :return: The matched request.
        :rtype: cdp.network.request
        """
        return (await self.response_future).request

    @property
    async def response_body(self) -> tuple[str, bool]:
        """
        Get the body of the matched response.
        :return: The response body.
        :rtype: str
        """
        request_id = (await self.response_future).request_id
        body = await self.tab.send(cdp.fetch.get_response_body(request_id=request_id))
        return body

    async def fail_request(self, error_reason: cdp.network.ErrorReason) -> None:
        request_id = (await self.response_future).request_id
        await self.tab.send(
            cdp.fetch.fail_request(request_id=request_id, error_reason=error_reason)
        )

    async def continue_request(
        self,
        url: typing.Optional[str] = None,
        method: typing.Optional[str] = None,
        post_data: typing.Optional[str] = None,
        headers: typing.Optional[typing.List[HeaderEntry]] = None,
        intercept_response: typing.Optional[bool] = None,
    ) -> None:
        request_id = (await self.response_future).request_id
        await self.tab.send(
            cdp.fetch.continue_request(
                request_id=request_id,
                url=url,
                method=method,
                post_data=post_data,
                headers=headers,
                intercept_response=intercept_response,
            )
        )

    async def fulfill_request(
        self,
        response_code: int,
        response_headers: typing.Optional[typing.List[HeaderEntry]] = None,
        binary_response_headers: typing.Optional[str] = None,
        body: typing.Optional[str] = None,
        response_phrase: typing.Optional[str] = None,
    ) -> None:
        request_id = (await self.response_future).request_id
        await self.tab.send(
            cdp.fetch.fulfill_request(
                request_id=request_id,
                response_code=response_code,
                response_headers=response_headers,
                binary_response_headers=binary_response_headers,
                body=body,
                response_phrase=response_phrase,
            )
        )

    async def continue_response(
        self,
        response_code: typing.Optional[int] = None,
        response_phrase: typing.Optional[str] = None,
        response_headers: typing.Optional[typing.List[HeaderEntry]] = None,
        binary_response_headers: typing.Optional[str] = None,
    ) -> None:
        request_id = (await self.response_future).request_id
        await self.tab.send(
            cdp.fetch.continue_response(
                request_id=request_id,
                response_code=response_code,
                response_phrase=response_phrase,
                response_headers=response_headers,
                binary_response_headers=binary_response_headers,
            )
        )

request: cdp.network.Request async property

Get the matched request.

Returns:

Type Description
cdp.network.request

The matched request.

request_stage = request_stage instance-attribute

resource_type = resource_type instance-attribute

response_body: tuple[str, bool] async property

Get the body of the matched response.

Returns:

Type Description
str

The response body.

response_future: asyncio.Future[cdp.fetch.RequestPaused] = asyncio.Future() instance-attribute

tab = tab instance-attribute

url_pattern = url_pattern instance-attribute

__aenter__() async

Enter the context manager, adding request and response handlers.

Source code in zendriver/core/intercept.py 
50
51
52
53
54
55
async def __aenter__(self) -> "BaseFetchInterception":
    """
    Enter the context manager, adding request and response handlers.
    """
    await self._setup()
    return self

__aexit__(*args) async

Exit the context manager, removing request and response handlers.

Source code in zendriver/core/intercept.py 
57
58
59
60
61
async def __aexit__(self, *args: typing.Any) -> None:
    """
    Exit the context manager, removing request and response handlers.
    """
    await self._teardown()

__init__(tab, url_pattern, request_stage, resource_type)

Source code in zendriver/core/intercept.py 
22
23
24
25
26
27
28
29
30
31
32
33
def __init__(
    self,
    tab: Connection,
    url_pattern: str,
    request_stage: RequestStage,
    resource_type: ResourceType,
):
    self.tab = tab
    self.url_pattern = url_pattern
    self.request_stage = request_stage
    self.resource_type = resource_type
    self.response_future: asyncio.Future[cdp.fetch.RequestPaused] = asyncio.Future()

continue_request(url=None, method=None, post_data=None, headers=None, intercept_response=None) async

Source code in zendriver/core/intercept.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
async def continue_request(
    self,
    url: typing.Optional[str] = None,
    method: typing.Optional[str] = None,
    post_data: typing.Optional[str] = None,
    headers: typing.Optional[typing.List[HeaderEntry]] = None,
    intercept_response: typing.Optional[bool] = None,
) -> None:
    request_id = (await self.response_future).request_id
    await self.tab.send(
        cdp.fetch.continue_request(
            request_id=request_id,
            url=url,
            method=method,
            post_data=post_data,
            headers=headers,
            intercept_response=intercept_response,
        )
    )

continue_response(response_code=None, response_phrase=None, response_headers=None, binary_response_headers=None) async

Source code in zendriver/core/intercept.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
async def continue_response(
    self,
    response_code: typing.Optional[int] = None,
    response_phrase: typing.Optional[str] = None,
    response_headers: typing.Optional[typing.List[HeaderEntry]] = None,
    binary_response_headers: typing.Optional[str] = None,
) -> None:
    request_id = (await self.response_future).request_id
    await self.tab.send(
        cdp.fetch.continue_response(
            request_id=request_id,
            response_code=response_code,
            response_phrase=response_phrase,
            response_headers=response_headers,
            binary_response_headers=binary_response_headers,
        )
    )

fail_request(error_reason) async

Source code in zendriver/core/intercept.py 
112
113
114
115
116
async def fail_request(self, error_reason: cdp.network.ErrorReason) -> None:
    request_id = (await self.response_future).request_id
    await self.tab.send(
        cdp.fetch.fail_request(request_id=request_id, error_reason=error_reason)
    )

fulfill_request(response_code, response_headers=None, binary_response_headers=None, body=None, response_phrase=None) async

Source code in zendriver/core/intercept.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
async def fulfill_request(
    self,
    response_code: int,
    response_headers: typing.Optional[typing.List[HeaderEntry]] = None,
    binary_response_headers: typing.Optional[str] = None,
    body: typing.Optional[str] = None,
    response_phrase: typing.Optional[str] = None,
) -> None:
    request_id = (await self.response_future).request_id
    await self.tab.send(
        cdp.fetch.fulfill_request(
            request_id=request_id,
            response_code=response_code,
            response_headers=response_headers,
            binary_response_headers=binary_response_headers,
            body=body,
            response_phrase=response_phrase,
        )
    )

reset() async

Resets the internal state, allowing the interception to be reused.

Source code in zendriver/core/intercept.py 
84
85
86
87
88
89
90
async def reset(self) -> None:
    """
    Resets the internal state, allowing the interception to be reused.
    """
    self.response_future = asyncio.Future()
    await self._teardown()
    await self._setup()