Edit on GitHub

agent_search_gateway.providers.web.scrapingant

ScrapingAnt v2 Markdown fetch adapter.

 1"""ScrapingAnt v2 Markdown fetch 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 (
 9    JsonRequester,
10    configured_string,
11    endpoint,
12    failure,
13    non_empty_string,
14    require_object,
15)
16
17
18class ScrapingAntAdapter:
19    def __init__(
20        self,
21        *,
22        name: str,
23        api_url: str,
24        secret: SecretValue,
25        http_executor: JsonRequester,
26    ) -> None:
27        self.name = name
28        self._api_url = configured_string(api_url, "api_url").rstrip("/")
29        self._secret = secret
30        self._http = http_executor
31
32    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
33        request_url = (
34            f"{endpoint(self._api_url, '/v2/markdown')}?"
35            f"{urlencode({'url': str(url), 'x-api-key': self._secret.reveal()})}"
36        )
37        payload = await self._http.request_json("GET", request_url, stage="fetch")
38        root = require_object(payload, self.name, "fetch", "response")
39        if root.get("error") is not None:
40            raise failure(self.name, "fetch", "provider reported failure")
41        markdown = non_empty_string(root.get("markdown"), self.name, "fetch", "markdown")
42        return URLFetchCandidate(markdown, markdown)
class ScrapingAntAdapter: 
19class ScrapingAntAdapter:
20    def __init__(
21        self,
22        *,
23        name: str,
24        api_url: str,
25        secret: SecretValue,
26        http_executor: JsonRequester,
27    ) -> None:
28        self.name = name
29        self._api_url = configured_string(api_url, "api_url").rstrip("/")
30        self._secret = secret
31        self._http = http_executor
32
33    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
34        request_url = (
35            f"{endpoint(self._api_url, '/v2/markdown')}?"
36            f"{urlencode({'url': str(url), 'x-api-key': self._secret.reveal()})}"
37        )
38        payload = await self._http.request_json("GET", request_url, stage="fetch")
39        root = require_object(payload, self.name, "fetch", "response")
40        if root.get("error") is not None:
41            raise failure(self.name, "fetch", "provider reported failure")
42        markdown = non_empty_string(root.get("markdown"), self.name, "fetch", "markdown")
43        return URLFetchCandidate(markdown, markdown)
ScrapingAntAdapter( *, name: str, api_url: str, secret: agent_search_gateway.observability.SecretValue, http_executor: agent_search_gateway.providers.web.common.JsonRequester) 
20    def __init__(
21        self,
22        *,
23        name: str,
24        api_url: str,
25        secret: SecretValue,
26        http_executor: JsonRequester,
27    ) -> None:
28        self.name = name
29        self._api_url = configured_string(api_url, "api_url").rstrip("/")
30        self._secret = secret
31        self._http = http_executor
name
async def fetch( self, url: agent_search_gateway.url_normalization.NormalizedURL) -> agent_search_gateway.providers.contracts.URLFetchCandidate: 
33    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
34        request_url = (
35            f"{endpoint(self._api_url, '/v2/markdown')}?"
36            f"{urlencode({'url': str(url), 'x-api-key': self._secret.reveal()})}"
37        )
38        payload = await self._http.request_json("GET", request_url, stage="fetch")
39        root = require_object(payload, self.name, "fetch", "response")
40        if root.get("error") is not None:
41            raise failure(self.name, "fetch", "provider reported failure")
42        markdown = non_empty_string(root.get("markdown"), self.name, "fetch", "markdown")
43        return URLFetchCandidate(markdown, markdown)