SH

Sascha Hess

xenosystems.de - IT-Consulting & Data Management

www.xenosystems.de

NOTFALL-KIT – SYSTEMINTEGRATION 2026

ERP-Migration

Schritt-für-Schritt

API, Middleware und Event-Driven Architecture — produktionsreif, sicher und skalierbar

1

Integrationsarchitektur

Entscheidungsmatrix: Point-to-Point, ESB, Event-driven — wann was

2

REST-API Design-Guide

Naming, Versionierung, OpenAPI-Spec — sofort einsetzbare Vorlage

3

Middleware & Message Broker

Kafka vs. RabbitMQ vs. Azure Service Bus — vollstaendiger Vergleich

4

Sicherheits-Framework

OAuth2, API-Gateway, Secrets-Management — produktionsreif abgesichert

5

45-Tage-Integrationsplan

Von der ersten API bis zum produktiven Middleware-Betrieb

WAS SIE IN DIESEM KIT ERWARTEN DÜRFEN

■  Integrationsarchitektur-Entscheidungsmatrix — Point-to-Point, Hub-and-Spoke, Event-driven: wann was.

■  REST-API-Design-Guide — Naming, Versionierung, Error Codes, OpenAPI-Spec-Vorlage.

■  Middleware-Vergleich — Kafka vs. RabbitMQ vs. Azure Service Bus vs. MuleSoft.

■  Sicherheits-Framework — OAuth2-Flow, API-Gateway-Konfiguration, mTLS-Setup.

■  Praxis-Szenarien — ERP↔CRM, ERP↔Webshop, ERP↔EDI vollständig dokumentiert.

INTEGRATIONSSTIL-ENTSCHEIDUNGSMATRIX:

POINT-TO-POINT (direkte Systemverbindung):

  Wann geeignet:  2-3 Systeme, einfache Datenflüsse, niedriges Volumen

  Vorteile:       Einfach, schnell implementiert, kein Middleware-Overhead

  Nachteile:      Skaliert nicht (n*(n-1)/2 Verbindungen bei n Systemen)

  Risiko:         Enge Kopplung — Aenderung an System A bricht System B

  Typisch:        Kleines Unternehmen, ERP → ein CRM, kein weiteres Wachstum

HUB-AND-SPOKE / ESB (zentrale Middleware):

  Wann geeignet:  5-20 Systeme, komplexe Transformationen, hohe Zuverlaessigkeit

  Vorteile:       Zentrale Governance, Monitoring, Transformation an einem Ort

  Nachteile:      Middleware ist Single Point of Failure (wenn nicht HA)

  Risiko:         Overhead-Kosten, Spezialkenntnisse fuer ESB-Plattform noetig

  Typisch:        Mittelstand mit SAP/Dynamics + CRM + WMS + EDI

EVENT-DRIVEN (Message Broker / Event Streaming):

  Wann geeignet:  Hohes Volumen (>10.000 Events/Tag), lose Kopplung gewuenscht

  Vorteile:       Maximale Entkopplung, Skalierbarkeit, Resilienz

  Nachteile:      Komplexitaet, Eventual Consistency, schwerer zu debuggen

  Risiko:         Broker-Infrastruktur braucht eigenes Ops-Know-how

  Typisch:        E-Commerce, Logistik, Event-getriebene Microservices

API-GATEWAY / MESH (API-first):

  Wann geeignet:  Viele externe Partner, SaaS-Integrationen, Cloud-native

  Vorteile:       Einheitlicher Eintrittspunkt, Sicherheit zentralisiert

  Nachteile:      Gateway wird kritische Infrastruktur, Kosten

  Typisch:        Unternehmen mit Partnernetz, Marketplace-Modelle

FRAGE 1: Wie viele Systeme muessen integriert werden?

  <= 3 Systeme → Point-to-Point ist ausreichend

  4-10 Systeme → Hub-and-Spoke / Middleware pruefen

  > 10 Systeme → Middleware oder API-Gateway zwingend

FRAGE 2: Wie zeitkritisch sind die Daten?

  Echtzeit (< 1 Sekunde) → REST-API (synchron) oder Event-Streaming

  Near-Realtime (< 1 Minute) → Message Broker (async)

  Batch (Stuendlich/Taeglich) → Datei-basiert oder Batch-Job

FRAGE 3: Was passiert wenn die Schnittstelle ausfaellt?

  Benutzer merkt es sofort, Prozess stoppt → Synchron + Circuit Breaker

  Daten werden verzoegert, Prozess laeuft weiter → Async + Queue

  Datenverlust inakzeptabel → Message Broker mit Persistenz

