Edit on GitHub

agent_search_gateway.providers.web.brightdata

Bright Data SERP and Web Unlocker adapter.

View Source

 1"""Bright Data SERP and Web Unlocker adapter."""
 2
 3from urllib.parse import urlencode
 4
 5from ...errors import ExecutionFailure
 6from ...observability import SecretValue
 7from ...providers.contracts import KeywordSearchHit, URLFetchCandidate
 8from ...url_normalization import NormalizedURL
 9from .common import (
10    HttpRequester,
11    configured_string,
12    endpoint,
13    failure,
14    non_empty_string,
15    optional_string,
16    require_list,
17    require_object,
18)
19
20
21class BrightDataAdapter:
22    def __init__(
23        self,
24        *,
25        name: str,
26        api_url: str,
27        search_zone: str,
28        fetch_zone: str,
29        secret: SecretValue,
30        http_executor: HttpRequester,
31    ) -> None:
32        self.name = name
33        self._api_url = configured_string(api_url, "api_url").rstrip("/")
34        self._search_zone = configured_string(search_zone, "search_zone")
35        self._fetch_zone = configured_string(fetch_zone, "fetch_zone")
36        self._secret = secret
37        self._http = http_executor
38
39    async def search(self, query: str) -> list[KeywordSearchHit]:
40        google_url = "https://www.google.com/search?" + urlencode({"q": query, "brd_json": "1"})
41        payload = await self._http.request_json(
42            "POST",
43            endpoint(self._api_url, "/request"),
44            stage="search",
45            headers={"Authorization": f"Bearer {self._secret.reveal()}"},
46            json_body={"zone": self._search_zone, "url": google_url, "format": "raw"},
47        )
48        root = require_object(payload, self.name, "search", "response")
49        if root.get("error") is not None:
50            raise failure(self.name, "search", "provider reported failure")
51        results = require_list(root.get("organic"), self.name, "search", "organic")
52        hits: list[KeywordSearchHit] = []
53        for item in results:
54            try:
55                result = require_object(item, self.name, "search", "result")
56                hits.append(
57                    KeywordSearchHit(
58                        url=non_empty_string(
59                            result.get("link"), self.name, "search", "result.link"
60                        ),
61                        title=optional_string(
62                            result.get("title"), self.name, "search", "result.title"
63                        ),
64                        snippet=optional_string(
65                            result.get("description"), self.name, "search", "result.description"
66                        ),
67                    )
68                )
69            except ExecutionFailure:
70                continue
71        return hits
72
73    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
74        text = await self._http.request_text(
75            "POST",
76            endpoint(self._api_url, "/request"),
77            stage="fetch",
78            headers={"Authorization": f"Bearer {self._secret.reveal()}"},
79            json_body={
80                "zone": self._fetch_zone,
81                "url": str(url),
82                "format": "raw",
83                "data_format": "markdown",
84            },
85        )
86        if not text.strip():
87            raise failure(self.name, "fetch", "page body is empty")
88        return URLFetchCandidate(text, text)

class BrightDataAdapter: View Source

