cloudme-notify-django

Django integration for the CloudMe Notify API, built on the cloudme-notify SDK: settings-driven configuration, a ready notify client, token caching in Django's cache, a webhook view with signature verification, and a notify_doctor command.

Requirements

• Python 3.9+, Django 4.2+
• A Notify API client: client_id, api_key and its RSA private key (dashboard -> API clients)

Installation

pip install cloudme-notify-django

# settings.py
INSTALLED_APPS = [
    # ...
    "cloudme_notify_django",
]

NOTIFY = {
    "ENVIRONMENT": "sandbox",                          # "production" for real sends
    "CLIENT_ID": os.environ["NOTIFY_CLIENT_ID"],
    "API_KEY": os.environ["NOTIFY_API_KEY"],
    "PRIVATE_KEY_PATH": BASE_DIR / "secrets" / "notify.pem",
    "WEBHOOK_SECRET": os.environ.get("NOTIFY_WEBHOOK_SECRET"),
}

Every key may be left out of NOTIFY when the matching environment variable is set (NOTIFY_ENVIRONMENT, NOTIFY_CLIENT_ID, NOTIFY_API_KEY, NOTIFY_PRIVATE_KEY_PATH, NOTIFY_WEBHOOK_SECRET, ...).

Check the setup — nothing is sent and no secret is printed:

python manage.py notify_doctor

Sending

from cloudme_notify_django import notify

notify.sms.send(to="+998901234567", message="Buyurtmangiz tayyor")
notify.telegram.send(to=user.phone, template_id=42, variables={"name": user.first_name})
notify.email.send(to=user.email, subject="Hisob-faktura", message="...")

notify is the full cloudme-notify client (otp, templates, debts, scheduled_messages, balance(), message()...), created on first use and shared by the process. The access token lives in your Django cache, so web workers and Celery tasks reuse it instead of authenticating on every request.

# OTP
otp = notify.otp.send(to=request.POST["phone"])
request.session["otp_id"] = otp.otp_id
result = notify.otp.verify(request.session["otp_id"], request.POST["code"])

# Scheduled messages
from cloudme_notify import Recurrence

notify.scheduled_messages.create(
    channel="sms",
    starts_at=appointment.starts_at - timedelta(days=1),   # aware datetimes keep their offset
    recipient=customer.phone,
    message=f"Eslatma: ertaga soat {appointment.starts_at:%H:%M} da qabul",
)
notify.scheduled_messages.create(channel="telegram", starts_at="2026-10-05T09:00:00+05:00",
                                 contact_group_id=12, message="Salom, {{name}}!", recurrence=Recurrence.weekly([1, 4]))

# Debt collection
debt = notify.debts.create(name=invoice.customer.name, phone=invoice.customer.phone,
                           amount=invoice.total, due_date=invoice.due_date, external_id=f"INV-{invoice.pk}")

Errors are the SDK's exceptions (cloudme_notify.ValidationError, InsufficientBalanceError, ForbiddenError, ...), each with error_code and a localized hint.

Sending from a request blocks for the API round-trip; for bulk or non-urgent sends call it from a Celery task (or use scheduled messages).

Receiving webhooks

# urls.py
from cloudme_notify_django.webhooks import webhook_view

urlpatterns = [
    path("webhooks/notify/", webhook_view),
]

# yourapp/signals.py (imported from AppConfig.ready)
from django.dispatch import receiver
from cloudme_notify_django.signals import webhook_received

@receiver(webhook_received)
def on_notify_event(sender, event, **kwargs):
    # a retried delivery carries the same event.id - dedupe on it
    if event.event == "message.failed":
        Order.objects.filter(notify_message_id=event.message_id).update(sms_failed=True)

Register the URL as the API client's webhook in the Notify dashboard and put its signing secret in NOTIFY["WEBHOOK_SECRET"]. The view is CSRF-exempt, accepts only POST, verifies the signature and timestamp (401 otherwise) and answers 200 — keep receivers fast, Notify waits 5 seconds.

Prefer your own view? Decorate it:

from cloudme_notify_django.webhooks import notify_webhook

@notify_webhook
def notify_events(request):
    event = request.notify_event          # cloudme_notify.WebhookEvent
    ...
    return HttpResponse(status=204)

Testing

Swap the client for one on an httpx.MockTransport, so tests never call the API:

import httpx
from cloudme_notify import NotifyClient
from cloudme_notify_django import set_client

def handler(request):
    if request.url.path.endswith("/oauth/token"):
        return httpx.Response(200, json={"access_token": "t", "expires_in": 900, "refresh_token": "r"})
    return httpx.Response(200, json={"success": True, "message_id": "m-1", "status": "queued", "price": "0", "currency": "UZS"})

set_client(NotifyClient(client_id="c", api_key="k", private_key=TEST_PEM,
                        base_url=NotifyClient.SANDBOX_URL, transport=httpx.MockTransport(handler)))
# ... set_client(None) afterwards

Changing settings.NOTIFY (e.g. with override_settings) resets the client automatically.

License

MIT