signalbird - Python

pip install signalbird

# .env
SIGNALBIRD_DOMAIN_KEY=sb_secret_live_xxxxxxxxxxxxxxxxxxxxxxxx
SIGNALBIRD_SOURCE=api-01
SIGNALBIRD_DOMAIN_KEY=sb_xxxxxxxxxxxxxxxxxxxxxxxx

from signalbird import signalbird

# SIGNALBIRD_DOMAIN_KEY ortam değişkeninden okunur, tekildir
signalbird().critical("critical", "ödeme servisi yanıt vermiyor", {
    "service": "iyzico",
    "attempt": 3,
})

signalbird().info("info", "ahmet@x.com yeni hesap oluşturdu")

from signalbird import SignalbirdClient

radio = SignalbirdClient(
    domain_key=settings.SIGNALBIRD_DOMAIN_KEY,
    source="api-01",
    timeout=5.0,
    throw_on_error=False,   # varsayılan: sessiz hata
)

radio.log("deploy", "v2.4.0 yayında", level="info")

result = radio.batch([
    {"channel": "deploy", "message": "v2.4.0 yayında", "level": "info"},
    {"channel": "critical", "message": "kuyruk doldu"},
])

# settings.py
SIGNALBIRD = SignalbirdClient(domain_key=env("SIGNALBIRD_DOMAIN_KEY"), source=env("APP_ENV"))

# middleware.py
class SignalbirdMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        return self.get_response(request)

    def process_exception(self, request, exception):
        # Sessiz hata: telsiz erişilemezse istek zaten 500 dönecek,
        # bir de SDK yüzünden ikinci bir hata üretmesin.
        settings.SIGNALBIRD.error("api", str(exception), {
            "path": request.path,
            "method": request.method,
        })

@app.task(bind=True)
def process_payment(self, order_id):
    try:
        charge(order_id)
        signalbird().info("odeme", f"Sipariş #{order_id} başarılı")
    except Exception as error:
        signalbird().critical("critical", f"Sipariş #{order_id}: {error}")
        raise

from signalbird import SignalbirdMessaging

sb = SignalbirdMessaging(domain_key=os.environ["SIGNALBIRD_DOMAIN_KEY"])

result = sb.send_email({
    "to": "ahmet@example.com",
    "class": "transactional",       # ZORUNLU, varsayılanı yok
    "subject": "Siparişiniz yola çıktı",
    "body": "<p>Kargo takip no: 1234</p>",
})

if not result.ok:
    logger.warning("signalbird: %s - %s", result.code, result.message)

sb.send_sms({"to": "+905551112233", "class": "transactional", "body": "Kodunuz: 123456"})
sb.preview_sms("Merhaba, siparişiniz hazır.")     # parça hesabı, kota harcamaz
sb.send_push({"to": "contact:42", "class": "transactional", "subject": "Sipariş", "body": "Yola çıktı"})

sb.list_contacts({"q": "ahmet", "page": 1})
sb.create_contact({"email": "ahmet@example.com", "first_name": "Ahmet"})
sb.update_contact(contact_id, {"phone": "+905551112233"})
sb.delete_contact(contact_id)

# 1000'lik parçalar hâlinde SIRAYLA yükler
sb.bulk_contacts({
    "contacts": rows,
    "list_id": 42,
    "consent_source": "web_form",
    "consent_text": "Bültene abone olmayı kabul ediyorum",
})

sb.create_campaign({
    "name": "Ağustos bülteni",
    "channel": "email",
    "list_id": 42,
    "subject": "Bu ay neler oldu?",
    "body": "<h1>Merhaba</h1>",
    "scheduled_at": "2026-09-01T09:00:00Z",
})

sb.get_campaign(batch_id)
sb.cancel_campaign(batch_id)

# Sayfa sayfa gezer, tüm listeyi belleğe almaz
for message in sb.iterate_campaign_messages(batch_id):
    reconcile(message["external_ref"], message["status"])

from signalbird import verify_webhook

# Django - request.body HAM gövdedir
if not verify_webhook(request.body, request.headers.get("X-Signalbird-Signature"), SECRET):
    return HttpResponseForbidden()

# FastAPI
raw = await request.body()
if not verify_webhook(raw, request.headers.get("x-signalbird-signature"), SECRET):
    raise HTTPException(403)

from signalbird import SignalbirdManagement

admin = SignalbirdManagement(domain_key=os.environ["SIGNALBIRD_DOMAIN_KEY"])

# Deploy betiğinde: yeni ortam için proje ve kanal aç
created = admin.create_radio_project({"name": "ödeme-servisi"})
secret = created.data["secret"]     # YALNIZ burada döner

admin.create_radio_channel(created.data["project"]["id"], {
    "key": "odeme",
    "name": "Ödeme",
    "level": "critical",
    "notify_push": True,
    "quiet_from": 0,
    "quiet_to": 7,
})

# Sohbet gelen kutusu botu
inbox = admin.list_conversations({"status": "open"})

for conversation in inbox.data["data"]:
    admin.reply(conversation["id"], {"body": "Merhaba! Birazdan dönüyoruz."})
    admin.set_conversation_status(conversation["id"], "pending")

result = sb.send_email({...})

result.ok         # bool
result.status     # int
result.data       # çözülmüş JSON gövde
result.code       # hata kodu (başarıda None)
result.message

result["data"]    # aynı şey

from signalbird import signalbird

hash = signalbird().identity_hash(str(user.id))

identity_hash = hex( HMAC-SHA256( hex(SHA-256(sb_secret_live_…)), external_id ) )

pip install -U signalbird