ساخت API JSON در جنگو برای نمایش آمار زنده در داشبورد؛ چرا یک endpoint ساده کافی نیست؟

اگر تصور می‌کنید یک JsonResponse ساده که چند عدد را از دیتابیس می‌خواند، برای نمایش آمار زنده در داشبورد کافی است، احتمالاً تا امروز هر بار باز کردن داشبورد شما چند ثانیه طول کشیده و در ساعات پرترافیک، همین endpoint به گلوگاهی برای کل سایت تبدیل شده است؛ چون آمار زنده، یک چالش تخصصی است، نه یک ویوی ساده.

چرا نمایش آمار زنده یک چالش مهندسی است؟

در یکی از پروژه‌های فروشگاهی که چند سال پیش روی آن کار می‌کردم، تیم محصول یک داشبورد آمار زنده خواست که هر ۵ ثانیه به‌روز شود. تیم توسعه یک endpoint ساده ساخت که روی هر درخواست، پنج کوئری سنگین روی جدول Interaction می‌زد. نتیجه: با ۵ کاربر همزمان داشبورد، دیتابیس اصلی به گلوگاه تبدیل شد و زمان پاسخ صفحات فروشگاه ۳۰٪ کندتر شد.

این تجربه نشان می‌دهد که نمایش آمار زنده، یک چالش مهندسی است، نه یک نمایش ساده. سه دلیل اصلی:

دلیل اول، فرکانس بالای درخواست. اگر هر ۵ ثانیه یک endpoint صدا زده شود، با ۱۰ کاربر همزمان، شما ۱۲۰ درخواست در دقیقه دارید. با ۱۰۰ کاربر، ۱۲۰۰ درخواست در دقیقه. این نرخ، بسیار بیشتر از بازدیدهای عادی است.

دلیل دوم، کوئری‌های سنگین. محاسبه‌ی آمار، نیازمند تجمیع روی جدول‌های بزرگ است. اگر این کوئری‌ها در هر درخواست تکرار شوند، دیتابیس به سرعت اشباع می‌شود.

دلیل سوم، حساسیت به تأخیر. آمار زنده باید در کمتر از یک ثانیه پاسخ دهد. اگر پاسخ کند باشد، داشبورد غیرقابل استفاده می‌شود.

آمار زنده، یک نمایش نیست؛ یک تعهد به تأخیر کم و کارایی پایدار است. اگر این تعهد را نمی‌توانید بدهید، بهتر است داشبورد را دوره‌ای به‌روز کنید.

این اهمیت، در ساختار پروژه‌های آماری مثل ساخت پنل ادمین سفارشی با کارت‌های quick view به‌طور مستقیم دیده می‌شود. اگر لایه‌ی نمایش آمار به‌درستی طراحی نشود، تمام داده‌های جمع‌آوری‌شده بی‌استفاده می‌مانند. برای درک عمیق‌تر این موضوع، پیشنهاد می‌کنم ابتدا طراحی مدل Visitor و Visit در جنگو را مطالعه کنید، چون ساختار داده، مستقیماً روی طراحی endpoint اثر می‌گذارد.

endpoint ساده‌ای که در پروژه‌های واقعی می‌شکند

endpoint ساده‌ی آمار زنده، در اکثر پروژه‌ها به این شکل است:

# analytics/api.py

from django.http import JsonResponse
from django.views.decorators.http import require_GET
from django.utils import timezone
from datetime import timedelta

from analytics.models import Visit, PageView


@require_GET
def stats_summary(request):
    now = timezone.now()
    threshold = now - timedelta(seconds=300)

    online = Visit.objects.filter(
        is_active=True,
        last_activity__gte=threshold,
    ).count()

    start = now.replace(hour=0, minute=0, second=0)
    today_visitors = Visit.objects.filter(
        entry_time__gte=start,
    ).values("visitor").distinct().count()

    today_pageviews = PageView.objects.filter(
        entered_at__gte=start,
    ).count()

    return JsonResponse({
        "online": online,
        "today_visitors": today_visitors,
        "today_pageviews": today_pageviews,
    })

