"""Per-request replay controls, carried through HTTPX request extensions."""
from dataclasses import dataclass
import httpx
POLICY_EXTENSION = "sf_toolkit.request_policy"
[docs]
@dataclass(frozen=True)
class RequestPolicy:
"""Control retries, expired-session replay and redirects independently.
Use ``extensions=NO_REPLAY.extensions`` with any HTTPX request method.
This controls toolkit transports, authentication and clients, not retries
implemented inside custom transports or proxies.
Initial authentication is still allowed: it does not replay the resource call.
"""
allow_retries: bool = True
allow_auth_replay: bool = True
allow_redirects: bool = True
@property
def extensions(self) -> dict[str, object]:
return {POLICY_EXTENSION: self}
NO_REPLAY = RequestPolicy(
allow_retries=False, allow_auth_replay=False, allow_redirects=False
)
DEFAULT_POLICY = RequestPolicy()
[docs]
def request_policy(request: httpx.Request) -> RequestPolicy:
policy = request.extensions.get(POLICY_EXTENSION, DEFAULT_POLICY)
if not isinstance(policy, RequestPolicy):
raise TypeError(f"{POLICY_EXTENSION} must be a RequestPolicy")
return policy