Asosiy kontentga o‘tish

Maqolalar 15 daqiqa 295 ko‘rish

FCM push: ilovasi yopiq foydalanuvchiga xabar yetkazish

WebSocket faqat ilovasi ochiq turganlarga yetadi. Qolganlarga — Firebase Cloud Messaging: qurilma tokenlarini to'g'ri saqlash, mehmon-foydalanuvchilarga ham push yuborish, o'lik tokenlarni tozalash — Metro backend'idagi to'liq amaliyot.

mobile django

Mundarija

Avvalgi qismda ogohlantirishni ilovasi ochiq turgan foydalanuvchiga WebSocket bilan yetkazdik. Lekin foydalanuvchilarning mutlaq ko'pchiligida ilova yopiq turadi. Ularga yo'l bitta — operatsion tizimning push-kanali, ya'ni FCM (Firebase Cloud Messaging): Android'da to'g'ridan-to'g'ri, iOS'da APNs ustidan ishlaydi — backend uchun ikkalasi bitta API.

Sxema shunday:

Sxema
Alert saqlandi (post_save signal)
        │
        ├──► Channels group_send ──► ochiq ilovalar (2-qism)
        │
        └──► FCM API ──► Google/Apple serverlari ──► yopiq ilovalar (shu qism)

1-qadam: Firebase kaliti va SDK#

Firebase Console'da loyiha ochiladi (mobil jamoa bilan bitta loyiha!), so'ng Project settings → Service accounts → Generate new private key — JSON fayl tushadi. Bu fayl — server kaliti, git'ga TUSHMAYDI.

Terminal
uv add firebase-admin
Python
# config/settings.py
import os

# kalit yo'li .env orqali beriladi: FIREBASE_CREDENTIALS=/opt/app/secrets/firebase.json
FIREBASE_CREDENTIALS = os.environ.get("FIREBASE_CREDENTIALS", "")
Python
# apps/notifications/firebase.py
import firebase_admin
from django.conf import settings
from firebase_admin import credentials


def init_firebase() -> None:
    # ikki marta chaqirilsa ham xato bermasin — idempotent
    if firebase_admin._apps:
        return
    cred = credentials.Certificate(settings.FIREBASE_CREDENTIALS)
    firebase_admin.initialize_app(cred)

Service account JSON — bu parol

Bu fayl bilan istalgan odam sizning nomingizdan push yuboradi. .gitignorega kiritish yetmaydi — u umuman repo papkasida turmasin: serverda alohida katalogda (/opt/app/secrets/), o'qish huquqi faqat servis foydalanuvchisida. Sirlarni boshqarish haqida DevOps seriyasida alohida maqola bor.

2-qadam: qurilma tokenlari modeli#

FCM har o'rnatilgan ilova nusxasiga token beradi. Bizning vazifa — shu tokenlarni yig'ish va toza saqlash. Metro'dagi muhim mahsulot qarori: push olish uchun ro'yxatdan o'tish shart emas — shuning uchun user maydoni bo'sh bo'lishi mumkin (mehmon-qurilma):

Fayl tuzilishi
apps/notifications/
├── migrations/
├── __init__.py
├── firebase.py      ← SDK init (yuqorida)
├── models.py        ← UserDevice
├── serializers.py
├── views.py         ← token ro'yxatga olish endpoint'i
├── urls.py
├── services.py      ← push yuborish logikasi
└── tests.py

Yangi app'ni yaratib, loyihaga ulaymiz:

Terminal
mkdir -p apps/notifications/migrations
touch apps/notifications/__init__.py apps/notifications/migrations/__init__.py
Python
# config/settings.py
INSTALLED_APPS = [
    # ...
    "apps.alerts",
    "apps.notifications",
]

Bu qatorni o'tkazib yubormang: 2-qismdagi signal fayli birozdan keyin apps.notifications ni import qiladi, va app ro'yxatda bo'lmasa loyiha umuman ishga tushmaydi — UserDevice ... isn't in an application in INSTALLED_APPS xatosi chiqadi.

Python
# apps/notifications/models.py
from django.conf import settings
from django.db import models


class UserDevice(models.Model):
    class Platform(models.TextChoices):
        ANDROID = "android", "Android"
        IOS = "ios", "iOS"

    # null=True — mehmon-qurilma: hali login qilmagan, lekin push oladi.
    # Keyin login qilsa, xuddi shu yozuvga user bog'lanadi.
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        null=True,
        blank=True,
        on_delete=models.CASCADE,
        related_name="devices",
    )
    token = models.CharField(max_length=512, unique=True)
    platform = models.CharField(max_length=8, choices=Platform.choices)
    is_active = models.BooleanField(default=True)
    updated_at = models.DateTimeField(auto_now=True)

    def __str__(self):
        return f"{self.platform} · {self.token[:12]}…"
Terminal
uv run python manage.py makemigrations notifications
uv run python manage.py migrate

