Come Gestire le Notifiche in Tempo Reale con Celery, Twilio e FastAPI

Quando si sviluppano applicazioni moderne, le notifiche in tempo reale sono una componente fondamentale, non solo tramite email ma anche via SMS. I servizi di invio email e messaggistica SMS devono essere affidabili, scalabili e non influire sulle performance generali dell’applicazione. In questo contesto, Celery si rivela uno strumento potente per gestire l'invio di email e SMS in modo asincrono, migliorando l'affidabilità e la reattività dell’applicazione.

Inviare Email con Celery

Quando un'applicazione riceve una richiesta che richiede l'invio di un'email, come ad esempio una registrazione di un nuovo utente, è fondamentale evitare di eseguire questa operazione in modo sincrono, cioè direttamente nel flusso della richiesta. Invece, l'invio delle email dovrebbe essere delegato a un task di Celery, che viene eseguito in background. Ciò consente di restituire subito la risposta all'utente, senza farlo aspettare mentre l'email viene inviata. Il task Celery per l'invio di un’email potrebbe apparire così:

python
Copy code
from fastapi import APIRouter

from app.tasks import send_email_task
router = APIRouter()
@router.post("/register")
def register_user(email: str):
    subject = "Benvenuto!"
    html_body = "<p>Grazie per esserti registrato!</p>"
    send_email_task.delay(email, subject, html_body)
    return {"msg": "Registrazione completata. Controlla la tua email."}

Con l’uso del metodo .delay(), il task viene messo in coda su Redis e successivamente processato da un worker Celery. Questo approccio assicura che l’API non venga rallentata, poiché il lavoro di invio dell’email è gestito in background.

Monitoraggio dei Task con Flower

Per monitorare l’esecuzione dei task, Flower è uno strumento essenziale. Flower è un’applicazione web che consente di visualizzare in tempo reale lo stato dei task di Celery, come il numero di task attivi, quelli riusciti o falliti, le ritentative e lo stato della coda. Questo aiuta a diagnosticare problemi senza dover consultare i log.

Per avviare Flower, basta eseguire il comando:

bash
Copy code
pip install flower

celery -A app.celery_worker.celery_app flower --port=5555

Una volta avviato Flower, sarà possibile accedere alla dashboard su http://localhost:5555 e visualizzare informazioni cruciali riguardo i task.

Notifiche SMS con Twilio

Oltre alle email, le applicazioni moderne devono gestire anche notifiche SMS, che sono spesso più immediate e visibili. Twilio è uno dei principali provider per l'invio di SMS a livello globale e può essere integrato facilmente con Celery per gestire l’invio asincrono di SMS.

python
Copy code
import os
TWILIO_ACCOUNT_SID = os.getenv("TWILIO_ACCOUNT_SID")
TWILIO_AUTH_TOKEN = os.getenv("TWILIO_AUTH_TOKEN")
TWILIO_FROM_NUMBER = os.getenv("TWILIO_FROM_NUMBER")
TWILIO_API_URL = f"https://api.twilio.com/2010-04-01/Accounts/{TWILIO_ACCOUNT_SID}/Messages.json"

Inviare SMS tramite HTTPX

Piuttosto che utilizzare il SDK ufficiale di Twilio, è possibile inviare SMS utilizzando HTTPX, che è un client HTTP asincrono. Un esempio di funzione riutilizzabile per inviare un SMS è il seguente:

python
Copy code
import httpx

from config import TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN, TWILIO_FROM_NUMBER, TWILIO_API_URL
async def send_sms_via_twilio(to_number: str, body: str, status_callback_url: str = None):
    data = {
        "From": TWILIO_FROM_NUMBER,
        "To": to_number,
        "Body": body,
    }
    if status_callback_url:
        data["StatusCallback"] = status_callback_url
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            TWILIO_API_URL,
            data=data,
            auth=(TWILIO_ACCOUNT_SID, TWILIO_AUTH_TOKEN)
        )
        resp.raise_for_status()
        return resp.json()

Gestire gli SMS con Celery

