Skip to content

http

azure_bootstrap.http

Hardened outbound HTTP client — sync requests stack.

Modules:

Name Description
async_client

Optional httpx async HTTP client stack.

Functions:

Name Description
build_session

Return a requests.Session with urllib3 Retry mounted.
request_with_retry

Issue an HTTP request with default timeout, traceparent, and SSRF guard.
normalize_pem

Normalize PEM strings mangled by env/Key Vault (spaces, CRLF).
write_temp_pem

Write normalized PEM to a temp file; caller owns cleanup.

build_session 

build_session(*, total_retries: int = 5, backoff_factor: float = 1.0, pool_connections: int = 10, pool_maxsize: int = 10) -> Any

Return a requests.Session with urllib3 Retry mounted.

Source code in azure_bootstrap/http/__init__.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def build_session(
    *,
    total_retries: int = 5,
    backoff_factor: float = 1.0,
    pool_connections: int = 10,
    pool_maxsize: int = 10,
) -> Any:
    """Return a ``requests.Session`` with urllib3 Retry mounted."""
    import requests  # type: ignore[import-untyped]
    from requests.adapters import HTTPAdapter  # type: ignore[import-untyped]
    from urllib3.util.retry import Retry

    retry = Retry(
        total=total_retries,
        backoff_factor=backoff_factor,
        backoff_jitter=0.3,
        status_forcelist=(408, 429, 500, 502, 503, 504),
        allowed_methods=frozenset(["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD"]),
        respect_retry_after_header=True,
        raise_on_status=False,
    )
    session = requests.Session()
    adapter = HTTPAdapter(
        max_retries=retry,
        pool_connections=pool_connections,
        pool_maxsize=pool_maxsize,
    )
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session

request_with_retry 

request_with_retry(method: str, url: str, *, session: Any | None = None, timeout: float = _DEFAULT_TIMEOUT, headers: dict[str, str] | None = None, allow_private: bool = False, **kwargs: Any) -> Any

Issue an HTTP request with default timeout, traceparent, and SSRF guard.

Source code in azure_bootstrap/http/__init__.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def request_with_retry(
    method: str,
    url: str,
    *,
    session: Any | None = None,
    timeout: float = _DEFAULT_TIMEOUT,
    headers: dict[str, str] | None = None,
    allow_private: bool = False,
    **kwargs: Any,
) -> Any:
    """Issue an HTTP request with default timeout, traceparent, and SSRF guard."""
    check_ssrf(url, allow_private=allow_private)
    sess = session or build_session()
    hdrs = inject_traceparent(headers)
    return sess.request(method.upper(), url, headers=hdrs, timeout=timeout, **kwargs)

normalize_pem 

normalize_pem(pem: str) -> str

Normalize PEM strings mangled by env/Key Vault (spaces, CRLF).

Source code in azure_bootstrap/http/__init__.py 
66
67
68
69
70
71
def normalize_pem(pem: str) -> str:
    """Normalize PEM strings mangled by env/Key Vault (spaces, CRLF)."""
    text = pem.strip().replace("\\n", "\n")
    if "BEGIN" not in text:
        text = re.sub(r"\s+", "\n", text)
    return text.replace("\r\n", "\n")

write_temp_pem 

write_temp_pem(pem: str, *, suffix: str = '.pem') -> str

Write normalized PEM to a temp file; caller owns cleanup.

Source code in azure_bootstrap/http/__init__.py 
74
75
76
77
78
79
def write_temp_pem(pem: str, *, suffix: str = ".pem") -> str:
    """Write normalized PEM to a temp file; caller owns cleanup."""
    path = tempfile.NamedTemporaryFile(mode="w", suffix=suffix, delete=False)
    path.write(normalize_pem(pem))
    path.close()
    return path.name