3-qadam: tokenni ro'yxatga oluvchi endpoint#

Mobil ilova ishga tushganda (va FCM token yangilanganda) shu endpoint'ni chaqiradi:

Python
# apps/notifications/serializers.py
from rest_framework import serializers

from .models import UserDevice


class DeviceSerializer(serializers.ModelSerializer):
    class Meta:
        model = UserDevice
        fields = ["token", "platform"]
        # token modelda unique=True — DRF buning uchun o'zi validator qo'shadi
        # va takroriy tokenni 400 bilan qaytaradi. Bizga esa aynan takroriy
        # so'rov kerak: view uni update_or_create bilan yangilaydi
        extra_kwargs = {"token": {"validators": []}}

extra_kwargs qatori mayda ko'rinadi, lekin busiz endpoint faqat birinchi chaqiruvda ishlaydi. Ilova tokenni har ishga tushganda qayta yuboradi — ikkinchi so'rov user device with this token already exists bilan 400 oladi. Natijada login qilgan foydalanuvchi qurilmasiga bog'lanmaydi va o'chirilgan (is_active=False) qurilma qayta yoqilmaydi.

Python
# apps/notifications/views.py
from rest_framework import status
from rest_framework.permissions import AllowAny
from rest_framework.response import Response
from rest_framework.views import APIView

from .models import UserDevice
from .serializers import DeviceSerializer


class RegisterDeviceView(APIView):
    # mehmonlar ham ro'yxatga olinadi — mahsulot qarori (yuqoriga qarang)
    permission_classes = [AllowAny]

    def post(self, request):
        serializer = DeviceSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        device, _ = UserDevice.objects.update_or_create(
            token=serializer.validated_data["token"],
            defaults={
                "platform": serializer.validated_data["platform"],
                # login bo'lgan bo'lsa qurilma egasini bog'laymiz,
                # mehmon bo'lsa None qoladi
                "user": request.user if request.user.is_authenticated else None,
                "is_active": True,
            },
        )
        return Response({"ok": True}, status=status.HTTP_201_CREATED)
Python
# apps/notifications/urls.py
from django.urls import path

from .views import RegisterDeviceView

urlpatterns = [
    path("devices/", RegisterDeviceView.as_view()),
]
Python
# config/urls.py — 1-qismdagi ro'yxatga yangi qator
urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/", include("apps.metro.urls")),
    path("api/", include("apps.notifications.urls")),
]
Terminal
curl -X POST http://127.0.0.1:8000/api/devices/ \
  -H "Content-Type: application/json" \
  -d '{"token": "sinov-token", "platform": "android"}'
# {"ok":true} — xuddi shu so'rovni qayta yuborsangiz ham shunday javob keladi

Ochiq endpoint — ochiq eshik

AllowAny turgan har endpoint'ga bot ham yozadi. Token formati baribir FCM'da tekshiriladi (yaroqsiz token hech kimga bormaydi), lekin bazani axlatga to'ldirishdan rate limiting saqlaydi — bu endpoint'ga so'rov chastotasini cheklab qo'ying.

4-qadam: push yuborish xizmati#

Yuborishning uch nozik joyi bor: (1) FCM bir chaqiruvda ko'pi bilan 500 token qabul qiladi, (2) o'lik tokenlar javobda UnregisteredError bilan qaytadi — ularni o'chirmasangiz baza yillar davomida shishadi, (3) yuborish sekundlar olishi mumkin — HTTP so'rov oqimini bloklamaslik kerak.

Python
# apps/notifications/services.py
from firebase_admin import messaging

from .firebase import init_firebase
from .models import UserDevice

BATCH = 500  # FCM cheklovi: bitta multicast'da 500 tagacha token


def send_push(title: str, body: str, data: dict[str, str] | None = None) -> dict:
    init_firebase()
    tokens = list(UserDevice.objects.filter(is_active=True).values_list("token", flat=True))
    stats = {"sent": 0, "dead": 0}

    for start in range(0, len(tokens), BATCH):
        chunk = tokens[start : start + BATCH]
        message = messaging.MulticastMessage(
            tokens=chunk,
            notification=messaging.Notification(title=title, body=body),
            # data — ilovaga "ko'rinmas" yuk: masalan qaysi ekranni ochish
            data=data or {},
        )
        response = messaging.send_each_for_multicast(message)
        stats["sent"] += response.success_count

        # javob tartibi token tartibiga mos — o'liklarni aniqlaymiz
        dead = [
            chunk[i]
            for i, result in enumerate(response.responses)
            if result.exception is not None
            and isinstance(result.exception, messaging.UnregisteredError)
        ]
        if dead:
            # qattiq o'chirmaymiz — flag: qurilma qaytib kelsa qayta yoqiladi
            UserDevice.objects.filter(token__in=dead).update(is_active=False)
            stats["dead"] += len(dead)

    return stats