این کد، شش مشکل جدی دارد.

مشکل اول، کوئری‌های سنگین در هر درخواست. سه کوئری، هر کدام روی جدول‌های بزرگ، در هر بار اجرا می‌شوند.

مشکل دوم، عدم کش. اگر ۱۰ کاربر همزمان داشبورد را باز کنند، ۳۰ کوئری یکسان اجرا می‌شود.

مشکل سوم، عدم rate limiting. یک کاربر می‌تواند هزاران درخواست در دقیقه بفرستد.

مشکل چهارم، عدم کنترل دسترسی. هر کاربری می‌تواند آمار را ببیند.

مشکل پنجم، محاسبه‌ی start نادرست. now.replace(hour=0, ...) در timezone-aware datetime می‌تواند خطا ایجاد کند.

مشکل ششم، عدم مدیریت خطا. اگر دیتابیس کند باشد، endpoint زمان‌بر می‌شود و داشبورد کاربر را معطل می‌کند.

در یکی از پروژه‌ها، بعد از فعال شدن این endpoint در داشبورد، دیتابیس اصلی به مدت دو هفته به‌طور پیوسته در حال ۷۰٪ CPU بود. علت: هیچ کشی وجود نداشت و هر کاربر هر ۵ ثانیه، سه کوئری سنگین می‌زد.

برای رفع این مشکلات، به یک معماری چندلایه نیاز دارید. اگر با الگوی بهینه‌سازی جنگو برای ترافیک بالا کار کرده باشید، این رویکرد برایتان آشناست.

طراحی payload؛ سبک، پایدار، قابل کش

payload پاسخ endpoint آمار زنده، سه ویژگی باید داشته باشد.

ویژگی اول، سبک بودن. payload باید کوچک باشد تا تأخیر شبکه کم شود. از فیلدهای اضافی و nested objects عمیق خودداری کنید.

ویژگی دوم، پایدار بودن. ساختار payload باید در طول زمان پایدار بماند. اگر فیلدی اضافه یا حذف می‌شود، باید نسخه‌بندی داشته باشید.

ویژگی سوم، قابل کش بودن. payload باید شامل یک فیلد timestamp باشد که به کلاینت اجازه دهد بداند داده چقدر تازه است.

{
    "generated_at": "2026-09-24T10:30:00Z",
    "cached": true,
    "cache_ttl": 30,
    "stats": {
        "online": 12,
        "today_visitors": 1245,
        "today_pageviews": 8760,
        "today_minutes": 452.5
    }
}

این ساختار، سه مزیت دارد.

مزیت اول، self-contained. کلاینت می‌داند داده چه زمانی تولید شده و چه مدت معتبر است.

مزیت دوم، نسخه‌بندی. اگر ساختار تغییر کند، می‌توانید یک فیلد version اضافه کنید.

مزیت سوم، کش‌پذیری. هدرهای HTTP cache می‌توانند بر اساس cache_ttl تنظیم شوند.

from django.http import JsonResponse


def stats_response(data, cache_ttl=30):
    response = JsonResponse({
        "generated_at": timezone.now().isoformat(),
        "cached": False,
        "cache_ttl": cache_ttl,
        "stats": data,
    })
    response["Cache-Control"] = f"private, max-age={cache_ttl}"
    return response

نکته‌ی مهم: هدر Cache-Control با private باید تنظیم شود تا CDN و پروکسی‌های میانی داده‌ی کاربر را کش نکنند. اگر با الگوی ساخت API Endpoint برای دریافت Beacon کار کرده باشید، می‌دانید که این نوع تنظیمات، بخشی از انضباط API است.

کش چندلایه؛ کلید کارایی آمار زنده

کش، ستون فقرات هر endpoint آمار زنده است. سه لایه کش را در نظر بگیرید.

لایه‌ی اول، کش محلی (in-memory). برای داده‌های خیلی پرفرکانس مثل «کاربران آنلاین»، می‌توانید از cache محلی پروسه استفاده کنید. TTL کوتاه (۵ تا ۱۰ ثانیه) کافی است.