FRAGE 4: Wie oft aendern sich die Systeme?

  Haeufige Aenderungen → Lose Kopplung (Event-driven, API-Gateway)

  Stabile Systeme → Enge Kopplung akzeptabel (Point-to-Point)

ERP → CRM

Push

Auftraege, Rechnungen

Near-Realtime

Webhook / Message Queue

CRM → ERP

Push

Neukunden, Angebote

Near-Realtime

REST-API (synchron)

ERP → WMS

Push

Lieferauftraege

Echtzeit

REST-API oder Queue

WMS → ERP

Push

Buchungsbelege

Near-Realtime

Queue + Batch-Commit

ERP → Webshop

Push + Pull

Bestaende, Preise

Stuendlich

REST-API Polling

Webshop → ERP

Push

Bestellungen

Echtzeit

Webhook → Queue

ERP → EDI-Partner

Push

ORDERS, DESADV

Trigger-basiert

AS2 / SFTP

ERP → BI/DWH

Push

Alle Bewegungsdaten

Taeglich

CDC oder Datei-Export

ERP → Banking

Push

SEPA-Dateien

1x taeglich

SFTP + EBICS

REST-DESIGN-GRUNDREGELN:

RESSOURCEN-NAMING:

  Korrekt:   /api/v1/customers          → Sammlung (Plural, Kleinbuchstaben)

  Korrekt:   /api/v1/customers/42       → Einzelne Ressource (ID im Pfad)

  Korrekt:   /api/v1/customers/42/orders → Unterressource

  Falsch:    /api/v1/getCustomer        → Kein Verb im Pfad

  Falsch:    /api/v1/CustomerList       → Kein CamelCase, kein Singular

HTTP-METHODEN KORREKT EINSETZEN:

  GET    → Lesen, niemals Zustand aendern, idempotent

  POST   → Neue Ressource erstellen, nicht idempotent

  PUT    → Ressource vollstaendig ersetzen, idempotent

  PATCH  → Ressource partiell aktualisieren

  DELETE → Ressource loeschen, idempotent

HTTP-STATUS-CODES:

  200 OK             → Erfolgreiche GET/PUT/PATCH

  201 Created        → Erfolgreiche POST (+ Location-Header)

  204 No Content     → Erfolgreiche DELETE

  400 Bad Request    → Ungueltige Anfrage (Validierungsfehler)

  401 Unauthorized   → Nicht authentifiziert

  403 Forbidden      → Authentifiziert, aber keine Berechtigung

  404 Not Found      → Ressource nicht gefunden

  409 Conflict       → Ressource existiert bereits / Konflikt

  429 Too Many Req.  → Rate Limit ueberschritten

  500 Internal Error → Serverfehler (Stack Trace NIEMALS zurueckgeben!)

openapi: 3.0.3

info:

  title: ERP Integration API

  description: |

    Schnittstelle fuer die ERP-Systemintegration.

    Authentifizierung via OAuth2 Client Credentials.

  version: 1.0.0

  contact:

    name: IT-Integration Team

    email: integration@firma.de

servers:

  - url: https://api.firma.de/v1

    description: Produktion

  - url: https://api-test.firma.de/v1

    description: Test/QAS

paths:

  /customers:

    get:

      summary: Kundenliste abrufen

      operationId: listCustomers

      parameters:

        - name: page

          in: query

          schema: { type: integer, default: 1 }

        - name: pageSize

          in: query

          schema: { type: integer, default: 50, maximum: 500 }

        - name: modifiedSince

          in: query

          description: Nur Kunden geaendert seit (ISO 8601)

          schema: { type: string, format: date-time }

      responses:

        '200':

          description: Erfolgreich

        '401':

          description: Nicht authentifiziert

        '429':

          description: Rate Limit ueberschritten

components:

  schemas:

    Customer:

      type: object

      required: [id, name, email]

      properties:

        id:         { type: string, example: "CUST-00042" }

        name:       { type: string, maxLength: 100 }

        email:      { type: string, format: email }

        createdAt:  { type: string, format: date-time, readOnly: true }

        modifiedAt: { type: string, format: date-time, readOnly: true }

  securitySchemes:

    oauth2:

      type: oauth2

      flows:

        clientCredentials:

          tokenUrl: https://auth.firma.de/oauth2/token

          scopes:

            read:  Lesezugriff

            write: Schreibzugriff

VERSIONIERUNGSOPTIONEN IM VERGLEICH:

URI-VERSIONING (empfohlen fuer Enterprise):

  URL:         /api/v1/customers, /api/v2/customers

  Vorteile:    Sichtbar, cacheable, einfach zu testen

  Nachteil:    Redundante URLs, Routing-Aufwand

  Empfehlung:  Standard fuer interne und Partner-APIs

