Edit on GitHub

agent_search_gateway.providers.web.serpapi

SerpApi Google Search adapter.

View Source

 1"""SerpApi Google Search adapter."""
 2
 3from urllib.parse import urlencode
 4
 5from ...errors import ExecutionFailure
 6from ...observability import SecretValue
 7from ...providers.contracts import KeywordSearchHit
 8from .common import (
 9    JsonRequester,
10    configured_string,
11    endpoint,
12    failure,
13    non_empty_string,
14    optional_string,
15    require_list,
16    require_object,
17)
18
19
20class SerpApiAdapter:
21    def __init__(
22        self,
23        *,
24        name: str,
25        api_url: str,
26        secret: SecretValue,
27        http_executor: JsonRequester,
28    ) -> None:
29        self.name = name
30        self._api_url = configured_string(api_url, "api_url").rstrip("?")
31        self._secret = secret
32        self._http = http_executor
33
34    async def search(self, query: str) -> list[KeywordSearchHit]:
35        request_url = (
36            f"{endpoint(self._api_url, '/search')}?"
37            f"{urlencode({'engine': 'google', 'q': query, 'api_key': self._secret.reveal()})}"
38        )
39        payload = await self._http.request_json("GET", request_url, stage="search")
40        root = require_object(payload, self.name, "search", "response")
41        if root.get("error") is not None:
42            raise failure(self.name, "search", "provider reported failure")
43        results = require_list(root.get("organic_results"), self.name, "search", "organic_results")
44        hits: list[KeywordSearchHit] = []
45        for item in results:
46            try:
47                result = require_object(item, self.name, "search", "result")
48                hits.append(
49                    KeywordSearchHit(
50                        url=non_empty_string(
51                            result.get("link"), self.name, "search", "result.link"
52                        ),
53                        title=optional_string(
54                            result.get("title"), self.name, "search", "result.title"
55                        ),
56                        snippet=optional_string(
57                            result.get("snippet"), self.name, "search", "result.snippet"
58                        ),
59                    )
60                )
61            except ExecutionFailure:
62                continue
63        return hits

class SerpApiAdapter: View Source

21class SerpApiAdapter:
22    def __init__(
23        self,
24        *,
25        name: str,
26        api_url: str,
27        secret: SecretValue,
28        http_executor: JsonRequester,
29    ) -> None:
30        self.name = name
31        self._api_url = configured_string(api_url, "api_url").rstrip("?")
32        self._secret = secret
33        self._http = http_executor
34
35    async def search(self, query: str) -> list[KeywordSearchHit]:
36        request_url = (
37            f"{endpoint(self._api_url, '/search')}?"
38            f"{urlencode({'engine': 'google', 'q': query, 'api_key': self._secret.reveal()})}"
39        )
40        payload = await self._http.request_json("GET", request_url, stage="search")
41        root = require_object(payload, self.name, "search", "response")
42        if root.get("error") is not None:
43            raise failure(self.name, "search", "provider reported failure")
44        results = require_list(root.get("organic_results"), self.name, "search", "organic_results")
45        hits: list[KeywordSearchHit] = []
46        for item in results:
47            try:
48                result = require_object(item, self.name, "search", "result")
49                hits.append(
50                    KeywordSearchHit(
51                        url=non_empty_string(
52                            result.get("link"), self.name, "search", "result.link"
53                        ),
54                        title=optional_string(
55                            result.get("title"), self.name, "search", "result.title"
56                        ),
57                        snippet=optional_string(
58                            result.get("snippet"), self.name, "search", "result.snippet"
59                        ),
60                    )
61                )
62            except ExecutionFailure:
63                continue
64        return hits

SerpApiAdapter( *, name: str, api_url: str, secret: agent_search_gateway.observability.SecretValue, http_executor: agent_search_gateway.providers.web.common.JsonRequester) View Source

22    def __init__(
23        self,
24        *,
25        name: str,
26        api_url: str,
27        secret: SecretValue,
28        http_executor: JsonRequester,
29    ) -> None:
30        self.name = name
31        self._api_url = configured_string(api_url, "api_url").rstrip("?")
32        self._secret = secret
33        self._http = http_executor

name

async def search( self, query: str) -> list[agent_search_gateway.providers.contracts.KeywordSearchHit]: View Source

35    async def search(self, query: str) -> list[KeywordSearchHit]:
36        request_url = (
37            f"{endpoint(self._api_url, '/search')}?"
38            f"{urlencode({'engine': 'google', 'q': query, 'api_key': self._secret.reveal()})}"
39        )
40        payload = await self._http.request_json("GET", request_url, stage="search")
41        root = require_object(payload, self.name, "search", "response")
42        if root.get("error") is not None:
43            raise failure(self.name, "search", "provider reported failure")
44        results = require_list(root.get("organic_results"), self.name, "search", "organic_results")
45        hits: list[KeywordSearchHit] = []
46        for item in results:
47            try:
48                result = require_object(item, self.name, "search", "result")
49                hits.append(
50                    KeywordSearchHit(
51                        url=non_empty_string(
52                            result.get("link"), self.name, "search", "result.link"
53                        ),
54                        title=optional_string(
55                            result.get("title"), self.name, "search", "result.title"
56                        ),
57                        snippet=optional_string(
58                            result.get("snippet"), self.name, "search", "result.snippet"
59                        ),
60                    )
61                )
62            except ExecutionFailure:
63                continue
64        return hits