لایه‌ی دوم، کش توزیع‌شده (Redis). برای داده‌هایی که بین چند worker مشترک هستند. TTL متوسط (۳۰ تا ۶۰ ثانیه).

لایه‌ی سوم، کش لایه‌ی پایگاه داده (materialized view). برای داده‌های تجمیعی روزانه یا هفتگی. TTL طولانی (۱ تا ۲۴ ساعت).

from django.core.cache import cache


def get_cached_stat(key, compute_fn, ttl=30):
    value = cache.get(key)
    if value is not None:
        return value, True

    value = compute_fn()
    cache.set(key, value, timeout=ttl)
    return value, False

و در endpoint:

@require_GET
def stats_summary(request):
    now = timezone.now()

    # لایه ۱: آنلاین (TTL کوتاه)
    online, cached1 = get_cached_stat(
        "analytics:online",
        lambda: count_online(now),
        ttl=15,
    )

    # لایه ۲: امروز (TTL متوسط)
    today_visitors, cached2 = get_cached_stat(
        "analytics:today_visitors",
        lambda: count_today_visitors(now),
        ttl=60,
    )

    today_pageviews, cached3 = get_cached_stat(
        "analytics:today_pageviews",
        lambda: count_today_pageviews(now),
        ttl=60,
    )

    return stats_response(
        {
            "online": online,
            "today_visitors": today_visitors,
            "today_pageviews": today_pageviews,
        },
        cache_ttl=15,
    )

نکته‌ی مهم: TTL هر لایه، باید بر اساس حساسیت داده و نرخ تغییر آن تنظیم شود. «کاربران آنلاین» سریع تغییر می‌کند، پس TTL کوتاه. «بازدیدکنندگان امروز» کندتر تغییر می‌کند، پس TTL متوسط.

در آمار زنده، کش یک ضرورت است، نه یک انتخاب. بدون کش، هر endpoint به یک گلوگاه تبدیل می‌شود.

جدول‌های تجمیعی؛ جایی که مقیاس شروع می‌شود

در مقیاس متوسط و بالا، حتی با کش، محاسبه‌ی تجمیع‌ها از داده‌ی خام می‌تواند کند باشد. راه‌حل، استفاده از جدول‌های تجمیعی است.

یک جدول DailyStats طراحی کنید که هر شب پر می‌شود:

class DailyStats(models.Model):
    date = models.DateField(unique=True, db_index=True)

    unique_visitors = models.PositiveIntegerField(default=0)
    total_visits = models.PositiveIntegerField(default=0)
    total_pageviews = models.PositiveIntegerField(default=0)
    total_seconds = models.PositiveIntegerField(default=0)

    mobile_count = models.PositiveIntegerField(default=0)
    desktop_count = models.PositiveIntegerField(default=0)
    tablet_count = models.PositiveIntegerField(default=0)

    bot_count = models.PositiveIntegerField(default=0)

    class Meta:
        db_table = "analytics_daily_stats"
        ordering = ["-date"]

و یک task شبانه که این جدول را پر می‌کند:

from django.db.models import Count, Sum
from analytics.models import Visit, PageView, DailyStats


def populate_daily_stats(date):
    visits = Visit.objects.filter(
        entry_time__date=date,
    ).exclude(visitor__device_type="bot")

    pageviews = PageView.objects.filter(
        entered_at__date=date,
    ).exclude(visitor__device_type="bot")

    stats = DailyStats.objects.update_or_create(
        date=date,
        defaults={
            "unique_visitors": visits.values("visitor").distinct().count(),
            "total_visits": visits.count(),
            "total_pageviews": pageviews.count(),
            "total_seconds": visits.aggregate(s=Sum("duration_seconds"))["s"] or 0,
            "mobile_count": visits.filter(visitor__device_type="mobile").count(),
            "desktop_count": visits.filter(visitor__device_type="desktop").count(),
            "tablet_count": visits.filter(visitor__device_type="tablet").count(),
        },
    )
    return stats[0]