HEADER-VERSIONING:

  Header:      Accept: application/vnd.firma.v2+json

  Vorteil:     URLs bleiben sauber

  Nachteil:    Schwerer zu testen, nicht cacheable

  Empfehlung:  Fuer oeffentliche APIs mit vielen Consumern

DEPRECATION-PROZESS:

  → Version V1 wird als deprecated markiert (Deprecation-Header)

  → Mindestens 6 Monate Uebergangszeit fuer Consumer

  → Sunset-Date im Header kommunizieren

  → Monitoring: Wer ruft V1 noch auf? → aktiv informieren

  → Erst wenn alle Consumer migriert: V1 abschalten

{

  "data": [

    { "id": "CUST-001", "name": "Mueller GmbH" },

    { "id": "CUST-002", "name": "Schneider AG" }

  ],

  "pagination": {

    "page": 1,

    "pageSize": 50,

    "totalItems": 1247,

    "totalPages": 25,

    "hasNextPage": true,

    "nextPageUrl": "/api/v1/customers?page=2&pageSize=50"

  },

  "meta": {

    "requestId": "req-a1b2c3d4",

    "timestamp": "2026-03-15T09:00:00Z",

    "processingTimeMs": 42

  }

}

Durchsatz

Sehr hoch (Mio/s)

Hoch (100k/s)

Hoch

Hoch

Mittel

Persistenz

Tage-Wochen

Konfigurierbar

14 Tage

14 Tage

Konfigurierbar

Routing

Topic-based

Flexible Exchange

Topic/Queue

Topic/Queue

Komplex

Komplexitaet

Hoch

Mittel

Niedrig

Niedrig

Sehr hoch

Kosten

OSS (Infra)

OSS (Infra)

Pay-per-use

Pay-per-use

Enterprise-Lizenz

Replay

Ja (Key-Feature)

Nein

Nein (begrenzt)

Nein

Nein

Geeignet fuer

Event Streaming

Task Queues

Azure-Umgebung

AWS-Umgebung

Enterprise ESB

# RabbitMQ-Topologie fuer ERP-Integration

# Exchange → Routing Key → Queue → Consumer

exchanges:

  - name: erp.events          # Topic Exchange fuer alle ERP-Events

    type: topic

    durable: true

  - name: erp.dlx             # Dead-Letter Exchange fuer Fehler

    type: direct

    durable: true

queues:

  # ERP → CRM: Kundenereignisse

  - name: crm.customer.events

    durable: true

    arguments:

      x-dead-letter-exchange: erp.dlx

      x-dead-letter-routing-key: crm.customer.dead

      x-message-ttl: 86400000        # 24h TTL

      x-max-length: 100000           # Max 100k Nachrichten

  # ERP → WMS: Lieferauftraege

  - name: wms.delivery.orders

    durable: true

    arguments:

      x-dead-letter-exchange: erp.dlx

      x-message-ttl: 3600000         # 1h TTL

  # Dead-Letter Queue (Fehler-Nachrichten zur Analyse)

  - name: erp.dead.letters

    durable: true

bindings:

  - exchange: erp.events

    routing_key: customer.*          # customer.created, customer.updated

    queue: crm.customer.events

  - exchange: erp.events

    routing_key: delivery.order.*

    queue: wms.delivery.orders

  - exchange: erp.dlx

    routing_key: crm.customer.dead

    queue: erp.dead.letters

KAFKA TOPIC-DESIGN PRINZIPIEN:

NAMING-KONVENTION: {domain}.{entity}.{event-type}

  Beispiele:

    erp.customer.created

    erp.customer.updated

    erp.order.placed

    erp.invoice.issued

    wms.shipment.dispatched

PARTITIONIERUNG:

  Faustregel: 1 Partition = 1 Consumer-Thread

  Fuer ERP-Integration: 3-12 Partitionen je Topic

  Partition-Key: Kunden-ID oder Auftrags-ID (fuer geordnete Events)

  ACHTUNG: Partitionsanzahl kann nicht reduziert werden!

RETENTION:

  Geschaeftsdaten:  7-30 Tage (je nach Regulatorik)

  Audit-Events:    90-365 Tage

  Echtzeit-Events: 1-3 Tage

CONSUMER-GROUPS:

  Jedes Zielsystem = eigene Consumer-Group

  Beispiel Topic erp.order.placed:

    → consumer-group: crm-system     (liest fuer CRM)

    → consumer-group: wms-system     (liest fuer WMS)

    → consumer-group: bi-dwh         (liest fuer BI)

  Alle drei lesen unabhaengig — kein Event geht verloren

# Kafka Producer Konfiguration (empfohlen fuer ERP):

