Spaces:
Running
Running
File size: 3,441 Bytes
6e64a91 3d8032f 6e64a91 3d8032f 6e64a91 3d8032f 6e64a91 3d8032f 6e64a91 3d8032f 6e64a91 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 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 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | 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)
@field_validator("from_number", "to")
@classmethod
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
@field_validator("content")
@classmethod
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
@router.post("/send")
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}
|