این رویکرد، سه مزیت کلیدی دارد.

مزیت اول، کوئری‌های سریع. endpoint آمار زنده به‌جای چند کوئری سنگین روی داده‌ی خام، یک کوئری ساده روی جدول تجمیعی می‌زند.

مزیت دوم، پایداری کارایی. با رشد داده، جدول تجمیعی همچنان کوچک و سریع باقی می‌ماند.

مزیت سوم، امکان تحلیل تاریخی. می‌توانید روندهای بلندمدت را از جدول تجمیعی استخراج کنید.

نکته‌ی مهم: در ساعات پرترافیک، می‌توانید یک جدول HourlyStats هم داشته باشید که ساعتی پر می‌شود. اگر با الگوی کامند مدیریتی پاک‌سازی داده‌های قدیمی کار کرده باشید، می‌دانید که این نوع جدول‌های تجمیعی، بخشی از معماری استاندارد است.

کنترل دسترسی و rate limiting

endpoint آمار زنده، داده‌های حساسی را نمایش می‌دهد. سه لایه‌ی امنیتی را در نظر بگیرید.

لایه‌ی اول، احراز هویت. endpoint باید فقط برای کاربران staff یا یک نقش خاص در دسترس باشد.

from django.contrib.auth.decorators import user_passes_test


def is_analytics_viewer(user):
    return user.is_authenticated and (
        user.is_staff
        or user.groups.filter(name="Analytics Viewers").exists()
    )


@require_GET
@user_passes_test(is_analytics_viewer)
def stats_summary(request):
    ...

لایه‌ی دوم، rate limiting. endpoint باید محدودیت نرخ داشته باشد. با django-ratelimit یا یک middleware سبک.

from django.core.cache import cache


def check_rate_limit(user_id, limit=60, window=60):
    key = f"ratelimit:stats:{user_id}"
    current = cache.get(key, 0)

    if current >= limit:
        return False

    if current == 0:
        cache.set(key, 1, timeout=window)
    else:
        cache.incr(key)

    return True

لایه‌ی سوم، CSRF. اگر endpoint فقط read-only است، CSRF نیازی نیست. ولی اگر state دارد (مثلاً ذخیره‌ی تنظیمات کاربر)، باید محافظت شود.

from django.views.decorators.csrf import csrf_protect


@require_POST
@csrf_protect
def save_dashboard_settings(request):
    ...

نکته‌ی مهم: در سیستم‌های multi-tenant، endpoint باید فقط داده‌ی همان tenant را برگرداند. این کار با یک filter اضافی در کوئری انجام می‌شود.

@require_GET
@user_passes_test(is_analytics_viewer)
def stats_summary(request):
    tenant = request.user.tenant
    visits = Visit.objects.filter(tenant=tenant)
    ...

تعامل با فرانت‌اند؛ polling، SSE یا WebSocket

سه رویکرد برای به‌روزرسانی زنده‌ی داشبورد وجود دارد. هر کدام مزایا و معایب خودش را دارد.

رویکرد اول، polling. کلاینت هر N ثانیه یک درخواست به endpoint می‌فرستد.

async function refreshStats() {
    const r = await fetch("/panel/stat/api/summary/");
    const data = await r.json();
    updateUI(data);
}

refreshStats();
setInterval(refreshStats, 30000);

مزیت: ساده، قابل کش، سازگار با همه‌ی مرورگرها. عیب: درخواست‌های اضافی وقتی داده تغییر نمی‌کند.

رویکرد دوم، SSE (Server-Sent Events). سرور یک کانال یک‌طرفه باز می‌کند و داده را push می‌کند.

from django.http import StreamingHttpResponse
import json
import time


def stats_stream(request):
    def event_stream():
        while True:
            data = compute_stats()
            yield f"data: {json.dumps(data)}\n\n"
            time.sleep(15)

    return StreamingHttpResponse(
        event_stream(),
        content_type="text/event-stream",
    )