22class BrightDataAdapter:
23    def __init__(
24        self,
25        *,
26        name: str,
27        api_url: str,
28        search_zone: str,
29        fetch_zone: str,
30        secret: SecretValue,
31        http_executor: HttpRequester,
32    ) -> None:
33        self.name = name
34        self._api_url = configured_string(api_url, "api_url").rstrip("/")
35        self._search_zone = configured_string(search_zone, "search_zone")
36        self._fetch_zone = configured_string(fetch_zone, "fetch_zone")
37        self._secret = secret
38        self._http = http_executor
39
40    async def search(self, query: str) -> list[KeywordSearchHit]:
41        google_url = "https://www.google.com/search?" + urlencode({"q": query, "brd_json": "1"})
42        payload = await self._http.request_json(
43            "POST",
44            endpoint(self._api_url, "/request"),
45            stage="search",
46            headers={"Authorization": f"Bearer {self._secret.reveal()}"},
47            json_body={"zone": self._search_zone, "url": google_url, "format": "raw"},
48        )
49        root = require_object(payload, self.name, "search", "response")
50        if root.get("error") is not None:
51            raise failure(self.name, "search", "provider reported failure")
52        results = require_list(root.get("organic"), self.name, "search", "organic")
53        hits: list[KeywordSearchHit] = []
54        for item in results:
55            try:
56                result = require_object(item, self.name, "search", "result")
57                hits.append(
58                    KeywordSearchHit(
59                        url=non_empty_string(
60                            result.get("link"), self.name, "search", "result.link"
61                        ),
62                        title=optional_string(
63                            result.get("title"), self.name, "search", "result.title"
64                        ),
65                        snippet=optional_string(
66                            result.get("description"), self.name, "search", "result.description"
67                        ),
68                    )
69                )
70            except ExecutionFailure:
71                continue
72        return hits
73
74    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
75        text = await self._http.request_text(
76            "POST",
77            endpoint(self._api_url, "/request"),
78            stage="fetch",
79            headers={"Authorization": f"Bearer {self._secret.reveal()}"},
80            json_body={
81                "zone": self._fetch_zone,
82                "url": str(url),
83                "format": "raw",
84                "data_format": "markdown",
85            },
86        )
87        if not text.strip():
88            raise failure(self.name, "fetch", "page body is empty")
89        return URLFetchCandidate(text, text)

BrightDataAdapter( *, name: str, api_url: str, search_zone: str, fetch_zone: str, secret: agent_search_gateway.observability.SecretValue, http_executor: agent_search_gateway.providers.web.common.HttpRequester) View Source

23    def __init__(
24        self,
25        *,
26        name: str,
27        api_url: str,
28        search_zone: str,
29        fetch_zone: str,
30        secret: SecretValue,
31        http_executor: HttpRequester,
32    ) -> None:
33        self.name = name
34        self._api_url = configured_string(api_url, "api_url").rstrip("/")
35        self._search_zone = configured_string(search_zone, "search_zone")
36        self._fetch_zone = configured_string(fetch_zone, "fetch_zone")
37        self._secret = secret
38        self._http = http_executor

name

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

40    async def search(self, query: str) -> list[KeywordSearchHit]:
41        google_url = "https://www.google.com/search?" + urlencode({"q": query, "brd_json": "1"})
42        payload = await self._http.request_json(
43            "POST",
44            endpoint(self._api_url, "/request"),
45            stage="search",
46            headers={"Authorization": f"Bearer {self._secret.reveal()}"},
47            json_body={"zone": self._search_zone, "url": google_url, "format": "raw"},
48        )
49        root = require_object(payload, self.name, "search", "response")
50        if root.get("error") is not None:
51            raise failure(self.name, "search", "provider reported failure")
52        results = require_list(root.get("organic"), self.name, "search", "organic")
53        hits: list[KeywordSearchHit] = []
54        for item in results:
55            try:
56                result = require_object(item, self.name, "search", "result")
57                hits.append(
58                    KeywordSearchHit(
59                        url=non_empty_string(
60                            result.get("link"), self.name, "search", "result.link"
61                        ),
62                        title=optional_string(
63                            result.get("title"), self.name, "search", "result.title"
64                        ),
65                        snippet=optional_string(
66                            result.get("description"), self.name, "search", "result.description"
67                        ),
68                    )
69                )
70            except ExecutionFailure:
71                continue
72        return hits

async def fetch( self, url: agent_search_gateway.url_normalization.NormalizedURL) -> agent_search_gateway.providers.contracts.URLFetchCandidate: View Source

74    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
75        text = await self._http.request_text(
76            "POST",
77            endpoint(self._api_url, "/request"),
78            stage="fetch",
79            headers={"Authorization": f"Bearer {self._secret.reveal()}"},
80            json_body={
81                "zone": self._fetch_zone,
82                "url": str(url),
83                "format": "raw",
84                "data_format": "markdown",
85            },
86        )
87        if not text.strip():
88            raise failure(self.name, "fetch", "page body is empty")
89        return URLFetchCandidate(text, text)