acks=all                   # Alle Replikas bestaetigen (kein Datenverlust)

retries=3                  # 3 Wiederholungen bei transientem Fehler

enable.idempotence=true    # Exakt-einmal Semantik

max.in.flight.requests=1   # Reihenfolge garantieren

compression.type=lz4       # Komprimierung fuer hohen Durchsatz

AUTHENTIFIZIERUNGSVERFAHREN — WANN WAS:

API-KEY (einfachste Methode):

  Wann:      Interne Systeme, niedrige Sicherheitsanforderungen

  Vorteil:   Einfach zu implementieren, kein Token-Refresh

  Nachteil:  Kein Ablaufdatum, schwer zu rotieren, kein Scope

  Sicher:    Key im Header (nicht in URL!), regelmaessige Rotation

  Header:    X-API-Key: sk_live_abc123...

OAUTH2 CLIENT CREDENTIALS (empfohlen fuer M2M):

  Wann:      System-zu-System, kein Benutzer beteiligt (M2M)

  Vorteil:   Token laeuft ab, Scopes, Revocation moeglich

  Ablauf:    Client → POST /oauth2/token → Access Token (JWT)

             Client → API-Aufruf mit Bearer Token

  Scope:     Fein granulare Berechtigungen (read:orders, write:orders)

OAUTH2 + PKCE (fuer Benutzer-delegierte Zugriffe):

  Wann:      Benutzer autorisiert App Zugriff auf seine Daten

  Vorteil:   Sicher auch fuer public Clients (SPAs, Mobile)

  Typisch:   CRM-Integration mit Benutzer-Kontext

MUTUAL TLS (mTLS):

  Wann:      Hoechste Sicherheitsanforderungen (Banking, Healthcare)

  Vorteil:   Client UND Server authentifizieren sich gegenseitig

  Nachteil:  Zertifikatsverwaltung aufwendig

  Typisch:   EBICS-Banking, B2B mit regulierten Partnern

import requests

import time

from dataclasses import dataclass

from typing import Optional

@dataclass

class TokenCache:

    access_token: str

    expires_at: float

class OAuth2Client:

    # OAuth2 Client Credentials Flow mit automatischem Token-Caching.

    # Token wird 60 Sekunden vor Ablauf erneuert.

    def __init__(self, token_url: str, client_id: str,

                 client_secret: str, scope: str = ""):

        self.token_url = token_url

        self.client_id = client_id

        self.client_secret = client_secret

        self.scope = scope

        self._cache: Optional[TokenCache] = None

    def get_token(self) -> str:

        if self._cache and time.time() < self._cache.expires_at - 60:

            return self._cache.access_token

        return self._fetch_token()

    def _fetch_token(self) -> str:

        response = requests.post(

            self.token_url,

            data={

                "grant_type": "client_credentials",

                "client_id": self.client_id,

                "client_secret": self.client_secret,

                "scope": self.scope,

            },

            timeout=10,

        )

        response.raise_for_status()

        data = response.json()

        expires_in = data.get("expires_in", 3600)

        self._cache = TokenCache(

            access_token=data["access_token"],

            expires_at=time.time() + expires_in,

        )

        return self._cache.access_token

    def get_headers(self) -> dict:

        return {

            "Authorization": f"Bearer {self.get_token()}",

            "Content-Type": "application/json",

        }

# Verwendung:

# import os

# client = OAuth2Client(

#     token_url="https://auth.firma.de/oauth2/token",

#     client_id=os.getenv("API_CLIENT_ID"),       # nie hardcoden!

#     client_secret=os.getenv("API_CLIENT_SECRET"),

#     scope="read:orders write:orders"

# )

# response = requests.get(url, headers=client.get_headers())

# Beispiel: Kong Gateway Sicherheits-Plugins (vereinfacht)

plugins:

  # JWT-Validierung am Gateway

  - name: jwt

    service: erp-api

    config:

      claims_to_verify: ["exp"]    # Ablaufzeit pruefen

  # Rate Limiting gegen Missbrauch

  - name: rate-limiting

    service: erp-api

    config:

      minute: 1000          # Max 1000 Requests/Minute

      hour: 50000           # Max 50.000 Requests/Stunde

      policy: redis         # Verteiltes Rate Limiting

  # Request-Groesse begrenzen (Schutz vor DoS)

  - name: request-size-limiting

    service: erp-api

    config:

      allowed_payload_size: 10    # Max 10 MB

  # Audit-Logging aller API-Aufrufe

  - name: http-log

    service: erp-api

    config:

      http_endpoint: http://audit-log-service:9000/logs

NIEMALS DIREKT IM CODE

■  API-Keys, Passwoerter, Client-Secrets NIEMALS in Quellcode oder Konfigurationsdateien.