Poiché l’invio di SMS è un’operazione asincrona, Celery gestisce la coda degli SMS, assicurando che i task vengano eseguiti senza ritardi nel flusso principale dell’applicazione. La definizione del task Celery per l’invio degli SMS è la seguente:

python
Copy code
from app.celery_worker import celery_app
from app.sms_utils import send_sms_via_twilio
import asyncio

@celery_app.task(bind=True, max_retries=4, default_retry_delay=40)

def send_sms_task(self, to_number, message, status_callback_url=None):
    try:
        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(
            send_sms_via_twilio(to_number, message, status_callback_url)
        )
        return result
    except Exception as exc:
        raise self.retry(exc=exc)

Ogni invio di SMS, che si tratti di un codice di recupero password o una notifica di autenticazione a due fattori, passa attraverso un task in background, evitando così di bloccare l’applicazione principale.

Webhook di Twilio per Monitoraggio

python
Copy code
from fastapi import APIRouter, Request, status
from fastapi.responses import JSONResponse
router = APIRouter()
@router.post("/webhooks/twilio-sms-status")

async def twilio_sms_status_webhook(request: Request):

    data = await request.form()
    message_sid = data.get("MessageSid")
    message_status = data.get("MessageStatus")
    to_number = data.get("To")
    print(f"Twilio SMS: {message_sid} to {to_number} is now {message_status}")
    return JSONResponse({"status": "received"}, status_code=status.HTTP_200_OK)

Questo endpoint può essere utilizzato per ricevere notifiche di stato da Twilio, aggiornare il database o inviare notifiche all'utente riguardo lo stato della loro richiesta.

Retry e Monitoraggio dei Fallimenti

La logica di retry di Celery è un’altra caratteristica che migliora la robustezza del sistema. Se l'invio di un SMS fallisce (ad esempio, a causa di un errore temporaneo di Twilio o di una connessione di rete instabile), Celery ritenta automaticamente il task secondo la configurazione. Ogni tentativo e risultato dei task viene visualizzato in Flower, consentendo agli amministratori di monitorare in tempo reale i problemi di consegna.

Importanza di Una Buona Gestione delle Notifiche

Come Implementare il Caching Efficiente con Redis in FastAPI

Nel contesto di applicazioni moderne, la gestione efficiente delle risorse e la riduzione dei tempi di risposta sono fattori cruciali per garantire prestazioni ottimali. Uno degli strumenti più efficaci in questo ambito è Redis, un sistema di archiviazione in memoria che permette di memorizzare e recuperare dati con una velocità straordinaria. Integrare Redis con FastAPI, un framework web Python che supporta async/await, consente di ottenere un caching altamente performante per applicazioni web di qualsiasi dimensione.

L'approccio che vedremo riguarda principalmente il caching a livello di funzione e a livello di rotta, con un occhio attento alla gestione della TTL (Time-To-Live) e alla strategia di invalidazione. Questi concetti sono fondamentali per ottimizzare l'uso della memoria e migliorare la reattività delle applicazioni.

Configurazione Iniziale di Redis

python
Copy code
import aioredis
import os
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
redis = None
async def get_redis():
    global redis
    if redis is None:
        redis = await aioredis.from_url(REDIS_URL, encoding="utf-8", decode_responses=True)
    return redis

Questa utility consente di gestire sia piccole che grandi implementazioni, poiché permette la configurazione tramite variabili d'ambiente e la connessione "pigra" su richiesta.

Caching a Livello di Funzione

Per implementare questa soluzione, possiamo creare un decoratore che memorizza in cache i risultati delle funzioni asincrone. Ecco un esempio di implementazione:

python
Copy code
import functools
import json
from app.redis_client import get_redis

def redis_cache(key_template: str, ttl: int = 300):

    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            cache_key = key_template.format(*args, **kwargs)
            redis = await get_redis()
            cached = await redis.get(cache_key)
            if cached is not None:
                return json.loads(cached)
            result = await func(*args, **kwargs)
            await redis.set(cache_key, json.dumps(result), ex=ttl)
            return result
        return wrapper
    return decorator

