Python

Python Examples

Complete integration examples using Python with the requests library.

Setup

pip install requests

API Client Class

import requests
from typing import Optional
from dataclasses import dataclass, asdict


BASE_URL = "https://backend.joinnentropy.com/api/v1/integration"


class EUDRClient:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        })

    def _request(self, method: str, path: str, json=None, params=None):
        response = self.session.request(
            method,
            f"{BASE_URL}{path}",
            json=json,
            params=params,
        )
        response.raise_for_status()
        data = response.json()
        if not data.get("success"):
            raise Exception(f"API Error: {data.get('error', {}).get('message')}")
        return data["data"]

    # Suppliers
    def upsert_suppliers(self, suppliers: list[dict]):
        return self._request("POST", "/suppliers", json={"suppliers": suppliers})

    def get_suppliers(self, page: int = 1, limit: int = 50):
        return self._request("GET", "/suppliers", params={"page": page, "limit": limit})

    # Products
    def upsert_products(self, products: list[dict]):
        return self._request("POST", "/products", json={"products": products})

    def get_products(self, page: int = 1, limit: int = 50):
        return self._request("GET", "/products", params={"page": page, "limit": limit})

    # Batches
    def upsert_batches(self, batches: list[dict]):
        return self._request("POST", "/batches", json={"batches": batches})

    def get_batches(self, page: int = 1, limit: int = 50):
        return self._request("GET", "/batches", params={"page": page, "limit": limit})

    # Sync status
    def get_sync_status(self, entity_type: Optional[str] = None):
        params = {"entityType": entity_type} if entity_type else {}
        return self._request("GET", "/sync-status", params=params)

    # DDS status
    def get_dds_status(self, batch_external_id: str):
        return self._request("GET", f"/dds/{batch_external_id}")

    # Field mappings
    def get_field_mappings(self):
        return self._request("GET", "/field-mappings")

Usage Examples

Sync Suppliers

client = EUDRClient("eudr_live_abc123...")

# From your database or ERP
suppliers = [
    {
        "externalId": "SUP-001",
        "name": "Amazon Timber Co.",
        "country": "BR",
        "address": "123 Forest Road",
        "city": "Manaus",
        "email": "contact@amazontimber.co",
        "taxId": "12.345.678/0001-90",
    },
    {
        "externalId": "SUP-002",
        "name": "Southeast Asia Palm Ltd.",
        "country": "ID",
        "address": "456 Palm Avenue",
        "city": "Jakarta",
        "email": "info@seapalm.id",
    },
]

result = client.upsert_suppliers(suppliers)
print(f"Created: {result['created']}, Updated: {result['updated']}")

Sync Products with Supplier Links

products = [
    {
        "externalId": "PROD-001",
        "name": "Brazilian Eucalyptus Timber",
        "hsCode": "4407.11",
        "description": "Sustainably sourced eucalyptus planks",
        "supplierExternalId": "SUP-001",
    },
    {
        "externalId": "PROD-002",
        "name": "Indonesian Palm Oil - Crude",
        "hsCode": "1511.10",
        "supplierExternalId": "SUP-002",
    },
]

result = client.upsert_products(products)
print(f"Created: {result['created']}, Updated: {result['updated']}")

Create Batches

batches = [
    {
        "externalId": "BATCH-2024-001",
        "productExternalId": "PROD-001",
        "quantity": 5000,
        "unit": "kg",
        "batchDate": "2024-06-15",
    },
]

result = client.upsert_batches(batches)
print(result)

Check Sync Status

# Overview
status = client.get_sync_status()
print(status)

# For specific entity type
supplier_status = client.get_sync_status("SUPPLIER")
for record in supplier_status["records"]:
    print(f"{record['externalId']}: {record['syncStatus']}")

Get DDS Status for a Batch

dds = client.get_dds_status("BATCH-2024-001")
print(f"Batch: {dds['batch']['externalId']}")
print(f"Product: {dds['product']['name']}")
print(f"Supplier: {dds['supplier']['name']}")
for submission in dds.get("ddsSubmissions", []):
    print(f"DDS: {submission['status']} at {submission['submittedAt']}")

Webhook Verification

import hmac
import hashlib
from flask import Flask, request, jsonify

app = Flask(__name__)
WEBHOOK_SECRET = "your_webhook_secret"


def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode("utf-8"),
        payload,
        hashlib.sha256,
    ).hexdigest()
    sig = signature.removeprefix("sha256=")
    return hmac.compare_digest(sig, expected)


@app.route("/webhooks/eudr", methods=["POST"])
def handle_webhook():
    signature = request.headers.get("X-Webhook-Signature", "")

    if not verify_signature(request.data, signature, WEBHOOK_SECRET):
        return jsonify({"error": "Invalid signature"}), 401

    event = request.json
    event_type = event["event"]

    if event_type == "supplier.created":
        print(f"New supplier: {event['data']}")
    elif event_type == "sync.completed":
        print(f"Sync done: {event['data']}")
    elif event_type == "dds.status_changed":
        print(f"DDS update: {event['data']}")

    return jsonify({"received": True})


if __name__ == "__main__":
    app.run(port=5000)

Pagination Helper

def fetch_all_suppliers(client: EUDRClient) -> list[dict]:
    """Fetch all suppliers across all pages."""
    all_suppliers = []
    page = 1

    while True:
        result = client.get_suppliers(page=page, limit=100)
        suppliers = result["records"]
        all_suppliers.extend(suppliers)

        if len(suppliers) < 100:
            break
        page += 1

    return all_suppliers


# Usage
all_suppliers = fetch_all_suppliers(client)
print(f"Total suppliers: {len(all_suppliers)}")

Error Handling with Retry

import time
from requests.exceptions import HTTPError


def sync_with_retry(client: EUDRClient, suppliers: list[dict], max_retries: int = 3):
    for attempt in range(1, max_retries + 1):
        try:
            return client.upsert_suppliers(suppliers)
        except HTTPError as e:
            if e.response.status_code == 429:
                wait = 2 ** attempt
                print(f"Rate limited, retrying in {wait}s...")
                time.sleep(wait)
                continue
            raise
    raise Exception("Max retries exceeded")

Django Integration

views.py

import json
import hmac
import hashlib
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from django.conf import settings


@csrf_exempt
@require_POST
def eudr_webhook(request):
    signature = request.headers.get("X-Webhook-Signature", "")
    expected = hmac.new(
        settings.EUDR_WEBHOOK_SECRET.encode(),
        request.body,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(signature.removeprefix("sha256="), expected):
        return JsonResponse({"error": "Invalid signature"}, status=401)

    event = json.loads(request.body)
    # Process event...

    return JsonResponse({"received": True})