Edit on GitHub

agent_search_gateway.providers.web.common

Shared pure helpers for built-in web provider adapters.

View Source

 1"""Shared pure helpers for built-in web provider adapters."""
 2
 3from collections.abc import Mapping
 4from typing import Protocol
 5
 6from ...errors import ErrorCode, ExecutionFailure
 7from ...url_normalization import NormalizedURL, normalize_url
 8
 9
10class JsonRequester(Protocol):
11    async def request_json(
12        self,
13        method: str,
14        url: str,
15        *,
16        stage: str,
17        headers: Mapping[str, str] | None = None,
18        json_body: object | None = None,
19    ) -> object: ...
20
21
22class TextRequester(Protocol):
23    async def request_text(
24        self,
25        method: str,
26        url: str,
27        *,
28        stage: str,
29        headers: Mapping[str, str] | None = None,
30        json_body: object | None = None,
31    ) -> str: ...
32
33
34class HttpRequester(JsonRequester, TextRequester, Protocol):
35    pass
36
37
38def endpoint(base_url: str, suffix: str) -> str:
39    return f"{base_url.rstrip('/')}/{suffix.lstrip('/')}"
40
41
42def configured_string(value: object, label: str) -> str:
43    if not isinstance(value, str) or not value.strip():
44        raise TypeError(f"{label} must be a non-empty string")
45    return value.strip()
46
47
48def failure(
49    provider: str,
50    stage: str,
51    reason: str,
52    *,
53    reason_code: str | None = None,
54) -> ExecutionFailure:
55    return ExecutionFailure(
56        ErrorCode.ALL_PROVIDERS_FAILED,
57        f"{provider}/{stage}: {reason}",
58        reason=reason_code,
59    )
60
61
62def require_object(value: object, provider: str, stage: str, label: str) -> dict[str, object]:
63    if not isinstance(value, dict):
64        raise failure(provider, stage, f"{label} must be an object")
65    return value
66
67
68def require_list(value: object, provider: str, stage: str, label: str) -> list[object]:
69    if not isinstance(value, list):
70        raise failure(provider, stage, f"{label} must be an array")
71    return value
72
73
74def require_string(value: object, provider: str, stage: str, label: str) -> str:
75    if not isinstance(value, str):
76        raise failure(provider, stage, f"{label} must be a string")
77    return value
78
79
80def non_empty_string(value: object, provider: str, stage: str, label: str) -> str:
81    text = require_string(value, provider, stage, label)
82    if not text.strip():
83        raise failure(provider, stage, f"{label} must be non-empty")
84    return text
85
86
87def optional_string(value: object, provider: str, stage: str, label: str) -> str:
88    if value is None:
89        return ""
90    return require_string(value, provider, stage, label)
91
92
93def normalized_match(candidate: object, target: NormalizedURL, provider: str, stage: str) -> bool:
94    text = non_empty_string(candidate, provider, stage, "result.url")
95    try:
96        return normalize_url(text) == target
97    except Exception as exc:
98        raise failure(provider, stage, "result URL is invalid") from exc

class JsonRequester(typing.Protocol): View Source

11class JsonRequester(Protocol):
12    async def request_json(
13        self,
14        method: str,
15        url: str,
16        *,
17        stage: str,
18        headers: Mapping[str, str] | None = None,
19        json_body: object | None = None,
20    ) -> object: ...

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as::

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

JsonRequester(*args, **kwargs) View Source

1739def _no_init_or_replace_init(self, *args, **kwargs):
1740    cls = type(self)
1741
1742    if cls._is_protocol:
1743        raise TypeError('Protocols cannot be instantiated')
1744
1745    # Already using a custom `__init__`. No need to calculate correct
1746    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1747    if cls.__init__ is not _no_init_or_replace_init:
1748        return
1749
1750    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1751    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1752    # searches for a proper new `__init__` in the MRO. The new `__init__`
1753    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1754    # instantiation of the protocol subclass will thus use the new
1755    # `__init__` and no longer call `_no_init_or_replace_init`.
1756    for base in cls.__mro__:
1757        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1758        if init is not _no_init_or_replace_init:
1759            cls.__init__ = init
1760            break
1761    else:
1762        # should not happen
1763        cls.__init__ = object.__init__
1764
1765    cls.__init__(self, *args, **kwargs)

async def request_json( self, method: str, url: str, *, stage: str, headers: Mapping[str, str] | None = None, json_body: object | None = None) -> object: View Source

12    async def request_json(
13        self,
14        method: str,
15        url: str,
16        *,
17        stage: str,
18        headers: Mapping[str, str] | None = None,
19        json_body: object | None = None,
20    ) -> object: ...

