Edit on GitHub

agent_search_gateway.providers.web.jina

Jina Reader fetch adapter.

 1"""Jina Reader fetch adapter."""
 2
 3from ...providers.contracts import URLFetchCandidate
 4from ...url_normalization import NormalizedURL
 5from .common import TextRequester, configured_string, failure
 6
 7
 8class JinaReaderAdapter:
 9    def __init__(
10        self,
11        *,
12        name: str,
13        api_url: str,
14        http_executor: TextRequester,
15    ) -> None:
16        self.name = name
17        self._api_url = configured_string(api_url, "api_url").rstrip("/")
18        self._http = http_executor
19
20    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
21        text = await self._http.request_text(
22            "POST",
23            self._api_url,
24            stage="fetch",
25            headers={"X-No-Cache": "true"},
26            json_body={"url": str(url)},
27        )
28        if not text.strip():
29            raise failure(self.name, "fetch", "page body is empty")
30        return URLFetchCandidate(text, text)
class JinaReaderAdapter: 
 9class JinaReaderAdapter:
10    def __init__(
11        self,
12        *,
13        name: str,
14        api_url: str,
15        http_executor: TextRequester,
16    ) -> None:
17        self.name = name
18        self._api_url = configured_string(api_url, "api_url").rstrip("/")
19        self._http = http_executor
20
21    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
22        text = await self._http.request_text(
23            "POST",
24            self._api_url,
25            stage="fetch",
26            headers={"X-No-Cache": "true"},
27            json_body={"url": str(url)},
28        )
29        if not text.strip():
30            raise failure(self.name, "fetch", "page body is empty")
31        return URLFetchCandidate(text, text)
JinaReaderAdapter( *, name: str, api_url: str, http_executor: agent_search_gateway.providers.web.common.TextRequester) 
10    def __init__(
11        self,
12        *,
13        name: str,
14        api_url: str,
15        http_executor: TextRequester,
16    ) -> None:
17        self.name = name
18        self._api_url = configured_string(api_url, "api_url").rstrip("/")
19        self._http = http_executor
name
async def fetch( self, url: agent_search_gateway.url_normalization.NormalizedURL) -> agent_search_gateway.providers.contracts.URLFetchCandidate: 
21    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
22        text = await self._http.request_text(
23            "POST",
24            self._api_url,
25            stage="fetch",
26            headers={"X-No-Cache": "true"},
27            json_body={"url": str(url)},
28        )
29        if not text.strip():
30            raise failure(self.name, "fetch", "page body is empty")
31        return URLFetchCandidate(text, text)