Un esempio di come utilizzare questo decoratore per memorizzare il risultato di una funzione che simula la chiamata a un database o API:

python
Copy code
@redis_cache('user_profile:{user_id}', ttl=600)

async def fetch_user_profile(user_id: int):
    await asyncio.sleep(2)  # Simula una chiamata lenta

    return {"user_id": user_id, "profile": "Sample user profile"}

La prima volta che la funzione viene chiamata, essa viene eseguita normalmente. Le chiamate successive, con gli stessi argomenti, restituiranno il risultato memorizzato in Redis, migliorando drasticamente la latenza.

Caching a Livello di Rotta

A volte, è necessario memorizzare l'intera risposta HTTP, in particolare per le rotte che restituiscono dati che cambiano raramente, come le pagine pubbliche o i report lenti. In questo caso, possiamo implementare il caching a livello di rotta. Qui di seguito viene mostrato come implementare un sistema di caching per una rotta in FastAPI.

python
Copy code
from fastapi import Request, Response

from app.redis_client import get_redis
async def cache_response(request: Request, ttl: int = 120):
    redis = await get_redis()
    cache_key = f"route_cache:{request.url.path}?{request.url.query}"
    cached = await redis.get(cache_key)
    if cached:
        return Response(content=cached, media_type="application/json")
    return None

Nel router, verifichiamo prima se esiste una risposta cache. Se non c'è, procediamo con il calcolo dei dati richiesti e poi memorizziamo la risposta.

python
Copy code
@router.get("/expensive-resource")

async def get_expensive_resource(request: Request):
    cached_response = await cache_response(request, ttl=180)
    if cached_response:
        return cached_response
    await asyncio.sleep(3)  # Simula operazioni lente
    data = {"expensive": "This result is cached for 3 minutes."}
    redis = await get_redis()
    cache_key = f"route_cache:{request.url.path}?{request.url.query}"
    await redis.set(cache_key, json.dumps(data), ex=180)
    return data

Con questo approccio, la rotta ricalcola i dati solo quando la cache è scaduta o è stata invalidata esplicitamente.

Gestione della TTL (Time-To-Live)

A seconda del tipo di dati, possiamo scegliere una TTL più breve o più lunga:

• TTL breve (30–120 secondi): per dati che cambiano rapidamente o risultati di API sensibili al tempo.

• TTL media (5–30 minuti): per dashboard utente, risultati di ricerca o report periodici.

• TTL lunga (ore o giorni): per contenuti statici, documentazione o dati di riferimento che cambiano raramente.

Strategie di Invalidazione

Sebbene la TTL garantisca l'espirazione automatica delle cache, in alcuni casi può essere necessario forzare un aggiornamento della cache quando i dati sottostanti cambiano. Redis offre la possibilità di invalidare manualmente le chiavi utilizzando la funzione delete. Ad esempio, se un utente aggiorna il proprio profilo, possiamo rimuovere la chiave relativa alla cache di quel profilo:

python
Copy code
redis = await get_redis()

await redis.delete(f"user_profile:{user_id}")

Per l'invalidazione in blocco, come nel caso di aggiornamenti di massa, possiamo utilizzare il comando SCAN o KEYS di Redis per trovare tutte le chiavi relative e rimuoverle.

python
Copy code
async def invalidate_all_route_caches():
    redis = await get_redis()
    keys = await redis.keys("route_cache:*")
    if keys:
        await redis.delete(*keys)

Gestione del Rate Limiting

Il rate limiting può essere implementato utilizzando uno dei due algoritmi principali: la finestra fissa e la finestra mobile, ciascuno con vantaggi e svantaggi in termini di accuratezza e tolleranza ai picchi.

Conclusione

L'integrazione di Redis con FastAPI per il caching e il rate limiting è una soluzione potente per migliorare le prestazioni delle applicazioni web moderne. Usando il caching a livello di funzione e rotta, la gestione della TTL, e le strategie di invalidazione, possiamo ottimizzare i tempi di risposta e ridurre il carico sui sistemi backend. Redis, con la sua capacità di gestire grandi volumi di dati in memoria, si dimostra un alleato imprescindibile per applicazioni scalabili ed efficienti.