class TextRequester(typing.Protocol): View Source

23class TextRequester(Protocol):
24    async def request_text(
25        self,
26        method: str,
27        url: str,
28        *,
29        stage: str,
30        headers: Mapping[str, str] | None = None,
31        json_body: object | None = None,
32    ) -> str: ...

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

TextRequester(*args, **kwargs) View Source

1739def _no_init_or_replace_init(self, *args, **kwargs):
1740    cls = type(self)
1741
1742    if cls._is_protocol:
1743        raise TypeError('Protocols cannot be instantiated')
1744
1745    # Already using a custom `__init__`. No need to calculate correct
1746    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1747    if cls.__init__ is not _no_init_or_replace_init:
1748        return
1749
1750    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1751    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1752    # searches for a proper new `__init__` in the MRO. The new `__init__`
1753    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1754    # instantiation of the protocol subclass will thus use the new
1755    # `__init__` and no longer call `_no_init_or_replace_init`.
1756    for base in cls.__mro__:
1757        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1758        if init is not _no_init_or_replace_init:
1759            cls.__init__ = init
1760            break
1761    else:
1762        # should not happen
1763        cls.__init__ = object.__init__
1764
1765    cls.__init__(self, *args, **kwargs)

async def request_text( self, method: str, url: str, *, stage: str, headers: Mapping[str, str] | None = None, json_body: object | None = None) -> str: View Source

24    async def request_text(
25        self,
26        method: str,
27        url: str,
28        *,
29        stage: str,
30        headers: Mapping[str, str] | None = None,
31        json_body: object | None = None,
32    ) -> str: ...

class HttpRequester(JsonRequester, TextRequester, typing.Protocol): View Source

35class HttpRequester(JsonRequester, TextRequester, Protocol):
36    pass

Base class for protocol classes.

Protocol classes are defined as::

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example::

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

Inherited Members

JsonRequester: JsonRequester; request_json
TextRequester: request_text

def endpoint(base_url: str, suffix: str) -> str: View Source

39def endpoint(base_url: str, suffix: str) -> str:
40    return f"{base_url.rstrip('/')}/{suffix.lstrip('/')}"

def configured_string(value: object, label: str) -> str: View Source

43def configured_string(value: object, label: str) -> str:
44    if not isinstance(value, str) or not value.strip():
45        raise TypeError(f"{label} must be a non-empty string")
46    return value.strip()

def failure( provider: str, stage: str, reason: str, *, reason_code: str | None = None) -> agent_search_gateway.errors.ExecutionFailure: View Source

49def failure(
50    provider: str,
51    stage: str,
52    reason: str,
53    *,
54    reason_code: str | None = None,
55) -> ExecutionFailure:
56    return ExecutionFailure(
57        ErrorCode.ALL_PROVIDERS_FAILED,
58        f"{provider}/{stage}: {reason}",
59        reason=reason_code,
60    )

def require_object( value: object, provider: str, stage: str, label: str) -> dict[str, object]: View Source

63def require_object(value: object, provider: str, stage: str, label: str) -> dict[str, object]:
64    if not isinstance(value, dict):
65        raise failure(provider, stage, f"{label} must be an object")
66    return value

def require_list(value: object, provider: str, stage: str, label: str) -> list[object]: View Source

69def require_list(value: object, provider: str, stage: str, label: str) -> list[object]:
70    if not isinstance(value, list):
71        raise failure(provider, stage, f"{label} must be an array")
72    return value

def require_string(value: object, provider: str, stage: str, label: str) -> str: View Source

75def require_string(value: object, provider: str, stage: str, label: str) -> str:
76    if not isinstance(value, str):
77        raise failure(provider, stage, f"{label} must be a string")
78    return value

def non_empty_string(value: object, provider: str, stage: str, label: str) -> str: View Source

81def non_empty_string(value: object, provider: str, stage: str, label: str) -> str:
82    text = require_string(value, provider, stage, label)
83    if not text.strip():
84        raise failure(provider, stage, f"{label} must be non-empty")
85    return text

def optional_string(value: object, provider: str, stage: str, label: str) -> str: View Source

88def optional_string(value: object, provider: str, stage: str, label: str) -> str:
89    if value is None:
90        return ""
91    return require_string(value, provider, stage, label)

def normalized_match( candidate: object, target: agent_search_gateway.url_normalization.NormalizedURL, provider: str, stage: str) -> bool: View Source

94def normalized_match(candidate: object, target: NormalizedURL, provider: str, stage: str) -> bool:
95    text = non_empty_string(candidate, provider, stage, "result.url")
96    try:
97        return normalize_url(text) == target
98    except Exception as exc:
99        raise failure(provider, stage, "result URL is invalid") from exc