Spaces:
Running
Running
| from __future__ import annotations | |
| import os | |
| import re | |
| from typing import Any | |
| from urllib.parse import urlsplit | |
| import httpx | |
| from fastapi import APIRouter, HTTPException | |
| from pydantic import BaseModel, ConfigDict, Field, field_validator | |
| router = APIRouter(prefix="/v1/sms", tags=["sms"]) | |
| _PHONE_RE = re.compile(r"^\+[1-9][0-9]{7,14}$") | |
| _MAX_CONTENT_LENGTH = 1600 | |
| _DEFAULT_BASE_URL = "https://api.httpsms.com" | |
| class SendSMSRequest(BaseModel): | |
| model_config = ConfigDict(populate_by_name=True, extra="forbid") | |
| from_number: str = Field(alias="from", min_length=9, max_length=16) | |
| to: str = Field(min_length=9, max_length=16) | |
| content: str = Field(min_length=1, max_length=_MAX_CONTENT_LENGTH) | |
| def validate_phone_number(cls, value: str) -> str: | |
| value = value.strip() | |
| if not _PHONE_RE.fullmatch(value): | |
| raise ValueError("Phone numbers must use international E.164 format, e.g. +2348012345678") | |
| return value | |
| def validate_content(cls, value: str) -> str: | |
| value = value.strip() | |
| if not value: | |
| raise ValueError("SMS content must not be blank") | |
| if any(ord(char) < 32 and char not in "\n\r\t" for char in value): | |
| raise ValueError("SMS content contains unsupported control characters") | |
| return value | |
| def _configuration() -> tuple[str, str]: | |
| api_key = os.getenv("HTTPSMS_API_KEY", "").strip() | |
| if not api_key: | |
| raise HTTPException(status_code=503, detail="httpSMS integration is not configured") | |
| base_url = os.getenv("HTTPSMS_API_BASE_URL", _DEFAULT_BASE_URL).strip().rstrip("/") | |
| parsed = urlsplit(base_url) | |
| if ( | |
| parsed.scheme != "https" | |
| or not parsed.hostname | |
| or parsed.username is not None | |
| or parsed.password is not None | |
| or parsed.query | |
| or parsed.fragment | |
| ): | |
| raise HTTPException(status_code=503, detail="httpSMS base URL configuration is invalid") | |
| return base_url, api_key | |
| async def send_sms(payload: SendSMSRequest) -> dict[str, Any]: | |
| """Send an SMS through the configured httpSMS API.""" | |
| base_url, api_key = _configuration() | |
| request_body = {"from": payload.from_number, "to": payload.to, "content": payload.content} | |
| try: | |
| async with httpx.AsyncClient( | |
| timeout=httpx.Timeout(20.0, connect=5.0), follow_redirects=False | |
| ) as client: | |
| response = await client.post( | |
| f"{base_url}/v1/messages/send", | |
| headers={"x-api-key": api_key, "Content-Type": "application/json"}, | |
| json=request_body, | |
| ) | |
| except httpx.TimeoutException as exc: | |
| raise HTTPException(status_code=504, detail="httpSMS request timed out") from exc | |
| except httpx.RequestError as exc: | |
| raise HTTPException(status_code=502, detail="Unable to reach the httpSMS API") from exc | |
| if not 200 <= response.status_code < 300: | |
| raise HTTPException( | |
| status_code=502, | |
| detail=f"httpSMS rejected the request (upstream status {response.status_code})", | |
| ) | |
| try: | |
| upstream: Any = response.json() | |
| except ValueError: | |
| upstream = {"message": response.text[:500]} if response.text else {} | |
| return {"success": True, "upstream_status": response.status_code, "result": upstream} | |