API BCE/KBO Intégration Tutoriel Performance

API BCE en Python : tutoriel complet avec FastAPI, async/await et code à copier

Intégrer l'API BCE en Python en moins d'une heure. Tutoriel complet : setup, requêtes httpx async, wrapper FastAPI, gestion d'erreurs, retry exponentiel, cache local. Tout le code est copiable directement.

6 mai 20269 min de lecture

En bref

Ce tutoriel vous montre comment intégrer l'API BCE/KBO de Company Belgium en Python en moins d'une heure, avec du code prêt à copier-coller. Vous construisez un client async httpx, un retry exponentiel pour les 429 et erreurs 5xx, un cache TTL en mémoire, et un wrapper FastAPI qui expose vos propres routes métier. Le résultat tient en 100 lignes et est directement déployable en production.

Le résultat à la fin de l'article

À la fin de ce tuto, vous aurez :

Un client Python async pour l'API BCE/KBO de Company Belgium

Un wrapper FastAPI qui expose une route /companies/{number} proxy

Une stratégie de retry exponentiel pour les erreurs réseau

Un cache local en mémoire pour réduire les appels

Une gestion d'erreurs propre (404, 429, 5xx)

Tout le code est testé et copiable directement dans votre projet.

1. Prérequis

Python 3.11+ (pour le support natif d'async)

Une clé API Company Belgium (plan gratuit suffisant pour suivre le tuto)

Un terminal et un éditeur

2. Setup

Terminal

1
2
3

mkdir bce-python-tuto && cd bce-python-tuto
python -m venv .venv && source .venv/bin/activate
pip install httpx fastapi uvicorn python-dotenv

Créez un fichier .env :

Terminal

1
2
3

COMPANY_BELGIUM_API_KEY=pk_live_xxxxxxxxxxxx
COMPANY_BELGIUM_API_SECRET=sk_live_xxxxxxxxxxxx
COMPANY_BELGIUM_BASE_URL=https://api.companybelgium.be/v1

3. Le client async minimal

Créez bce_client.py :

Python

import os
import asyncio
import httpx
from dotenv import load_dotenv

load_dotenv()

class BCEClient:
    def __init__(self, base_url=None, api_key=None, api_secret=None, timeout=10.0):
        self.base_url = base_url or os.getenv("COMPANY_BELGIUM_BASE_URL")
        self.api_key = api_key or os.getenv("COMPANY_BELGIUM_API_KEY")
        self.api_secret = api_secret or os.getenv("COMPANY_BELGIUM_API_SECRET")
        self.timeout = timeout
        self._client = httpx.AsyncClient(
            base_url=self.base_url,
            headers={
                "X-API-Key": self.api_key,
                "X-API-Secret": self.api_secret,
            },
            timeout=timeout,
        )

    async def get_company(self, enterprise_number: str) -> dict:
        clean = enterprise_number.replace(".", "").replace(" ", "")
        r = await self._client.get(class="code-string">f"/companies/{clean}")
        r.raise_for_status()
        return r.json()["data"]

    async def search(self, query: str, limit: int = 20) -> list[dict]:
        r = await self._client.get("/companies/search", params={"q": query, "limit": limit})
        r.raise_for_status()
        return r.json()["data"]

    async def close(self):
        await self._client.aclose()


async def demo():
    client = BCEClient()
    try:
        company = await client.get_company("0200.065.765")
        print(class="code-string">f"Nom: {company[class="code-string">'name']}")
        print(class="code-string">f"Forme: {company[class="code-string">'legalForm']}")
        print(class="code-string">f"Adresse: {company[class="code-string">'address'][class="code-string">'fullAddress']}")
    finally:
        await client.close()


if __name__ == "__main__":
    asyncio.run(demo())

Exécutez :

Terminal

python bce_client.py

Vous obtenez la fiche d'une entreprise belge en JSON, en une requête.

4. Retry exponentiel pour les erreurs transitoires

L'API peut renvoyer 429 (rate limit) ou 5xx (panne ponctuelle). Au lieu d'échouer, on retry avec backoff.

Ajoutez retry.py :

Python

import asyncio
import httpx
from typing import Callable, TypeVar, Awaitable

T = TypeVar("T")

async def with_retry(
    fn: Callable[[], Awaitable[T]],
    max_attempts: int = 4,
    base_delay: float = 0.5,
) -> T:
    last_exc = None
    for attempt in range(max_attempts):
        try:
            return await fn()
        except httpx.HTTPStatusError as e:
            status = e.response.status_code
            if status in (429, 500, 502, 503, 504) and attempt < max_attempts - 1:
                delay = base_delay * (2 ** attempt)
                retry_after = e.response.headers.get("Retry-After")
                if retry_after and retry_after.isdigit():
                    delay = max(delay, float(retry_after))
                await asyncio.sleep(delay)
                last_exc = e
                continue
            raise
        except (httpx.TimeoutException, httpx.NetworkError) as e:
            if attempt < max_attempts - 1:
                await asyncio.sleep(base_delay * (2 ** attempt))
                last_exc = e
                continue
            raise
    if last_exc:
        raise last_exc
    raise RuntimeError("retry exhausted")

Usage dans le client :

Python

from retry import with_retry

async def get_company(self, enterprise_number: str) -> dict:
    clean = enterprise_number.replace(".", "").replace(" ", "")
    return await with_retry(lambda: self._get_company_once(clean))

async def _get_company_once(self, num: str) -> dict:
    r = await self._client.get(class="code-string">f"/companies/{num}")
    r.raise_for_status()
    return r.json()["data"]

5. Cache local en mémoire

Pour éviter de re-payer la même requête en boucle, ajoutez un cache TTL :

Python

import time
from typing import Any

class TTLCache:
    def __init__(self, ttl_seconds: int = 300):
        self.ttl = ttl_seconds
        self._store: dict[str, tuple[float, Any]] = {}

    def get(self, key: str):
        item = self._store.get(key)
        if not item:
            return None
        expires_at, value = item
        if time.time() > expires_at:
            del self._store[key]
            return None
        return value

    def set(self, key: str, value: Any):
        self._store[key] = (time.time() + self.ttl, value)

Intégrez-le au client :

Python

class BCEClient:
    def __init__(self, ..., cache_ttl: int = 300):
        ...
        self.cache = TTLCache(ttl_seconds=cache_ttl)

    async def get_company(self, enterprise_number: str) -> dict:
        clean = enterprise_number.replace(".", "").replace(" ", "")
        cached = self.cache.get(class="code-string">f"company:{clean}")
        if cached:
            return cached
        data = await with_retry(lambda: self._get_company_once(clean))
        self.cache.set(class="code-string">f"company:{clean}", data)
        return data

Pour un cache partagé entre processus (production), basculez sur memcached, ou utilisez la couche de cache HTTP via les en-têtes Cache-Control. Le principe reste le même.

6. Wrapper FastAPI

Exposez votre propre API qui consomme l'API BCE et ajoute votre logique métier.

Créez main.py :

Python

from fastapi import FastAPI, HTTPException
from contextlib import asynccontextmanager
import httpx
from bce_client import BCEClient

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.bce = BCEClient()
    try:
        yield
    finally:
        await app.state.bce.close()

app = FastAPI(lifespan=lifespan)

@app.get("/companies/{number}")
async def get_company(number: str):
    try:
        return await app.state.bce.get_company(number)
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            raise HTTPException(404, "Company not found")
        raise HTTPException(502, class="code-string">f"Upstream error: {e.response.status_code}")

@app.get("/search")
async def search(q: str, limit: int = 20):
    try:
        return await app.state.bce.search(q, limit)
    except httpx.HTTPStatusError as e:
        raise HTTPException(502, class="code-string">f"Upstream error: {e.response.status_code}")

Lancez :

Terminal

uvicorn main:app --reload --port 8000

Testez :

Terminal

1
2

curl http://localhost:8000/companies/0200.065.765
curl "http://localhost:8000/search?q=espero"

7. Cas d'usage typiques

A. Vérifier un fournisseur avant de payer

Python

async def is_supplier_valid(vat_number: str) -> bool:
    client = BCEClient()
    try:
        company = await client.get_company(vat_number)
        return company["status"] == "AC"  class="code-comment"># AC = Actif
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return False
        raise
    finally:
        await client.close()

B. Pré-remplir un formulaire client

Python

async def prefill_client_form(vat_number: str) -> dict:
    client = BCEClient()
    try:
        c = await client.get_company(vat_number)
        return {
            "name": c["name"],
            "legalForm": c["legalForm"],
            "address": c["address"]["fullAddress"],
            "nace_main": c["mainActivity"]["naceCode"],
        }
    finally:
        await client.close()

C. Recherche en masse (avec sémaphore pour respecter le rate limit)

Python

import asyncio

async def verify_many(vat_numbers: list[str]) -> dict[str, bool]:
    client = BCEClient()
    sem = asyncio.Semaphore(5)  class="code-comment"># max 5 requêtes simultanées

    async def check(n: str):
        async with sem:
            try:
                company = await client.get_company(n)
                return n, company["status"] == "AC"
            except Exception:
                return n, False

    try:
        results = await asyncio.gather(*[check(n) for n in vat_numbers])
        return dict(results)
    finally:
        await client.close()

8. Erreurs à éviter

Ouvrir un client HTTP par requête — coûteux, casse les connexions keep-alive. Réutilisez un client.

Ne pas gérer 429 — vous serez throttlé rapidement. Implémentez le retry.

Ne pas cacher — vous paierez 10× le même appel pour la même PME.

Pas de timeout — vos workers se bloquent. Mettez systématiquement un timeout.

Logger la clé API — utilisez os.getenv, ne hardcodez jamais. Voir les bonnes pratiques de sécurité.

Comment Company Belgium se branche dans une stack Python

L'API Company Belgium est REST/JSON et compatible avec n'importe quel client HTTP Python (requests, httpx, aiohttp). Aucun SDK requis pour démarrer, mais :

Doc OpenAPI disponible — vous pouvez générer un client typé avec openapi-python-client

Webhooks : recevez les changements sur votre endpoint FastAPI, plutôt que de poller

Rate limit headers : X-RateLimit-Remaining, Retry-After — votre code de retry les exploite directement

SDK officiel Python disponible sur les plans pro (avec cache, retry et types intégrés)

Pour aller plus loin : voir le comparatif des API BCE, la référence des endpoints et l'architecture rate limits / caching.

En résumé

L'intégration de l'API BCE en Python tient en 100 lignes :

Client async httpx avec auth header

Retry exponentiel sur les codes transitoires

Cache TTL en mémoire

Wrapper FastAPI pour exposer ses propres routes métier

Sémaphore pour les opérations en masse

Le code de ce tutoriel est immédiatement productionnable. Adaptez les TTL et les concurrences en fonction de votre plan, et vous avez une intégration robuste prête à l'emploi.

Questions fréquentes

Comment integrer l'API BCE/KBO en Python avec httpx et async/await ?

Créez une classe BCEClient qui instancie un httpx.AsyncClient avec les headers X-API-Key et X-API-Secret de votre compte Company Belgium. Les méthodes get_company et search utilisent des appels await pour effectuer des requêtes non bloquantes. Tout le code tient en environ 40 lignes et s'intègre naturellement dans une stack FastAPI ou Django async. Les clés API sont chargées via os.getenv pour ne jamais les hardcoder.

Pourquoi utiliser httpx plutot que requests pour consommer l'API BCE en Python ?

httpx supporte nativement l'async/await, ce qui est essentiel pour des applications à forte concurrence comme les apps FastAPI ou les scripts de vérification de masse. Avec requests (synchrone), chaque appel API bloque le thread ; avec httpx async, vous pouvez traiter des centaines de requêtes en parallèle avec un sémaphore. httpx est aussi compatible avec les contextes de test async et génère des clients réutilisables avec connexions HTTP keep-alive, ce qui réduit la latence.

Comment implémenter un retry exponentiel pour les erreurs 429 de l'API KBO/BCE en Python ?

Créez une fonction with_retry qui encapsule votre appel API dans une boucle. En cas de status 429 ou 5xx, attendez base_delay * 2^attempt secondes avant de réessayer, en honorant le header Retry-After si présent. Ajoutez un jitter aléatoire pour éviter le thundering herd si plusieurs instances tournent en parallèle. Après 4 tentatives sans succès, levez une exception. Ce pattern est identique en Python et en Node.js.

Peut-on utiliser FastAPI comme proxy de l'API BCE/KBO pour une application belge ?

Oui, c'est une architecture courante. Créez une application FastAPI qui instancie votre BCEClient au démarrage via un context manager lifespan, puis exposez des routes comme GET /companies/{number} et GET /search qui délèguent les appels au BCEClient. Ce proxy vous permet d'ajouter votre propre authentification, de la logique métier, du caching supplémentaire et des logs personnalisés avant d'appeler l'API officielle. La latence ajoutée est inférieure à 5 ms en local.