مزیت: به‌روزرسانی خودکار، بدون درخواست اضافی. عیب: پیچیدگی سرور، محدودیت در Gunicorn sync workers.

رویکرد سوم، WebSocket. کانال دوطرفه با Django Channels.

مزیت: تعامل دوطرفه، تأخیر بسیار کم. عیب: پیچیدگی زیرساخت (ASGI، Redis channel layer).

توصیه‌ی من: برای اکثر پروژه‌ها، polling با TTL مناسب کافی است. اگر ترافیک بالا دارید و می‌خواهید منابع را ذخیره کنید، SSE انتخاب بهتری است. WebSocket فقط برای سناریوهای بسیار خاص.

ساخت اولین endpoint آمار زنده

حالا بیایید یک endpoint کامل و مقاوم بسازیم.

# analytics/api.py

from datetime import timedelta

from django.contrib.auth.decorators import user_passes_test
from django.core.cache import cache
from django.http import JsonResponse
from django.utils import timezone
from django.views.decorators.http import require_GET

from analytics.models import Visit, PageView
from analytics.templatetags.analytics_tags import (
    online_count,
    today_unique_visitors,
    today_pageviews,
    today_total_minutes,
)


def is_analytics_viewer(user):
    return user.is_authenticated and (
        user.is_staff
        or user.groups.filter(name="Analytics Viewers").exists()
    )


def _check_rate_limit(user_id, limit=120, window=60):
    key = f"ratelimit:stats:{user_id}"
    current = cache.get(key, 0)

    if current >= limit:
        return False

    if current == 0:
        cache.set(key, 1, timeout=window)
    else:
        cache.incr(key)

    return True


def _cached(key, compute_fn, ttl):
    value = cache.get(key)
    if value is not None:
        return value, True

    value = compute_fn()
    cache.set(key, value, timeout=ttl)
    return value, False


@require_GET
@user_passes_test(is_analytics_viewer)
def stats_summary(request):
    if not _check_rate_limit(request.user.pk):
        return JsonResponse(
            {"error": "rate_limited"},
            status=429,
        )

    online, c1 = _cached("analytics:online", online_count, ttl=15)
    visitors, c2 = _cached(
        "analytics:today_visitors", today_unique_visitors, ttl=60,
    )
    pageviews, c3 = _cached(
        "analytics:today_pageviews", today_pageviews, ttl=60,
    )
    minutes, c4 = _cached(
        "analytics:today_minutes", today_total_minutes, ttl=60,
    )

    payload = {
        "generated_at": timezone.now().isoformat(),
        "cached": c1 and c2 and c3 and c4,
        "cache_ttl": 15,
        "stats": {
            "online": online,
            "today_visitors": visitors,
            "today_pageviews": pageviews,
            "today_minutes": minutes,
        },
    }

    response = JsonResponse(payload)
    response["Cache-Control"] = "private, max-age=15"
    return response

این endpoint، هفت مزیت نسبت به نسخه‌ی ساده دارد:

مزیت اول، کش چندلایه با TTLهای متفاوت.

مزیت دوم، rate limiting بر اساس user.

مزیت سوم، کنترل دسترسی با group.

مزیت چهارم، استفاده از template tags مشترک. یعنی همان منطق در پنل و در endpoint.

مزیت پنجم، payload ساختاریافته. شامل timestamp و وضعیت کش.

مزیت ششم، هدر Cache-Control. کلاینت می‌داند داده چقدر معتبر است.

مزیت هفتم، استفاده از HTTP 429 برای rate limit. استاندارد و قابل تشخیص برای کلاینت.

اگر با الگوی انتقال منطق از services.py به templatetags کار کرده باشید، می‌دانید که این نوع بازاستفاده، بخشی از انضباط معماری است.

کارایی؛ اعداد واقعی در مقیاس

کارایی endpoint آمار زنده، به چند عامل بستگی دارد. این اعداد، از تجربه‌ی واقعی پروژه‌های مختلف است.