Come Implementare un Sistema di Rate Limiting e Gestire i Processi in Background con Celery e Redis in FastAPI

Nel contesto di una moderna applicazione web, la gestione delle richieste e la prevenzione di un sovraccarico del server sono aspetti cruciali per mantenere un servizio fluido e sicuro. Due concetti che emergono in questo ambito sono il rate limiting e l'uso di code di lavoro in background per gestire processi complessi e a lungo termine. In questa sezione, esploreremo come implementare un sistema di rate limiting utilizzando algoritmi come il "fixed window" e lo "sliding window" con Redis e come delegare attività in background tramite Celery.

Il rate limiting è una tecnica fondamentale per impedire che gli utenti o le applicazioni effettuino un numero eccessivo di richieste in un intervallo di tempo troppo breve. Questo approccio previene abusi, attacchi DDoS e migliora le prestazioni globali dell'applicazione. Un esempio pratico di implementazione in FastAPI è rappresentato dal codice che utilizza Redis per contare il numero di richieste effettuate da un determinato utente o IP in una finestra temporale definita.

Algoritmo del Fixed Window Rate Limiting

L'algoritmo di rate limiting basato sulla finestra fissa è uno dei metodi più semplici ed efficaci. In questo approccio, tutte le richieste effettuate all'interno di una finestra di tempo predefinita (ad esempio 60 secondi) vengono conteggiate e, se il numero di richieste supera un limite definito, il server restituisce un errore HTTP 429 (Too Many Requests).

Nel codice che segue, ogni richiesta incrementa un contatore salvato in Redis. Se è la prima richiesta in una finestra temporale, viene impostata una scadenza per il contatore, che corrisponde alla durata della finestra. Quando il numero di richieste supera il limite impostato, il sistema restituisce un errore con un'intestazione Retry-After che indica il tempo rimanente prima che la finestra venga resettata.

python
Copy code
async def fixed_window_rate_limit(request: Request, user_id: int = None, limit: int = 100, window_seconds: int = 60):

    redis = await get_redis()
    key = get_rate_limit_key(request, user_id, window="minute")
    current_count = await redis.incr(key)
    if current_count == 1:
        await redis.expire(key, window_seconds)
    if current_count > limit:
        ttl = await redis.ttl(key)
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Too many requests. Please try again later.",
            headers={"Retry-After": str(ttl if ttl > 0 else window_seconds)},
        )

Algoritmo dello Sliding Window Rate Limiting

Se il metodo precedente è relativamente semplice e facile da implementare, l'algoritmo dello sliding window (finestra scorrevole) è una versione più avanzata, che offre un controllo più preciso sulle richieste. In questo caso, ogni richiesta è associata a un timestamp specifico, e il conteggio delle richieste avviene su un intervallo mobile di tempo, invece di un periodo fisso. In altre parole, il sistema tiene traccia delle richieste effettuate negli ultimi N secondi, senza essere influenzato da "salti" ai confini dell'intervallo.

Con Redis, questo algoritmo può essere facilmente implementato utilizzando set ordinati (ZADD, ZREMRANGEBYSCORE, ZCARD). Ogni richiesta viene aggiunta a un set ordinato con il timestamp corrente e vengono rimosse le richieste al di fuori della finestra temporale.

python
Copy code
async def sliding_window_rate_limit(request: Request, user_id: int = None, limit: int = 100, window_seconds: int = 60):

    redis = await get_redis()
    key = get_rate_limit_key(request, user_id, window="sliding")
    now = int(time.time())
    window_start = now - window_seconds
    await redis.zadd(key, {str(now): now})
    await redis.zremrangebyscore(key, 0, window_start)
    req_count = await redis.zcard(key)
    await redis.expire(key, window_seconds)
    if req_count > limit:
        oldest = await redis.zrange(key, 0, 0, withscores=True)

        if oldest and len(oldest[0]) > 1:

            retry_after = window_seconds - (now - int(float(oldest[0][1])))
        else:
            retry_after = window_seconds
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded. Please try again later.",

            headers={"Retry-After": str(max(1, int(retry_after)))},

        )

