File size: 2,682 Bytes
102dd4f dc8cb49 102dd4f dc8cb49 102dd4f dc8cb49 102dd4f dc8cb49 | 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 | import httpx
from app.config import Settings
class WhatsAppClientError(RuntimeError):
pass
class WhatsAppClient:
def __init__(self, settings: Settings, client: httpx.AsyncClient | None = None) -> None:
self._settings = settings
self._client = client
@property
def _messages_url(self) -> str:
base = self._settings.whatsapp_graph_url.rstrip("/")
version = self._settings.whatsapp_api_version.strip("/")
phone_number_id = self._settings.whatsapp_phone_number_id
print(base)
return f"{base}/{version}/{phone_number_id}/messages"
async def send_text(self, remoteJid: str, text: str) -> dict:
payload = {
"messaging_product": "whatsapp",
"recipient_type": "individual",
"to": remoteJid,
"type": "text",
"text": {"preview_url": False, "body": text},
}
headers = {
"Authorization": f"Bearer {self._settings.whatsapp_access_token}",
"Content-Type": "application/json",
}
if self._client is not None:
response = await self._client.post(self._messages_url, json=payload, headers=headers)
else:
async with httpx.AsyncClient(timeout=self._settings.request_timeout_seconds) as client:
response = await client.post(self._messages_url, json=payload, headers=headers)
if response.status_code >= 400:
raise WhatsAppClientError(
f"WhatsApp API failed with {response.status_code}: {response.text}"
)
return response.json()
async def send_interactive_list(
self,
remoteJid: str,
interactive: dict[str, Any],
) -> dict:
payload = {
"messaging_product": "whatsapp",
"recipient_type": "individual",
"to": remoteJid,
"type": "interactive",
"interactive": interactive,
}
headers = {
"Authorization": f"Bearer {self._settings.whatsapp_access_token}",
"Content-Type": "application/json",
}
if self._client is not None:
response = await self._client.post(self._messages_url, json=payload, headers=headers)
else:
async with httpx.AsyncClient(timeout=self._settings.request_timeout_seconds) as client:
response = await client.post(self._messages_url, json=payload, headers=headers)
if response.status_code >= 400:
raise WhatsAppClientError(
f"WhatsApp API failed with {response.status_code}: {response.text}"
)
return response.json()
|