agent_search_gateway.url_normalization

 1"""Stable URL normalization used for state and concurrency keys."""
 2
 3from typing import NewType
 4from urllib.parse import SplitResult, urlsplit, urlunsplit
 5
 6from .errors import ErrorCode, InputFailure
 7
 8NormalizedURL = NewType("NormalizedURL", str)
 9
10
11def _invalid_url() -> InputFailure:
12    return InputFailure(ErrorCode.INVALID_URL, "URL must be a valid HTTP or HTTPS URL")
13
14
15def _lowercase_host(parsed: SplitResult) -> str:
16    netloc = parsed.netloc
17    userinfo, separator, hostport = netloc.rpartition("@")
18    prefix = f"{userinfo}@" if separator else ""
19
20    if hostport.startswith("["):
21        closing = hostport.find("]")
22        if closing < 0:
23            raise _invalid_url()
24        host = hostport[1:closing].lower()
25        return f"{prefix}[{host}]{hostport[closing + 1 :]}"
26
27    host, port_separator, port = hostport.rpartition(":")
28    if port_separator:
29        return f"{prefix}{host.lower()}:{port}"
30    return f"{prefix}{hostport.lower()}"
31
32
33def normalize_url(value: str) -> NormalizedURL:
34    stripped = value.strip()
35    if not stripped:
36        raise _invalid_url()
37
38    try:
39        parsed = urlsplit(stripped)
40        if parsed.scheme not in {"http", "https"} or parsed.hostname is None:
41            raise _invalid_url()
42        _ = parsed.port
43        netloc = _lowercase_host(parsed)
44    except (ValueError, UnicodeError) as exc:
45        raise _invalid_url() from exc
46
47    return NormalizedURL(
48        urlunsplit((parsed.scheme, netloc, parsed.path, parsed.query, parsed.fragment))
49    )