saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
"""Twilio SMS provider via Twilio's plain REST API (no SDK dependency -
|
|
just an authenticated POST). No-ops with a clear error if Twilio isn't
|
|
configured; the caller (alert_service) is what additionally gates on
|
|
`NOTIFICATION_SMS_ENABLED` before ever constructing this provider.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
|
|
from app.core.config import Settings
|
|
from app.notifications.base import DeliveryResult, NotificationMessage
|
|
|
|
_API_BASE = "https://api.twilio.com/2010-04-01"
|
|
|
|
|
|
class TwilioSmsProvider:
|
|
provider_name = "twilio_sms"
|
|
|
|
def __init__(self, settings: Settings) -> None:
|
|
self._settings = settings
|
|
|
|
async def send(self, message: NotificationMessage) -> DeliveryResult:
|
|
settings = self._settings
|
|
if not (
|
|
settings.twilio_account_sid
|
|
and settings.twilio_auth_token
|
|
and settings.twilio_from_number
|
|
):
|
|
return DeliveryResult(success=False, error="Twilio is not configured")
|
|
|
|
url = f"{_API_BASE}/Accounts/{settings.twilio_account_sid}/Messages.json"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
response = await client.post(
|
|
url,
|
|
auth=(settings.twilio_account_sid, settings.twilio_auth_token),
|
|
data={
|
|
"From": settings.twilio_from_number,
|
|
"To": message.destination_value,
|
|
"Body": message.body_text,
|
|
},
|
|
)
|
|
except httpx.HTTPError as exc:
|
|
return DeliveryResult(success=False, error=str(exc))
|
|
|
|
if response.status_code >= 400:
|
|
return DeliveryResult(
|
|
success=False, error=f"Twilio error {response.status_code}: {response.text[:200]}"
|
|
)
|
|
data = response.json()
|
|
return DeliveryResult(success=True, external_message_id=data.get("sid"))
|