← All PostsJul 21, 2026

Push notifications from Django

Your Django app processes something. Your Celery task finishes. You find out by checking the admin or tailing logs.

Here’s how to make Django tell you directly.

The simplest version: requests

import requests
import os
 
def notify(title, body=None, level=None, **meta):
    key = os.environ.get("TRIGGER_FYI_SECRET_KEY")
    if not key:
        return
 
    payload = {"title": title}
    if body:
        payload["body"] = body
    if level:
        payload["level"] = level
    if meta:
        payload["meta"] = meta
 
    try:
        requests.post(
            f"https://trigger.fyi/{key}",
            json=payload,
            timeout=3
        )
    except Exception:
        pass  # fire-and-forget

With the SDK

pip install trigger-fyi

from trigger_fyi import fyi
 
# Normal push
fyi("New signup", body="[email protected]", plan="pro")
 
# Feed only
fyi.log("Cache invalidated", body="12,400 keys")
 
# High urgency
fyi.critical("Payment failed", body="[email protected]")

In a Django view

After the response — don’t block the request:

from django.http import JsonResponse
from trigger_fyi import fyi
import threading
 
def signup(request):
    if request.method != "POST":
        return JsonResponse({"error": "Method not allowed"}, status=405)
 
    user = User.objects.create(**parse_signup(request))
 
    # Fire in background — response doesn't wait
    threading.Thread(
        target=fyi,
        args=("New signup",),
        kwargs={"body": user.email, "plan": user.plan},
        daemon=True
    ).start()
 
    return JsonResponse({"id": user.id, "email": user.email}, status=201)

Or use Django’s transaction.on_commit to fire after the database transaction commits:

from django.db import transaction
from trigger_fyi import fyi
 
def signup(request):
    with transaction.atomic():
        user = User.objects.create(**parse_signup(request))
        transaction.on_commit(
            lambda: threading.Thread(
                target=fyi,
                args=("New signup",),
                kwargs={"body": user.email},
                daemon=True
            ).start()
        )
 
    return JsonResponse({"id": user.id})

on_commit ensures the notification fires only after the user is committed to the database — useful if you’re checking the database on the other end.

Using Django signals

# apps/accounts/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth import get_user_model
from trigger_fyi import fyi
import threading
 
User = get_user_model()
 
@receiver(post_save, sender=User)
def on_user_created(sender, instance, created, **kwargs):
    if not created:
        return
 
    threading.Thread(
        target=fyi,
        args=("New signup",),
        kwargs={"body": instance.email},
        daemon=True
    ).start()

# apps/accounts/apps.py
from django.apps import AppConfig
 
class AccountsConfig(AppConfig):
    name = "apps.accounts"
 
    def ready(self):
        import apps.accounts.signals  # noqa

In a Celery task

from celery import shared_task
from trigger_fyi import fyi
import time
 
@shared_task(bind=True, max_retries=3)
def generate_report(self, user_id):
    start = time.time()
    try:
        result = ReportGenerator.run(user_id)
        elapsed = int(time.time() - start)
        fyi("Report complete", body=f"{result['rows']} rows · {elapsed}s", user_id=str(user_id))
        return result
    except Exception as exc:
        fyi.critical("Report failed", body=str(exc), user_id=str(user_id))
        raise self.retry(exc=exc, countdown=60)

In a management command

# management/commands/sync_data.py
from django.core.management.base import BaseCommand
from trigger_fyi import fyi
import time
 
class Command(BaseCommand):
    help = "Sync data from external API"
 
    def handle(self, *args, **options):
        start = time.time()
        try:
            count = sync_external_data()
            elapsed = int(time.time() - start)
            fyi("Sync complete", body=f"{count} records · {elapsed}s")
            self.stdout.write(f"Synced {count} records")
        except Exception as e:
            fyi.critical("Sync failed", body=str(e))
            raise

Environment variable

In settings.py or .env:

TRIGGER_FYI_SECRET_KEY=your_key_here

The SDK reads os.environ["TRIGGER_FYI_SECRET_KEY"] at call time — no initialization needed.

Setup

npx trigger.fyi

Generates a key, subscribes your device. Add the key to your environment and start notifying.

#The simplest version: requests

#With the SDK

#In a Django view

#Using Django signals

#In a Celery task

#In a management command

#Environment variable

#Setup