■  NIEMALS in Git einchecken — auch nicht in privaten Repositories.

■  Stattdessen: Azure Key Vault, HashiCorp Vault, AWS Secrets Manager oder Kubernetes Secrets.

■  Rotation: automatisch alle 90 Tage, ohne Anwendungs-Neustart — Vault-Integration verwenden.

FEHLERKLASSEN UND RICHTIGE REAKTION:

TRANSIENTE FEHLER (voruebergehend — immer wiederholen):

  HTTP 429 Too Many Requests  → Retry mit Backoff (Rate Limit respektieren)

  HTTP 503 Service Unavailable → Retry mit Backoff

  TCP Timeout / Connection Reset → Retry mit Backoff

  DNS temporaer nicht aufluesbar → Retry nach kurzer Pause

PERMANENTE FEHLER (nicht wiederholen — sofort abbrechen):

  HTTP 400 Bad Request        → Bug im Aufrufer, kein Retry

  HTTP 401 Unauthorized       → Token erneuern, dann neu versuchen

  HTTP 403 Forbidden          → Keine Berechtigung, kein Retry

  HTTP 404 Not Found          → Ressource existiert nicht, kein Retry

RETRY-STRATEGIE (Exponential Backoff mit Jitter):

  Versuch 1: sofort

  Versuch 2: 1 Sekunde warten

  Versuch 3: 2 Sekunden warten

  Versuch 4: 4 Sekunden warten

  Versuch 5: 8 Sekunden warten + zufaelliger Jitter (0-2 Sek)

  Max: 5 Versuche, dann → Dead-Letter-Queue

JITTER ist entscheidend: Ohne Jitter versuchen alle ausgefallenen

Consumer gleichzeitig den Neustart — und ueberlasten das Zielsystem.

import time

from enum import Enum

class CircuitState(Enum):

    CLOSED    = "closed"    # Normal: Anfragen durchgelassen

    OPEN      = "open"      # Gesperrt: Zielsystem ausgefallen

    HALF_OPEN = "half_open" # Test: Eine Anfrage zur Pruefung

class CircuitBreaker:

    def __init__(self, failure_threshold=5,

                 timeout=60.0, success_threshold=2):

        self.failure_threshold = failure_threshold

        self.timeout = timeout

        self.success_threshold = success_threshold

        self.failure_count = 0

        self.success_count = 0

        self.last_failure_time = 0

        self.state = CircuitState.CLOSED

    def call(self, func, *args, **kwargs):

        if self.state == CircuitState.OPEN:

            elapsed = time.time() - self.last_failure_time

            if elapsed > self.timeout:

                self.state = CircuitState.HALF_OPEN

                self.success_count = 0

            else:

                remaining = int(self.timeout - elapsed)

                raise Exception(

                    f"Circuit OPEN — Zielsystem ausgefallen. "

                    f"Naechster Versuch in {remaining}s."

                )

        try:

            result = func(*args, **kwargs)

            self._on_success()

            return result

        except Exception:

            self._on_failure()

            raise

    def _on_success(self):

        if self.state == CircuitState.HALF_OPEN:

            self.success_count += 1

            if self.success_count >= self.success_threshold:

                self.state = CircuitState.CLOSED

                self.failure_count = 0

        else:

            self.failure_count = 0

    def _on_failure(self):

        self.failure_count += 1

        self.last_failure_time = time.time()

        if (self.state == CircuitState.HALF_OPEN or

                self.failure_count >= self.failure_threshold):

            self.state = CircuitState.OPEN

-- DLQ-Tabelle fuer SQL-basiertes Fehler-Management

CREATE TABLE Integration_DLQ (

    MessageID       UNIQUEIDENTIFIER DEFAULT NEWID() PRIMARY KEY,

    Eingang_Zeit    DATETIME2 DEFAULT GETDATE(),

    Queue_Name      NVARCHAR(100),

    Payload         NVARCHAR(MAX),

    Fehler_Meldung  NVARCHAR(MAX),

    Fehler_Code     NVARCHAR(50),

    Retry_Count     INT DEFAULT 0,

    Status          NVARCHAR(20) DEFAULT 'pending'

    -- 'pending', 'retrying', 'resolved', 'discarded'

);

-- Taeglich: Fehler-Analyse (welche Fehler haeufigen sich?)

SELECT

    Queue_Name,

    Fehler_Code,

    COUNT(*) AS Anzahl,

    MIN(Eingang_Zeit) AS Erster_Fehler,

    MAX(Eingang_Zeit) AS Letzter_Fehler

FROM Integration_DLQ

