Synchroniser les consentements clients vers Imagino via une API Python Django

Dans le cadre d'un projet Imagino, j'ai eu à concevoir et implémenter une brique de synchronisation des consentements clients en temps réel. L'objectif : chaque modification de consentement opt-in email ou SMS dans notre application Django devait être immédiatement répercutée dans les tables standard d'Imagino, sans intervention manuelle, sans batch.

Cet article détaille l'architecture mise en place, les choix techniques et les points de vigilance rencontrés.

Le contexte et le besoin

L'application Django hébergée chez le client proposait une interface utilisateur dédiée à la gestion des consentements, un formulaire de préférences de communication accessible aux clients finaux, directement dans leur espace personnel. C'est via cette interface que les utilisateurs activaient ou désactivaient leurs opt-ins email et SMS.

Chaque soumission du formulaire créait une nouvelle entrée dans la table ConsentLog, avec un historique complet des modifications. Imagino, de son côté, attendait que ses tables standard de gestion des opt-ins soient à jour pour alimenter les campagnes email et SMS.

Le besoin était clair : à chaque nouvelle entrée dans ConsentLog : déclenchée par une action utilisateur sur l'interface, pousser les informations vers l'API Imagino pour mettre à jour le profil client correspondant en temps réel.

Le modèle de données Django

La table ConsentLog stockait chaque modification de consentement avec ses métadonnées :

from django.db import models
from django.utils import timezone

class ConsentLog(models.Model):
    # Clés d'identification, matching avec profil Imagino
    email      = models.EmailField(db_index=True)
    nom        = models.CharField(max_length=100)
    prenom     = models.CharField(max_length=100)

    # Consentements
    optin_email = models.BooleanField(default=False)
    optin_sms   = models.BooleanField(default=False)

    # Métadonnées
    created_at = models.DateTimeField(default=timezone.now)
    source     = models.CharField(max_length=50)  # web, app, crm...

    class Meta:
        # Pas de doublon : un seul enregistrement actif par email
        unique_together = ['email', 'nom', 'prenom']

La contrainte unique_together sur le triplet email + nom + prénom est importante : elle garantit qu'on ne crée pas de doublon côté Django, et c'est aussi la clé de matching utilisée pour retrouver le profil dans Imagino.

Le signal Django : déclenchement temps réel

Le signal post_save de Django est déclenché automatiquement après chaque sauvegarde d'une instance de modèle. C'est le point d'entrée de la synchronisation.

from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import ConsentLog
from .imagino_client import push_consent_to_imagino

@receiver(post_save, sender=ConsentLog)
def sync_consent_to_imagino(sender, instance, created, **kwargs):
    # On ne synchronise que les nouvelles entrées
    # Pas d'update, chaque changement crée un nouvel enregistrement
    if not created:
        return

    push_consent_to_imagino(instance)

Le client API Imagino

La logique d'appel à l'API Imagino est isolée dans un module dédié. Ça facilite les tests unitaires et la maintenance.

import requests
from django.conf import settings
from .models import ErrorLogConsentAPI

IMAGINO_API_URL  = settings.IMAGINO_API_URL
IMAGINO_API_KEY  = settings.IMAGINO_API_KEY

def push_consent_to_imagino(consent: ConsentLog) -> None:
    payload = {
        "email":       consent.email,
        "nom":         consent.nom,
        "prenom":      consent.prenom,
        "optin_email": consent.optin_email,
        "optin_sms":   consent.optin_sms,
    }

    try:
        response = requests.post(
            f"{IMAGINO_API_URL}/consents",
            json=payload,
            headers={
                "Authorization": f"Bearer {IMAGINO_API_KEY}",
                "Content-Type": "application/json",
            },
            timeout=5,
        )
        response.raise_for_status()

    except requests.RequestException as exc:
        # Log de l'erreur pour traitement ultérieur
        ErrorLogConsentAPI.objects.create(
            consent_log = consent,
            error_code  = getattr(exc.response, "status_code", 0),
            error_msg   = str(exc),
        )

La gestion des erreurs : errorLogConsentAPI

Plutôt que de lever une exception qui bloquerait le flux utilisateur, les erreurs d'appel API sont capturées et stockées dans une table dédiée ErrorLogConsentAPI.

class ErrorLogConsentAPI(models.Model):
    consent_log = models.ForeignKey(
        ConsentLog,
        on_delete=models.CASCADE,
        related_name='errors'
    )
    error_code  = models.IntegerField()
    error_msg   = models.TextField()
    created_at  = models.DateTimeField(auto_now_add=True)
    resolved    = models.BooleanField(default=False)

    class Meta:
        ordering = ['-created_at']

Cette table est stockée directement dans Imagino, c'est une table custom créée spécifiquement pour ce projet, en dehors des tables standard de la plateforme. Elle permet à l'équipe de voir en un coup d'œil les synchronisations qui ont échoué, d'en connaître la cause (timeout, profil introuvable, doublon…) et de rejouer manuellement ou automatiquement les entrées en erreur, sans quitter l'environnement Imagino.

Les cas d'erreur rencontrés

Profil introuvable dans Imagino

L'API Imagino retourne une erreur quand le triplet email + nom + prénom ne correspond à aucun profil existant. C'est le cas le plus fréquent, un utilisateur s'inscrit et donne son consentement avant que son profil ait été créé dans Imagino (délai d'ingestion).

La solution : l'erreur est loggée avec le code approprié, et un job de réconciliation nocturne rejoue les entrées en erreur une fois les profils créés.

Doublon détecté

Quand l'API Imagino détecte plusieurs profils correspondant aux mêmes clés d'identification (email + nom + prénom non unique dans Imagino), elle retourne une erreur de lecture client. Ce cas révèle un problème de qualité de données en amont, des profils en doublon dans Imagino qu'il faut fusionner.

Timeout API

Un timeout de 5 secondes est configuré sur chaque appel. En cas de dépassement, l'erreur est loggée et le consentement sera rejoué lors de la réconciliation. L'utilisateur n'est jamais bloqué par un problème de disponibilité de l'API Imagino.

Configuration et sécurité

Les paramètres d'API ne sont jamais dans le code, ils sont dans les variables d'environnement :

import os

IMAGINO_API_URL = os.environ.get("IMAGINO_API_URL")
IMAGINO_API_KEY = os.environ.get("IMAGINO_API_KEY")

Ce qu'on aurait fait différemment

• Déléguer à Celery dès le début. L'appel API synchrone dans le signal fonctionne, mais une tâche Celery asynchrone serait plus propre, meilleure gestion des retry, pas d'impact sur le temps de réponse utilisateur.
• Implémenter un mécanisme de retry automatique. Rejouer manuellement les erreurs fonctionne mais un retry exponentiel automatique sur les timeouts aurait réduit la charge de réconciliation nocturne.
• Monitorer les erreurs en temps réel. Une alerte sur un taux d'erreur anormal aurait permis de détecter plus vite les problèmes de qualité de données dans Imagino.

Conclusion

Cette architecture de synchronisation temps réel est simple, robuste et maintenable. Le signal Django garantit qu'aucun consentement n'est perdu, la table d'erreurs permet de tracer et rejouer les échecs, et la logique d'insert-only simplifie considérablement la gestion des états.

Elle fonctionne bien pour des volumes modérés (quelques dizaines de modifications par minute). Au-delà, le passage à Celery s'impose pour éviter toute dégradation des temps de réponse.