"""Shared HTTP-post transport for outbound message-delivery helpers. Slack, Discord, and Telegram delivery modules each issue a JSON ``POST`` to a provider endpoint, parse the response body, and return a ``(success, error, ...)`` tuple. The transport pieces of that flow — making the request, applying a timeout, catching network exceptions, and attempting JSON decode — are identical; only the success criteria, authentication scheme, and error-message extraction differ per provider. This module hosts the shared transport so each delivery module can keep its provider-specific payload building and result interpretation while sharing one well-tested HTTP code path. The helper deliberately does **not** decide whether the call succeeded at the provider level. It returns ``ok=True`` whenever the request completed without raising; callers then inspect ``status_code`` and ``data``/``text`` to apply provider semantics (e.g. ``data["ok"]`` for Slack, ``status_code in (200, 201)`` for Discord, ``status_code == 200`` for Telegram). """ from __future__ import annotations from collections.abc import Mapping from dataclasses import dataclass, field from types import MappingProxyType from typing import Any import httpx @dataclass(frozen=True) class DeliveryResponse: """Normalized result of a delivery POST. Attributes: ok: ``True`` iff the request completed without raising. This is a transport-level signal only; provider-level success requires inspecting ``status_code`` / ``data`` per provider semantics. status_code: HTTP status code from the response, or ``0`` when the request itself raised before a response was received. data: Parsed JSON body when the response was a JSON object, otherwise an empty mapping. Never ``None``, so callers can chain ``.get(...)`` safely without a None-check. The mapping is read-only (``MappingProxyType``) so the frozen dataclass stays fully immutable end-to-end. text: Raw response body, useful for fallback error extraction when the body is not valid JSON or is empty. error: String form of the exception that aborted the request. Empty when ``ok`` is True. exc_type: Class name of the exception that aborted the request (e.g. ``"TimeoutError"``, ``"ConnectError"``). Empty when ``ok`` is True. Surfaced separately so callers can include the exception shape in triage logs without parsing ``error``. Named ``exc_type`` rather than ``type`` because ``type`` is a Python builtin. """ ok: bool status_code: int = 0 data: Mapping[str, Any] = field(default_factory=dict) text: str = "" error: str = "" exc_type: str = "" def __post_init__(self) -> None: # Wrap ``data`` in a read-only view so callers cannot mutate the # response after the fact and break the frozen-dataclass contract. # ``object.__setattr__`` is required because ``frozen=True`` blocks # normal attribute assignment. if not isinstance(self.data, MappingProxyType): object.__setattr__(self, "data", MappingProxyType(dict(self.data))) def post_form( url: str, data: dict[str, str], *, auth: tuple[str, str] | None = None, headers: dict[str, str] | None = None, timeout: float = 15.0, follow_redirects: bool = False, ) -> DeliveryResponse: """POST ``data`` as form-encoded to ``url`` and return a normalized result. Mirrors :func:`post_json` but sends ``application/x-www-form-urlencoded`` bodies with optional HTTP Basic Auth — the shape required by Twilio and WhatsApp delivery endpoints. Args: url: Absolute URL to post to. data: Form fields to encode in the request body. auth: Optional ``(username, password)`` tuple for HTTP Basic Auth. headers: Optional extra headers. timeout: Request timeout in seconds. Defaults to 15s. follow_redirects: Whether to follow 3xx redirects. Returns: ``DeliveryResponse`` with ``ok``, ``status_code``, ``data``, ``text``, and ``error`` populated. """ try: response = httpx.post( url, data=data, auth=auth, headers=headers or {}, timeout=timeout, follow_redirects=follow_redirects, ) except Exception as exc: return DeliveryResponse(ok=False, error=str(exc), exc_type=type(exc).__name__) text = response.text parsed_data: dict[str, Any] = {} try: parsed = response.json() if isinstance(parsed, dict): parsed_data = parsed except Exception: # non-JSON body is permitted; fall through with empty data pass return DeliveryResponse( ok=True, status_code=response.status_code, data=parsed_data, text=text, ) def post_json( url: str, payload: dict[str, Any], *, headers: dict[str, str] | None = None, timeout: float = 15.0, follow_redirects: bool = False, ) -> DeliveryResponse: """POST ``payload`` as JSON to ``url`` and return a normalized result. On request exceptions the result carries ``ok=False`` and ``error`` set to the exception message — callers are not expected to handle raised errors. The transport never re-raises. Args: url: Absolute URL to post to. payload: JSON-serializable dict body. headers: Optional headers (e.g. ``Authorization``). Defaults to an empty dict; httpx will still set ``Content-Type`` and the standard headers it manages. timeout: Request timeout in seconds. Defaults to 15s, matching the pre-existing per-provider timeouts. follow_redirects: Whether to follow 3xx redirects. Disabled by default to match Slack/Discord/Telegram REST APIs (which never redirect on success). Slack incoming webhooks and the NextJS ``/api/slack`` proxy enable it. Returns: ``DeliveryResponse`` with ``ok``, ``status_code``, ``data``, ``text``, and ``error`` populated. JSON decode failures are non-fatal: ``data`` falls back to ``{}`` and ``text`` always carries the raw body. """ try: response = httpx.post( url, json=payload, headers=headers or {}, timeout=timeout, follow_redirects=follow_redirects, ) except Exception as exc: return DeliveryResponse(ok=False, error=str(exc), exc_type=type(exc).__name__) text = response.text data: dict[str, Any] = {} try: parsed = response.json() if isinstance(parsed, dict): data = parsed except Exception: # non-JSON body is permitted; fall through with empty data pass return DeliveryResponse( ok=True, status_code=response.status_code, data=data, text=text, )