WHERE Status = 'pending'

  AND Eingang_Zeit >= DATEADD(DAY, -1, GETDATE())

GROUP BY Queue_Name, Fehler_Code

ORDER BY Anzahl DESC;

-- Manueller Retry einer Nachricht

UPDATE Integration_DLQ

SET Status = 'retrying',

    Retry_Count = Retry_Count + 1,

    Bearbeitet_Von = SYSTEM_USER

WHERE MessageID = 'GUID-HIER-EINTRAGEN'

  AND Status = 'pending';

SAEULE 1 — LOGS (Was ist passiert?):

  Format:     Strukturiertes JSON (kein Freitext — nicht durchsuchbar!)

  Pflichtfelder: timestamp, level, requestId, service, message, durationMs

  Korrelation: requestId durch alle Systeme durchschleifen

  Retention:  7 Tage operativ, 90 Tage Compliance, 1 Jahr Audit

  Tools:      ELK Stack, Azure Log Analytics, Datadog, Grafana Loki

SAEULE 2 — METRIKEN (Wie laeuft es?):

  Key Metrics:

    → Request Rate (Anfragen pro Sekunde)

    → Error Rate (% fehlgeschlagener Anfragen — SLO!)

    → Latency (P50, P90, P99 — nicht nur Durchschnitt!)

    → Throughput (Bytes/Sekunde)

    → Queue Depth (wie voll ist die Message Queue?)

  Tools:      Prometheus + Grafana, Azure Monitor, Datadog

SAEULE 3 — TRACES (Warum dauert es so lang?):

  Distributed Tracing: Eine Anfrage ueber mehrere Systeme verfolgen

  → API-Gateway → Middleware → ERP → Datenbank: jeder Schritt gemessen

  Trace-ID: Durch alle Systeme weiterleiten (W3C TraceContext-Header)

  Tools:      Jaeger, Zipkin, Azure Application Insights, Datadog APM

Error Rate

% 4xx/5xx

> 1 %

> 5 %

API liefert Fehler

P99 Latency

ms

> 500 ms

> 2.000 ms

Langsame Antwortzeiten

Request Rate

Req/s

Anomalie +50 %

Anomalie +200 %

Ungewoehnlicher Traffic

Queue Depth

Nachrichten

> 1.000

> 10.000

Consumer kommt nicht nach

Consumer Lag

Nachrichten

> 500

> 5.000

Kafka-Consumer zu langsam

DLQ Messages

Nachrichten

> 10/h

> 100/h

Verarbeitungsfehler

Token Errors

% 401/403

> 0,1 %

> 1 %

Auth-Problem

import logging, json, time

class StructuredLogger:

    # JSON-strukturierter Logger fuer API-Integrationen.

    def __init__(self, service_name: str):

        self.service = service_name

    def _log(self, level: str, message: str, **extra):

        record = {

            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),

            "level": level,

            "service": self.service,

            "message": message,

            **extra

        }

        print(json.dumps(record, ensure_ascii=False))

    def info(self, msg, **kw):  self._log("INFO",  msg, **kw)

    def warn(self, msg, **kw):  self._log("WARN",  msg, **kw)

    def error(self, msg, **kw): self._log("ERROR", msg, **kw)

# Verwendung:

# log = StructuredLogger("crm-erp-integration")

# log.info("api.request.success",

#          requestId="req-abc123",

#          endpoint="/api/v1/orders",

#          durationMs=42,

#          statusCode=200)

#

# Ausgabe:

# {"timestamp":"2026-03-15T09:00:00Z","level":"INFO",

#  "service":"crm-erp-integration","message":"api.request.success",

#  "requestId":"req-abc123","endpoint":"/api/v1/orders",

#  "durationMs":42,"statusCode":200}

FORMAT-VERGLEICH:

JSON (JavaScript Object Notation):

  Staerken:   Lesbar, kompakt, native Web-Unterstuetzung

  Schwaechen: Kein Schema erzwungen (ohne JSON Schema), kein Namespace

  Typisch:   REST-APIs, Web-Schnittstellen, moderne ERP-APIs

  Validierung: JSON Schema (draft-07 / 2020-12)

XML (Extensible Markup Language):

  Staerken:   Namespaces, XSD-Schema, XSLT-Transformation

  Schwaechen: Verbose, schwergewichtig, hoehere Parse-Kosten

  Typisch:   SOAP-Services, SAP-Schnittstellen, aeltere ERP-Systeme

EDIFACT / X12 (Electronic Data Interchange):

  Staerken:   Industriestandard, deterministische Struktur

  Schwaechen: Schwer lesbar, spezialisiertes Know-how noetig

  Typisch:   B2B: ORDERS, DESADV, INVOIC mit Handelspartnern

  Nachrichten: ORDERS (Bestellung), DESADV (Lieferavis), INVOIC (Rechnung)