نکته‌ی مهم: بدون کش، دیتابیس در سطح ۱۰۰ کاربر همزمان داشبورد اشباع می‌شود. با کش ۳۰ ثانیه، می‌توانید تا هزاران کاربر همزمان را سرو کنید. با تجمیع، حتی به تأخیر پایین‌تر هم می‌رسید.

در یکی از پروژه‌ها، بعد از اعمال کش و جدول تجمیعی، تأخیر endpoint از ۱.۲ ثانیه به ۱۲ میلی‌ثانیه رسید. این بهبود، تأثیر مستقیم روی تجربه‌ی کاربران داشبورد داشت.

مانیتورینگ و هشدار

endpoint آمار زنده، یک endpoint حساس است. باید به‌طور مداوم مانیتور شود. سه لایه‌ی مانیتورینگ را در نظر بگیرید.

لایه‌ی اول، لاگ ساختاریافته. هر درخواست، باید لاگ شود با زمان پاسخ، وضعیت کش و کاربر.

import logging

logger = logging.getLogger("analytics.api")


def log_stats_request(user, duration_ms, cached):
    logger.info(
        "stats_request",
        extra={
            "user_id": user.pk,
            "duration_ms": duration_ms,
            "cached": cached,
        },
    )

لایه‌ی دوم، متریک‌های Prometheus. اگر از Prometheus استفاده می‌کنید، تعداد درخواست‌ها، تأخیر و نرخ خطا را ثبت کنید.

from prometheus_client import Counter, Histogram


STATS_REQUESTS = Counter(
    "analytics_stats_requests_total",
    "Total stats requests",
    ["status"],
)

STATS_LATENCY = Histogram(
    "analytics_stats_latency_seconds",
    "Stats request latency",
)

لایه‌ی سوم، هشدار. اگر تأخیر یا نرخ خطا از آستانه گذشت، هشدار دریافت کنید.

تست endpoint آمار زنده

سه سطح تست را در نظر بگیرید.

سطح اول، تست دسترسی. بررسی کنید که فقط کاربران مجاز دسترسی دارند.

@pytest.mark.django_db
def test_anonymous_gets_403(client):
    response = client.get("/panel/stat/api/summary/")
    assert response.status_code in (302, 403)


@pytest.mark.django_db
def test_staff_gets_200(client, django_user_model):
    user = django_user_model.objects.create_user(
        "admin", "a@example.com", "pass", is_staff=True,
    )
    client.force_login(user)

    response = client.get("/panel/stat/api/summary/")
    assert response.status_code == 200

سطح دوم، تست rate limiting. بررسی کنید که محدودیت نرخ اعمال می‌شود.

@pytest.mark.django_db
def test_rate_limit(client, django_user_model):
    user = django_user_model.objects.create_user(
        "admin", "a@example.com", "pass", is_staff=True,
    )
    client.force_login(user)

    for i in range(150):
        response = client.get("/panel/stat/api/summary/")

    assert response.status_code == 429

سطح سوم، تست کش. بررسی کنید که کش درست کار می‌کند.

@pytest.mark.django_db
def test_response_uses_cache(client, django_user_model):
    from django.core.cache import cache

    cache.clear()

    user = django_user_model.objects.create_user(
        "admin", "a@example.com", "pass", is_staff=True,
    )
    client.force_login(user)

    response1 = client.get("/panel/stat/api/summary/")
    data1 = response1.json()

    response2 = client.get("/panel/stat/api/summary/")
    data2 = response2.json()

    assert data2["cached"] is True
    assert data2["stats"] == data1["stats"]

این سه سطح تست، به شما اجازه می‌دهند که در طول زمان، تغییرات را با اطمینان اعمال کنید.

anti-patternهای رایج در ساخت API آمار

در بازبینی پروژه‌های مختلف، این اشتباهات را زیاد دیده‌ام:

۱. عدم کش. بدون کش، هر درخواست چندین کوئری سنگین می‌زند. راه‌حل: کش چندلایه با TTLهای متفاوت.

۲. کش با TTL یکسان. اگر همه‌ی داده‌ها با TTL یکسان کش شوند، داده‌های زنده کند می‌شوند یا داده‌های سنگین سریع expire می‌شوند.