tokens → fids: FCM o'tish davrida

firebase-admin 7.5.0 (2026-yil iyul) dan boshlab MulticastMessage(tokens=...) DeprecationWarning beradi: FCM registration token'lardan Firebase Installation ID (FID) ga o'tmoqda. Firebase hujjatiga ko'ra hozircha ikkala usul ham to'liq qo'llab-quvvatlanadi, shuning uchun yuqoridagi kod ishlaydi. Mobil jamoa ilovani FID yuboradigan qilganda, backend'da tokens=chunk o'rniga fids=chunk yoziladi. Testlarni -W error bilan yurgizsangiz, bu ogohlantirish xatoga aylanadi.

Endi 2-qismdagi signalga ikkinchi kanalni ulaymiz — "bir manba, ikki kanal" printsipi kodda tugallanadi. Avval fayl boshidagi import'larga ikki qator qo'shiladi (threading va send_push):

Python
# apps/alerts/signals.py — fayl boshi
import logging
import threading

from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from django.db.models.signals import post_save
from django.dispatch import receiver

from apps.notifications.services import send_push

from .consumers import ALERTS_GROUP
from .models import Alert

Keyin fayl oxiriga, broadcast_alert dan keyin, ikkinchi receiver:

Python
# apps/alerts/signals.py — fayl oxiri
@receiver(post_save, sender=Alert)
def push_alert(sender, instance: Alert, created: bool, **kwargs):
    if not created or not instance.is_active:
        return
    # FCM chaqiruvi soniyalar olishi mumkin — admin so'rovini kutdirmaslik
    # uchun alohida oqimda. Yuk oshsa bu joy Celery vazifasiga ko'chadi:
    # https://solijonov.uz/writing/celery-fon-vazifalar/
    threading.Thread(
        target=send_push,
        args=(instance.title, instance.body),
        kwargs={"data": {"alert_id": str(instance.id)}},
        daemon=True,
    ).start()

Qachon thread yetadi, qachon Celery kerak

Ogohlantirish kuniga bir necha marta kiritiladi — bunga daemon-thread ortiq yetadi va infratuzilma talab qilmaydi. Push har daqiqada ketadigan, qayta urinish (retry) va kuzatuv kerak bo'lgan tizimda esa navbat (Celery) to'g'ri yechim. Oddiy boshlang, o'lchab ko'chiring.

Ikki receiver bir-biriga bog'liq emasligi 2-qismda qo'yilgan try/except tufayli: Redis yotgan kuni ham broadcast_alert xatoni logga yozib qaytadi va Django navbatdagi push_alert ni chaqiradi — yopiq ilovalarga push baribir ketadi.

5-qadam: test — Firebase'siz#

Tashqi servisga test paytida chiqilmaydi — send_each_for_multicastni mock qilamiz:

Python
# apps/notifications/tests.py
from unittest.mock import MagicMock, patch

import pytest
from firebase_admin import messaging

from apps.notifications.models import UserDevice
from apps.notifications.services import send_push


@pytest.fixture
def devices(db):
    UserDevice.objects.create(token="tok-tirik", platform="android")
    UserDevice.objects.create(token="tok-olik", platform="ios")


def _fake_response():
    ok = MagicMock(exception=None)
    dead = MagicMock(exception=messaging.UnregisteredError("gone"))
    return MagicMock(success_count=1, responses=[ok, dead])


@patch("apps.notifications.services.init_firebase")
@patch("apps.notifications.services.messaging.send_each_for_multicast")
def test_dead_tokens_get_deactivated(send_mock, init_mock, devices):
    send_mock.return_value = _fake_response()

    stats = send_push("Sinov", "Matn")

    assert stats == {"sent": 1, "dead": 1}
    assert UserDevice.objects.get(token="tok-olik").is_active is False
    assert UserDevice.objects.get(token="tok-tirik").is_active is True
Terminal
uv run pytest apps/notifications/ -q

Xulosa#

  • Push — WebSocket'ning davomi emas, ikkinchi mustaqil kanal: manba bitta (Alert + signal), yetkazish yo'llari ikkita;

  • Token — foydalanuvchiga emas, qurilmaga tegishli; user maydonini ixtiyoriy qilish mehmonlarga ham push berish imkonini beradi (Metro'da bu qamrovni keskin oshirgan);

  • O'lik tokenlarni har yuborishda tozalang — aks holda baza va FCM kvota bekor ishlaydi;

  • Tashqi servis chaqiruvini so'rov oqimidan chiqaring: avval thread, yuk o'sganda Celery.

Keyingi qismda autentifikatsiyaga o'tamiz: Google Sign-In — iOS va Android ilovalardan kelgan Google tokenini backend'da qanday tekshiramiz va nega bu "token'ni qabul qilib qo'yish"dan ancha nozik ish.

15 daqiqa 295 marta o‘qildi