Connectors bulk et la clé d'unicité

L'interface : des Protocols qui streament

Importer un catalogue peut venir de partout : CSV uploadé, ERP Odoo, PIM Akeneo. Le package connectors définit des Protocols par domaine (CatalogSource, AccountSource, PriceSource…) avec une signature commune :

class CatalogSource(Protocol):
    def fetch_all(self) -> AsyncIterator[ProductHit]: ...     # STREAMING, pas une liste
    @property
    def errors(self) -> list[ImportRowError]: ...             # les erreurs par ligne, collectées

Deux choix de design à retenir : AsyncIterator (un catalogue de 500 k produits ne tient pas en RAM — on streame) et les erreurs collectées ligne à ligne au lieu d'un échec global (une ligne pourrie ne doit pas faire échouer 499 999 bonnes). Les DTOs normalisés (ProductHit…) sont des Pydantic extra="forbid" avec external_id comme identité et le reste en extensions JSONB. Un registry fait factory par type d'intégration ; les connectors non implémentés renvoient un 501 propre (ConnectorNotImplementedError). Les credentials des intégrations sont chiffrés Fernet en base (module 11).

Étude de mini-cas : le connecteur Odoo, bidirectionnel

Le premier connecteur ERP complet du projet illustre la maturité du pattern — livré en une semaine, en quatre briques indépendantes :

1. le transport : client XML-RPC (l'API historique d'Odoo) + transport JSON-2 pour Odoo 19+ — le protocole isolé derrière la même interface ;
2. le test de connexion : POST /integrations/{id}/test — valider credentials et joignabilité AVANT le premier import ; le petit endpoint qui économise des heures de debug « pourquoi mon import est vide » ;
3. l'import référentiel (source) : product.product → variants via le même chemin fetch_all() que le CSV — le registry rend le nouveau connecteur invisible pour l'importeur ;
4. l'export (sink) : OdooOrderSink — la commande confirmée part vers l'ERP à la confirmation, idempotente par document.id.

La leçon d'architecture : parce que les Protocols étaient en place, ajouter un ERP entier n'a pas touché une ligne de l'importeur ni du pipeline — uniquement des adaptateurs. C'est le test ultime d'une bonne interface : le jour où le « vrai » fournisseur arrive, il se glisse dedans.

L'upsert : create vs update sans N+1

existants = await load_by_external_ids(hits)        # UNE requête batch, pas 50 000
for hit in hits:
    if hit.external_id in existants: update(...)    # merge des extensions
    else: insert(...)
# résultat : ImportResult(created=1200, updated=48800)

Rejouer le même fichier = created=0, updated=50000 — l'import est idempotent par construction, et le rapport created/updated est le premier contrôle qualité de l'opérateur.

La clé d'unicité : le hash canonique (ADR-0006)

Comment garantir l'unicité quand l'identité dépend de plusieurs champs configurables (un prix = variant + account + devise + tier…) ? Le pattern du projet :

dimensions (configurables par FieldDefinition)
  → JSON canonique : clés triées, None→"", Decimal normalisé, UUID lowercase
  → SHA-256 → colonne unicity_key
  → UNIQUE(tenant_id, unicity_key)
  → INSERT ... ON CONFLICT (tenant_id, unicity_key) DO UPDATE

Toute la valeur est dans le mot canonique : la moindre variation de représentation (1.000000 vs 1 sur un NUMERIC, un UUID en majuscules) produirait deux hashs pour la même identité — d'où la normalisation stricte, et l'invariant redoutable : un backfill SQL doit produire exactement le même hash que Python. Le gain : l'unicité composite et configurable tient dans une contrainte simple, et l'upsert devient un ON CONFLICT standard.