۳. عدم rate limiting. endpoint آمار باید محدودیت نرخ داشته باشد.

۴. عدم کنترل دسترسی. آمار نباید برای همه در دسترس باشد.

۵. محاسبه در ویو به‌جای tag مشترک. اگر منطق آماری در ویو باشد، با پنل تکرار می‌شود. راه‌حل: استفاده از tagهای مشترک.

۶. محاسبه در هر درخواست از داده‌ی خام. باید از جدول‌های تجمیعی استفاده کنید.

۷. عدم cache-control header. کلاینت باید بداند داده چقدر معتبر است.

۸. عدم مدیریت خطا. اگر یک کوئری شکست خورد، endpoint باید به‌جای 500، یک پاسخ ساختاریافته بدهد.

۹. polling با نرخ بالا. اگر polling هر ثانیه باشد، بار سرور بالا می‌رود. حداقل ۱۰ تا ۳۰ ثانیه.

۱۰. عدم monitoring. بدون monitoring، مشکلات کارایی دیر کشف می‌شوند.

۱۱. ذخیره‌ی session در دیتابیس. اگر session روی دیتابیس باشد، هر درخواست AJAX یک کوئری اضافه دارد. راه‌حل: session در Redis.

۱۲. عدم استفاده از HTTP/2. با HTTP/2، درخواست‌های موازی به یک endpoint سریع‌تر انجام می‌شوند.

اگر با الگوی بهینه‌سازی جنگو برای ترافیک بالا کار کرده باشید، این anti-patternها برایتان آشناست.

پرسش‌های پرتکرار درباره‌ی API آمار زنده

چند وقت یک بار داشبورد را به‌روز کنم؟ بین ۱۵ تا ۳۰ ثانیه. کمتر از ۱۰ ثانیه معمولاً ارزشش را ندارد و بار سرور را بالا می‌برد.

آیا باید از WebSocket استفاده کنم؟ فقط برای داشبوردهای تعاملی با نیاز به تأخیر بسیار کم. برای اکثر پروژه‌ها، polling کافی است.

TTL کش مناسب چقدر است؟ برای داده‌ی خیلی زنده، ۱۵ ثانیه. برای داده‌ی روزانه، ۶۰ ثانیه. برای داده‌ی تاریخی، می‌توانید کش طولانی‌تر داشته باشید.

چطور کش را invalidate کنم؟ در کامندهای پاک‌سازی داده و در beaconهای مهم. یک تابع متمرکز بنویسید.

آیا باید هدر Cache-Control بفرستم؟ بله، همیشه. با private برای داده‌ی کاربر-محور.

چطور با دیتابیس اشباع مقابله کنم؟ با کش، جدول‌های تجمیعی و rate limiting.

آیا باید API را نسخه‌بندی کنم؟ برای APIهای عمومی، بله. برای داشبورد داخلی، معمولاً نیازی نیست.

چطور با کاربران موبایل که در حرکت هستند کار کنم؟ داشبورد باید responsive باشد و در شبکه‌های ضعیف، کش طولانی‌تر داشته باشد.

آیا باید از select_related و prefetch_related استفاده کنم؟ در کوئری‌های aggregation، معمولاً نه. این ابزارها برای دسترسی به روابط طراحی شده‌اند، نه برای aggregation.

چطور با cache stampede مقابله کنم؟ با یک lock (Redis یا cache محلی) که فقط یک worker اجازه‌ی محاسبه‌ی مجدد را دارد.

def get_cached_with_lock(key, compute_fn, ttl=30):
    value = cache.get(key)
    if value is not None:
        return value

    lock_key = f"{key}:lock"
    if cache.add(lock_key, 1, timeout=10):
        try:
            value = compute_fn()
            cache.set(key, value, timeout=ttl)
            return value
        finally:
            cache.delete(lock_key)
    else:
        # صبر کن و دوباره بخوان
        import time
        for _ in range(20):
            time.sleep(0.1)
            value = cache.get(key)
            if value is not None:
                return value
        return compute_fn()

