Edit on GitHub

agent_search_gateway.providers.web.zenrows

ZenRows current Fetch API adapter.

 1"""ZenRows current Fetch API adapter."""
 2
 3from urllib.parse import urlencode
 4
 5from ...observability import SecretValue
 6from ...providers.contracts import URLFetchCandidate
 7from ...url_normalization import NormalizedURL
 8from .common import TextRequester, configured_string, endpoint, failure
 9
10
11class ZenRowsAdapter:
12    def __init__(
13        self,
14        *,
15        name: str,
16        api_url: str,
17        secret: SecretValue,
18        http_executor: TextRequester,
19    ) -> None:
20        self.name = name
21        self._api_url = configured_string(api_url, "api_url").rstrip("/")
22        self._secret = secret
23        self._http = http_executor
24
25    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
26        params = urlencode(
27            {
28                "apikey": self._secret.reveal(),
29                "url": str(url),
30                "response_type": "markdown",
31            }
32        )
33        request_url = f"{endpoint(self._api_url, '/v1')}?{params}"
34        text = await self._http.request_text("GET", request_url, stage="fetch")
35        if not text.strip():
36            raise failure(self.name, "fetch", "page body is empty")
37        return URLFetchCandidate(text, text)
class ZenRowsAdapter: 
12class ZenRowsAdapter:
13    def __init__(
14        self,
15        *,
16        name: str,
17        api_url: str,
18        secret: SecretValue,
19        http_executor: TextRequester,
20    ) -> None:
21        self.name = name
22        self._api_url = configured_string(api_url, "api_url").rstrip("/")
23        self._secret = secret
24        self._http = http_executor
25
26    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
27        params = urlencode(
28            {
29                "apikey": self._secret.reveal(),
30                "url": str(url),
31                "response_type": "markdown",
32            }
33        )
34        request_url = f"{endpoint(self._api_url, '/v1')}?{params}"
35        text = await self._http.request_text("GET", request_url, stage="fetch")
36        if not text.strip():
37            raise failure(self.name, "fetch", "page body is empty")
38        return URLFetchCandidate(text, text)
ZenRowsAdapter( *, name: str, api_url: str, secret: agent_search_gateway.observability.SecretValue, http_executor: agent_search_gateway.providers.web.common.TextRequester) 
13    def __init__(
14        self,
15        *,
16        name: str,
17        api_url: str,
18        secret: SecretValue,
19        http_executor: TextRequester,
20    ) -> None:
21        self.name = name
22        self._api_url = configured_string(api_url, "api_url").rstrip("/")
23        self._secret = secret
24        self._http = http_executor
name
async def fetch( self, url: agent_search_gateway.url_normalization.NormalizedURL) -> agent_search_gateway.providers.contracts.URLFetchCandidate: 
26    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
27        params = urlencode(
28            {
29                "apikey": self._secret.reveal(),
30                "url": str(url),
31                "response_type": "markdown",
32            }
33        )
34        request_url = f"{endpoint(self._api_url, '/v1')}?{params}"
35        text = await self._http.request_text("GET", request_url, stage="fetch")
36        if not text.strip():
37            raise failure(self.name, "fetch", "page body is empty")
38        return URLFetchCandidate(text, text)