L’algoritmo della finestra scorrevole aiuta a prevenire i picchi di traffico vicino ai confini dell’intervallo e offre un metodo più equo di limitazione, essendo meno vulnerabile agli attacchi o abusi.

Integrazione del Rate Limiting con le Rotte di FastAPI

Una volta implementato il sistema di rate limiting, possiamo integrarlo direttamente nelle rotte di FastAPI come una dipendenza o un middleware. Di seguito è riportato un esempio di come applicare il rate limiter a una rotta:

python
Copy code
from fastapi import APIRouter, Request, Depends
router = APIRouter()
@router.get("/public-data")

async def public_data(request: Request):

    await fixed_window_rate_limit(request, limit=20, window_seconds=60)
    return {"data": "This endpoint is rate-limited to 20 requests per minute per client."}

In questo esempio, ogni richiesta alla rotta /public-data è limitata a 20 richieste al minuto per client. Si può anche applicare lo sliding window per rotti più sensibili o per API a pagamento, adattando così il sistema alle esigenze specifiche dell'applicazione.

L'Importanza dell'Intestazione Retry-After

Un aspetto fondamentale nella gestione delle richieste è il corretto utilizzo dell'intestazione Retry-After, che viene restituita quando il limite di richieste è stato superato. Questa intestazione informa il client del tempo che deve attendere prima di poter effettuare una nuova richiesta. La corretta gestione di questa intestazione è essenziale per evitare sovraccarichi e per una comunicazione chiara con i client, come browser e librerie HTTP, che rispettano automaticamente questo valore.

L'uso di Celery per i Processi in Background

Mentre il rate limiting aiuta a controllare il flusso di richieste, ci sono operazioni che non possono essere completate immediatamente a causa della loro complessità. Ad esempio, la generazione di report, l'esportazione di grandi dataset o l'invio di notifiche in batch richiedono l'uso di code di lavoro in background per evitare di bloccare il flusso principale delle richieste. Celery, combinato con Redis come broker, è una delle soluzioni più potenti e testate per gestire questi compiti.

python
Copy code
from app.celery_worker import celery_app
import time
import random
@celery_app.task(bind=True)
def generate_report_task(self, report_id, user_id):
    time.sleep(random.randint(5, 10))

    result = {"report_id": report_id, "status": "ready", "url": f"/reports/{report_id}.pdf"}

    print(f"Report for {user_id} generated: {result['url']}")
    return result

Celery permette anche di eseguire attività di pulizia, come la rimozione di sessioni scadute o la compattazione dei log, automatizzando i processi di manutenzione dell'applicazione.

Conclusioni

Le tecniche di rate limiting e l'uso delle code in background sono strumenti indispensabili per la gestione efficace di un'applicazione web scalabile e reattiva. Implementare correttamente questi sistemi non solo protegge il server da richieste abusive, ma migliora anche l'esperienza dell'utente, riducendo i tempi di attesa e delegando compiti intensivi ai processi in background. Inoltre, l'integrazione con tecnologie come FastAPI, Redis e Celery fornisce un framework robusto per affrontare la complessità di sistemi moderni.

Come implementare una paginazione efficace e un filtraggio dinamico nei sistemi API

La gestione di grandi moli di dati nelle API rappresenta una sfida significativa, specialmente quando si tratta di consentire agli utenti di navigare agevolmente tra le informazioni. Una delle tecniche più comuni e utili per affrontare questo problema è la paginazione, che permette di suddividere i dati in piccoli "blocchi" (pagine), riducendo il carico di traffico e migliorando l'esperienza utente. Un sistema di paginazione che può risolvere in modo più raffinato questa sfida è la paginazione tramite cursore, che si distingue dalla tradizionale paginazione basata su numeri di pagina.