CSV / XLSX:

  Staerken:   Universell verstaendlich, kein Tool noetig

  Schwaechen: Kein Schema, Encoding-Probleme, Sonderzeichen

  Fallstrick: IMMER Encoding pruefen (UTF-8 vs. Windows-1252!)

def transform_erp_order_to_crm(erp_payload: dict) -> dict:

    # Transformiert ERP-Auftragsformat in CRM-Opportunity-Format.

    customer = erp_payload.get("customer", {})

    positions = erp_payload.get("positions", [])

    return {

        "opportunityId": f"OPP-{erp_payload['orderId']}",

        "stage": "won",

        "title": f"Auftrag {erp_payload['orderId']}",

        "value": {

            "amount": sum(

                p.get("quantity", 0) * p.get("unitPrice", 0)

                for p in positions

            ),

            "currency": erp_payload.get("currency", "EUR"),

        },

        "customer": {

            "externalId": customer.get("customerId"),

            "name": customer.get("name"),

            "email": customer.get("email"),

        },

        "closedAt": erp_payload.get("orderDate"),

        "products": [

            {

                "sku": p.get("articleId"),

                "name": p.get("description"),

                "quantity": p.get("quantity"),

                "unitPrice": p.get("unitPrice"),

            }

            for p in positions

        ],

        "_meta": {

            "source": "erp",

            "sourceId": erp_payload["orderId"],

        }

    }

Encoding-Fehler

Windows-1252 vs. UTF-8

Immer explizit UTF-8 dekodieren/enkodieren

Datumsformat

DD.MM.YYYY vs. YYYY-MM-DD

Immer ISO 8601 intern, am Rand konvertieren

Dezimaltrennzeichen

Komma (DE) vs. Punkt (EN)

Locale explizit setzen, Float-Parse mit Locale

Nullwerte

null vs. "" vs. fehlendes Feld

Explizites Null-Handling, nie implizit annehmen

Trailing Whitespace

Felder aus Legacy-Systemen

TRIM() auf alle String-Felder bei Eingang

Zeitzone

UTC vs. lokale Zeit ohne TZ-Info

Immer UTC speichern, TZ nur fuer Anzeige

ANFORDERUNG:

  Wenn ein Auftrag im ERP angelegt wird:

  → CRM soll sofort die Opportunity auf "Won" setzen

  → CRM soll den Umsatz fuer das Kundenkonto aktualisieren

  → Fehler duerfen keinen Datenverlust verursachen

LOESUNG — EVENT-DRIVEN MIT QUEUE:

  ERP → Event → Message Queue → CRM-Consumer → CRM-API

Schritt 1: ERP feuert Event bei Auftragsanlage

  POST http://queue-service/events

  Body: { "type": "order.created", "orderId": "ORD-2026-4711", ... }

Schritt 2: Queue speichert Event persistent (Kafka / RabbitMQ)

  → Bei CRM-Ausfall: Events akkumulieren in Queue, kein Verlust

Schritt 3: CRM-Consumer verarbeitet Event

  → Holt Event aus Queue

  → Transformiert in CRM-Format

  → PUT https://crm.firma.de/api/v1/opportunities/{id}

  → Bei Fehler: Retry mit Backoff (max. 5), dann → DLQ

Schritt 4: Idempotenz sicherstellen

  → CRM prueft: Habe ich diesen Event (orderId) schon verarbeitet?

  → Idempotency-Key: orderId als eindeutiger Identifier

  → Duplikate werden ignoriert (nicht zweimal gebucht)

KRITISCH: Immer Idempotenz implementieren!

Message Broker garantieren "at-least-once" — doppelte Nachrichten sind normal.

