Python SDK

pip install splashifypro
# or with poetry
poetry add splashifypro

import os
from splashifypro import Splashify
 
splashify = Splashify(
    api_key=os.environ["SPLASHIFY_API_KEY"],
    # optional:
    base_url="https://apis.splashifypro.com",
    timeout=30,        # seconds
    retries=3,         # 5xx auto-retry with backoff
)

import os
from splashifypro import Splashify
 
splashify = Splashify(api_key=os.environ["SPLASHIFY_API_KEY"])
 
result = splashify.messages.send_text(
    to="919876543210",
    text="Hello from Splashify Pro!",
)
 
print(f"Sent: {result.message_id}")

import asyncio
from splashifypro.aio import AsyncSplashify
 
async def main():
    splashify = AsyncSplashify(api_key=os.environ["SPLASHIFY_API_KEY"])
    result = await splashify.messages.send_text(
        to="919876543210",
        text="Hello async!",
    )
    print(result.message_id)
 
asyncio.run(main())

splashify.messages.send_template(
    to="919876543210",
    template_name="order_confirmation",
    language="en",
    body_parameters=["Aditya", "#1234", "₹2,499"],
)

splashify.messages.send_media(
    to="919876543210",
    type="image",
    media_url="https://your-cdn.com/photo.jpg",
    caption="Check this out!",
)

broadcast = splashify.broadcasts.create(
    name="Diwali sale",
    template_name="diwali_offer_v1",
    language="en",
    audience_type="segments",
    segment_ids=["<segment-uuid>"],
)
print(f"Broadcast {broadcast.id} queued")
 
report = splashify.broadcasts.report(broadcast.id)
print(f"{report.sent}/{report.total} sent")

cursor = None
while True:
    page = splashify.contacts.list(limit=100, cursor=cursor)
    for c in page.edges:
        print(c.phone, c.name)
    if not page.page_info.has_next_page:
        break
    cursor = page.page_info.end_cursor

from flask import Flask, request
from splashifypro import verify_webhook
 
app = Flask(__name__)
 
@app.post("/splashify/webhook")
def webhook():
    if not verify_webhook(
        raw_body=request.get_data(),
        signature=request.headers.get("X-Splashify-Signature", ""),
        secret=os.environ["SPLASHIFY_WEBHOOK_SECRET"],
    ):
        return "", 401
    event = request.get_json()
    print(f"Event: {event['type']}", event["payload"])
    return "", 200

from fastapi import FastAPI, Request, HTTPException
from splashifypro import verify_webhook
 
app = FastAPI()
 
@app.post("/splashify/webhook")
async def webhook(request: Request):
    body = await request.body()
    if not verify_webhook(
        raw_body=body,
        signature=request.headers.get("x-splashify-signature", ""),
        secret=os.environ["SPLASHIFY_WEBHOOK_SECRET"],
    ):
        raise HTTPException(401)
    event = await request.json()
    return {"received": True}

from splashifypro import (
    SplashifyError,
    SplashifyRateLimitError,
    SplashifyAuthError,
    SplashifyValidationError,
    SplashifyNotFoundError,
)
 
try:
    splashify.messages.send_text(to="919876543210", text="Hi")
except SplashifyRateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after_sec}s")
except SplashifyAuthError:
    print("Bad API key")
except SplashifyValidationError as e:
    print(f"Bad input: {e.field} — {e.reason}")
except SplashifyError as e:
    print(f"API error {e.code}: {e.message}")

from splashifypro.types import SendResult, Contact, Broadcast