Django to najpopularniejszy framework Python do aplikacji webowych. Dowiedz się, jak zintegrować SMS API — z użyciem requests, Celery i sygnałów.
Django to najpopularniejszy framework Python do aplikacji webowych. Jeśli budujesz aplikację w Django i chcesz wysyłać SMS-y, ten artykuł pokaże Ci, jak zrobić to dobrze — z użyciem requests, Celery i sygnałów.
Wymagania
- Django 4.2+ (Python 3.10+)
- Konto SMSIK z kluczem API
- Opcjonalnie: Celery + Redis (dla kolejki)
Krok 1: Instalacja zależności
pip install requests
# opcjonalnie:
pip install celery redis
Krok 2: Konfiguracja klucza API
W settings.py:
import os
SMSIK_API_KEY = os.environ.get('SMSIK_API_KEY', '')
SMSIK_API_URL = os.environ.get('SMSIK_API_URL', 'https://smsik.pl/api/v1')
I w .env:
SMSIK_API_KEY=sk_live_TWOJ_KLUCZ
Krok 3: Serwis SMS
Utwórz moja_aplikacja/services/smsik.py:
import logging
import requests
from django.conf import settings
logger = logging.getLogger(__name__)
class SmsikError(Exception):
"""Błąd komunikacji z SMSIK API."""
def send_sms(to_number: str, message: str, idempotency_key: str | None = None) -> dict:
"""Wysyła SMS przez SMSIK API.
Zwraca dict z odpowiedzią API (id, status, ...).
Rzuca SmsikError przy błędzie.
"""
payload = {
'to_number': to_number,
'message': message,
}
if idempotency_key:
payload['idempotency_key'] = idempotency_key
try:
response = requests.post(
f'{settings.SMSIK_API_URL}/sms/send',
headers={
'X-API-Key': settings.SMSIK_API_KEY,
'Content-Type': 'application/json',
},
json=payload,
timeout=30,
)
except requests.RequestException as e:
logger.error('SMSIK request failed: %s', e)
raise SmsikError(f'Błąd połączenia z SMSIK: {e}') from e
if response.status_code == 200:
return response.json()
# Mapowanie kodów błędów
error_messages = {
400: 'Błąd walidacji (numer, treść)',
401: 'Nieprawidłowy klucz API',
402: 'Za mało środków na koncie',
403: 'Konto nieaktywne',
429: 'Przekroczony limit zapytań',
}
error_msg = error_messages.get(response.status_code, f'HTTP {response.status_code}')
logger.error('SMSIK API error %s: %s', response.status_code, response.text)
raise SmsikError(f'{error_msg}: {response.text[:200]}')
Krok 4: Task w Celery (opcjonalnie)
Jeśli używasz Celery, wysyłkę SMS-a przenieś do taska:
# moja_aplikacja/tasks.py
from celery import shared_task
from .services.smsik import send_sms, SmsikError
@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def send_sms_task(self, to_number: str, message: str, idempotency_key: str | None = None):
try:
return send_sms(to_number, message, idempotency_key)
except SmsikError as e:
# Retry tylko przy błędach sieci i 5xx (nie przy 400, 401, 402)
if 'HTTP 5' in str(e) or 'połączenia' in str(e):
raise self.retry(exc=e)
raise
Krok 5: Użycie w widoku
from django.http import JsonResponse
from django.views.decorators.http import require_POST
from .models import Order
from .tasks import send_sms_task
@require_POST
def create_order(request):
order = Order.objects.create(...)
# Wyślij SMS w tle (jeśli używasz Celery)
send_sms_task.delay(
order.phone,
f'Zamowienie #{order.id} przyjete. Dziekujemy!',
f'order-{order.id}-confirmation',
)
# Lub bezpośrednio (bez Celery):
# send_sms(order.phone, f'...', f'...')
return JsonResponse({'order_id': order.id})
Krok 6: Sygnały Django (dla zaawansowanych)
Jeśli chcesz oddzielić logikę biznesową od wysyłki, użyj sygnałów:
# moja_aplikacja/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Order
from .tasks import send_sms_task
@receiver(post_save, sender=Order)
def order_created(sender, instance, created, **kwargs):
if created and instance.phone:
send_sms_task.delay(
instance.phone,
f'Zamowienie #{instance.id} przyjete. Dziekujemy!',
f'order-{instance.id}-confirmation',
)
I w apps.py:
class MojaAplikacjaConfig(AppConfig):
name = 'moja_aplikacja'
def ready(self):
import moja_aplikacja.signals # noqa
Krok 7: Test
W tests/test_sms.py:
from unittest.mock import patch
from django.test import TestCase
from moja_aplikacja.services.smsik import send_sms
class SendSmsTest(TestCase):
@patch('moja_aplikacja.services.smsik.requests.post')
def test_send_sms_success(self, mock_post):
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {
'id': 123,
'status': 'queued',
'to_number': '+48512017029',
'price_grosze': 1,
}
result = send_sms('+48512017029', 'Test')
self.assertEqual(result['id'], 123)
self.assertEqual(result['status'], 'queued')
mock_post.assert_called_once()
Obsługa błędów — co robić?
| HTTP | Co robić |
|---|---|
400 | Nie retry — błąd walidacji (zły numer, treść) |
401 | Nie retry — zły klucz API |
402 | Nie retry — doładuj konto |
429 | Retry po Retry-After |
5xx | Retry z backoffem (3 próby, 60s, 2min, 10min) |
Podsumowanie
- ✅ Serwis
send_sms()— reużywalny klient HTTP - ✅ Celery task — nie blokuje requestu
- ✅ Sygnały Django — czysta architektura
- ✅ Mapowanie błędów — wiesz, co retry a co nie
- ✅ Testy —
unittest.mock
Zarejestruj się bezpłatnie — 5 groszy na start.