class InventorySyncJob:

    # Synchronisiert Lagerbestaende vom ERP in den Webshop.

    # Laeuft alle 15 Minuten, nur geaenderte Artikel (Delta-Sync).

    def __init__(self, erp_client, shop_client, state_store):

        self.erp = erp_client

        self.shop = shop_client

        self.state = state_store

    def run(self) -> dict:

        from datetime import datetime, timezone, timedelta

        last_sync = self.state.get("last_sync_time") or \

                    (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()

        # 1. Delta aus ERP laden (nur geaenderte Artikel)

        changed_items = self.erp.get(

            "/api/v1/inventory",

            params={"modifiedSince": last_sync}

        ).json()["data"]

        stats = {"total": len(changed_items), "success": 0, "failed": 0}

        for item in changed_items:

            try:

                shop_payload = {

                    "sku": item["articleId"],

                    "quantity": max(0, item["stockQuantity"]),

                }

                self.shop.patch(

                    f"/api/products/{item['articleId']}/inventory",

                    json=shop_payload

                )

                stats["success"] += 1

            except Exception as e:

                stats["failed"] += 1

                # Einzelfehler stoppen nicht den gesamten Job

        self.state.set("last_sync_time",

                       datetime.now(timezone.utc).isoformat())

        return stats

EDIFACT ORDERS — VOLLSTAENDIGER ABLAUF:

  Handelspartner → EDIFACT-Datei → AS2-Empfang → EDI-Parser

  → ERP-Bestellung → Bestaetigung (ORDRSP) → Partner

SCHRITT 1: AS2-Empfang

  → Datei kommt per AS2-Protokoll (verschluesselt, signiert)

  → AS2-Server (Mendelson, B2BGateway) empfaengt und bestaetigt MDN

  → Datei in Verarbeitungsordner ablegen

SCHRITT 2: EDIFACT-Segmente (Beispiel)

  UNB+UNOC:3+LIEFERANT:14+KAEUFER:14+260315:1200+1'

  BGM+220+BEST-2026-001+9'       -- Bestellnummer

  DTM+137:20260315:102'          -- Bestelldatum

  NAD+BY+4012345000000::9'       -- Kaeufer (GLN)

  NAD+SU+4099999000000::9'       -- Lieferant (GLN)

  LIN+1++4000862141404:SRV'      -- Position 1, Artikel-EAN

  QTY+21:50:PCE'                 -- Menge: 50 Stueck

  PRI+AAA:19.95:CA'              -- Preis: 19,95 EUR

SCHRITT 3: ERP-Import

  → Parser wandelt EDIFACT-Segmente in ERP-Bestellformat

  → Pruefungen: Artikel existiert? Preis stimmt? Liefertermin realistisch?

SCHRITT 4: ORDRSP senden (Auftragsbestaetigung)

  → ORDRSP-Datei generieren, per AS2 zurueckschicken

  → Inhalt: Bestaetigt / Geaendert / Abgelehnt je Position

DIE GEFAEHRLICHSTEN ANTIPATTERNS

■  Datenbank-zu-Datenbank direkt: Direkter DB-Zugriff auf fremde Systeme bricht bei jedem Schema-Update und erzeugt unsichtbare Kopplung.

■  Polling ohne Delta: Statt "geaendert seit X" taeglich alle 500.000 Datensaetze laden — massiver Lastpeak, langsam, fehleranfaellig.

■  Fire-and-Forget ohne Bestaetigung: Nachricht senden, nicht pruefen ob angekommen — Datenverlust bei Netzwerkproblem garantiert.

■  Synchrone Kette: System A ruft B ruft C ruft D — ein Timeout in D blockiert A komplett.

■  Ohne Idempotenz: Netzwerkfehler → Retry → doppelte Bestellung im Zielsystem.

VOR DEM START

Alle beteiligten Systemverantwortlichen identifizieren und einbinden. API-Dokumentationen aller Quell- und Zielsysteme beschaffen. Middleware-Infrastruktur (Server, Lizenzen, Cloud-Ressourcen) bestellen. Secrets-Management-System einrichten. Test-Zugaenge zu allen Systemen beantragen.

ERGEBNIS NACH 45 TAGEN

Nach 45 Tagen haben Sie: eine vollstaendige, dokumentierte Integrationslandschaft mit API-Gateway, Message Broker und strukturiertem Fehlerhandling, OAuth2-gesicherte APIs mit zentralem Secrets-Management, ein Monitoring-Dashboard mit Alerting fuer alle kritischen Metriken, vollstaendige Runbooks fuer jeden Fehlerfall und eine skalierbare Architektur die neue Integrationen in Tagen statt Wochen ermoeglicht.

Integrations-Architektur-Beratung

Analyse Ihrer bestehenden Schnittstellenlandschaft, Architekturentscheidung und Roadmap — als Workshop oder laufende Begleitung.

API-Design & Implementierung

Design produktionsreifer REST-APIs mit OpenAPI-Spec, Sicherheitskonzept und vollstaendiger Dokumentation.

Middleware-Aufbau (Kafka / RabbitMQ)

Konzeption, Aufbau und Betriebsuebergabe von Message-Broker-Infrastrukturen fuer unternehmenskritische Integrationen.

EDI-Integration (EDIFACT / AS2)

B2B-Schnittstellen mit Handelspartnern — von der Partner-Onboarding-Strategie bis zur produktiven AS2-Strecke.

Integrations-Monitoring & Observability

Aufbau von Dashboards, Alerting und Runbooks fuer alle produktiven Schnittstellen Ihrer IT-Landschaft.