La chiave per gestire una navigazione senza interruzioni è l'uso di metadati, che forniscono informazioni supplementari sullo stato di navigazione. Ad esempio, l'inclusione di un'intestazione X-Next-Cursor nella risposta consente al client di sapere qual è il prossimo cursore da utilizzare per la richiesta successiva. Se non ci sono ulteriori risultati, l'assenza di questa intestazione segnala che l'elenco dei dati è stato completamente esplorato. Un altro vantaggio di questo approccio è la sua flessibilità, che consente di implementare una navigazione bidirezionale, ossia sia avanti che indietro, in modo dinamico, attraverso l'uso di due cursori distinti, uno per il "prossimo" e uno per il "precedente".

Per quanto riguarda la filtrazione avanzata e l'ordinamento dei dati, le necessità degli utenti diventano sempre più specifiche e complesse. In un'applicazione di successo, è fondamentale permettere agli utenti di cercare, filtrare e ordinare i dati secondo criteri specifici. Ad esempio, in un sistema di gestione di libri, un utente potrebbe voler filtrare i libri in base all'autore, al titolo, all'anno di pubblicazione o ad altri parametri, e allo stesso tempo ordinarli per data o per popolarità.

FastAPI facilita l'implementazione di queste funzionalità attraverso l'uso di query parameters, che permettono agli utenti di passare filtri dinamici direttamente nella richiesta. Questi parametri possono essere opzionali e, se presenti, vengono utilizzati per costruire dinamicamente la query al database. Il motore di database come SQLAlchemy consente di modificare la query in base ai filtri passati, mantenendo il codice pulito e facilmente scalabile.

Per implementare questa funzionalità, è possibile scrivere un endpoint che accetta parametri opzionali come l'autore, l'anno minimo e massimo di pubblicazione, e il titolo. Il codice costruisce la query in modo da includere solo i criteri pertinenti, garantendo così che il risultato sia sempre corretto, senza dover scrivere condizioni per ogni possibile combinazione di filtri. L'uso di ordinamento dinamico aggiunge ulteriore flessibilità, consentendo agli utenti di decidere come ordinare i risultati, ad esempio in ordine crescente o decrescente per titolo, anno, o autore. Questo approccio migliora l'esperienza dell'utente, permettendo un controllo completo sulla modalità di visualizzazione dei dati.

Un esempio pratico di come implementare la ricerca avanzata di libri con filtri e ordinamento dinamico è il seguente:

python
Copy code
from typing import Optional, List

from fastapi import Query
from sqlalchemy import asc, desc
@app.get("/books/search", response_model=List[Book])
def search_books(
    author: Optional[str] = Query(None, min_length=1, max_length=100),

    year_min: Optional[int] = Query(None, ge=1000, le=2100),

    year_max: Optional[int] = Query(None, ge=1000, le=2100),

    title: Optional[str] = Query(None, min_length=1, max_length=255),

    sort_by: Optional[str] = Query("year"),
    sort_dir: Optional[str] = Query("desc")
):
    sort_column = getattr(BookModel, sort_by, None)
    if not sort_column:
        sort_column = BookModel.year
    direction = desc if sort_dir.lower() == "desc" else asc
    query = db.query(BookModel)
    if author:
        query = query.filter(BookModel.author.ilike(f"% {author}%"))
    if year_min:
        query = query.filter(BookModel.year >= year_min)
    if year_max:
        query = query.filter(BookModel.year <= year_max)
    if title:
        query = query.filter(BookModel.title.ilike(f"% {title}%"))
    query = query.order_by(direction(sort_column))
    results = query.all()
    return [Book.from_orm(b) for b in results]

Questa architettura consente una gestione efficiente dei dati, senza sacrificare la flessibilità o la performance. Il sistema può facilmente adattarsi a nuove esigenze senza dover rifare tutto il backend da zero. Un'ulteriore potenzialità di questo approccio è la possibilità di scalare il sistema aggiungendo capacità di caching o utilizzando database distribuiti, a seconda delle necessità di traffico.