آیا باید داده را در JSON یا فرمت سبک‌تری بفرستم؟ JSON کافی است. اگر payload بزرگ است، می‌توانید از msgpack استفاده کنید، ولی معمولاً نیازی نیست.

چطور با گم شدن session در polling مقابله کنم؟ session معمولاً پایدار است چون درخواست‌ها منظم هستند.

آیا باید authentication را در هر درخواست انجام دهم؟ بله، endpoint باید در هر درخواست احراز هویت کند.

چطور با multi-tenant مقابله کنم؟ با یک فیلتر بر اساس tenant در کوئری.

آیا باید داده‌ی آماری را در localStorage ذخیره کنم؟ برای نمایش سریع در بازگشت به صفحه، مفید است. ولی داده‌ی نشان‌داده‌شده باید مشخص شود که از cache محلی می‌آید.

نگاهی از منظر مهندس داده در مقیاس میلیونی

در مقیاس میلیون‌ها درخواست در روز، endpoint آمار زنده تبدیل به یک مسئله‌ی مهندسی داده می‌شود. سه مفهوم بنیادین را باید بازتعریف کنید.

مفهوم اول، جدول‌های تجمیعی چندسطحی. به‌جای محاسبه‌ی هر بار از داده‌ی خام، از چند سطح تجمیع استفاده کنید: دقیقه‌ای، ساعتی، روزانه، ماهانه. هر سطح، یک جدول جداگانه دارد و بر اساس نیاز، از سطح مناسب استفاده می‌شود.

مفهوم دوم، streaming aggregation. به‌جای batchهای شبانه، می‌توانید از یک stream processor (مثل Kafka Streams یا Flink) برای تجمیع لحظه‌ای استفاده کنید. این معماری، تأخیر داده را از چند ساعت به چند ثانیه کاهش می‌دهد. مفهوم Stream Processing در ویکی‌پدیا توضیح داده شده است.

مفهوم سوم، read replica. به‌جای کوئری زدن روی دیتابیس اصلی، از یک read replica استفاده کنید. این کار، فشار آماری را از دیتابیس اصلی جدا می‌کند.

class AnalyticsRouter:
    def db_for_read(self, model, **hints):
        if model._meta.app_label == "analytics":
            return "analytics_replica"
        return None

نکته‌ی آخر: در مقیاس بالا، endpoint آمار زنده نباید مستقیم از دیتابیس رابطه‌ای بخواند. باید از یک cache توزیع‌شده (Redis) یا یک دیتابیس ستونی (ClickHouse، Druid) که برای خواندن سریع طراحی شده استفاده کند. اگر با الگوی ساخت API JSON برای آمار زنده کار کرده باشید، این معماری برایتان آشناست.

پرسشی که در پایان باید پاسخ دهید

قبل از اینکه API آمار زنده خود را نهایی کنید، یک پرسش را از خودتان بپرسید: «اگر امروز ۵۰۰ کاربر همزمان داشبورد را باز کنند، آیا سیستم من پایداری خود را حفظ می‌کند؟» اگر پاسخ شما «بستگی دارد» است، یعنی سیستم شما به لایه‌های کش و rate limiting بیشتری نیاز دارد. اگر پاسخ شما «بله، بدون مشکل» است، یعنی سیستم شما به‌درستی طراحی شده است.

طراحی API آمار زنده، در نهایت یک تصمیم مهندسی است که به تجربه‌ی کاربر و پایداری سیستم شما گره خورده است. اگر این تجربه را در پروژه‌ی خودتان داشته‌اید — مثلاً جایی که endpoint شما در حجم بالا کند شده یا جایی که کش نجات‌تان داده — برایم جالب است بدانید. مخصوصاً اگر راه‌حل خاصی برای یک سناریوی خاص پیدا کرده‌اید، چون همان راه‌حل‌ها می‌توانند به خواننده‌ی بعدی کمک کنند. تجربه‌ی خودتان را در دیدگاه‌ها بنویسید.