REWE Modell mit DATEVconnect per Python Container

Kurzanleitung für die Einrichtung des REWE Modells mit DATEVconnect Anbindungin der Cloud, hierbei wird auf DATEV per Riecken Cloud Gateway zugegriffen.

Funktion des Riecken Gateways über ein Python Skript testen

https://api.cloudgateway.riecken.io/datev/api/accounting/v1/clients → Test des Mandantenabrufs, muss die Liste der relevanten Mandanten liefern
https://api.cloudgateway.riecken.io/datev/api/accounting/v1/clients/xxx-yyy-xxx-dda/fiscal-years-id/20250101/account-postings?filter=account_number ge 10000000 and account_number le 20000000 → Test des Buchungsabrufs, muss die Buchungen auf Konto 1000 bis 2000 liefern (Bereich ggf. anpassen)

Testscript für Riecken Gateway

Python


# Source: logikrepo-datev/src/main/Gateways/DATEVconnect/Entwicklung/Komponenten/Testabruf_Clients_Riecken.py
# Testscript für die Riecken Web API 
# Zuletzt modifiziert: 04/2026 - Gerd Tautenhahn für Saxess Software GmbH
# es wird testweise die Liste der clients (Mandanten abgerufen)
# bei erfolgreichem Aufruf erscheint ein JSON Objekt mit Mandantennamen und Nummern
# vom ersten Mandanten werden die Buchungen der Konten 1000 bis 2000 abgerufen, um zu testen um die Lizenz für Kontobuchungen vorhanden ist

import requests
from requests.auth import HTTPBasicAuth
from urllib.parse import quote
import json

username    = 'xxxxx-xxxxx-xxxxx-xxxxx'
password    = 'yyyy-yyyy-yyyy-yyyyy'
base_url    = 'https://api.cloudgateway.riecken.io/datev/api/accounting/v1'
url_clients = f'{base_url}/clients'
# url_clients = 'https://datevconnect.riecken-webservices.at/datev/api/accounting/v1/clients' -- alte URL Route


# ---------------------------------------------------------------
# 1. Abruf: Liste der Clients (Mandanten)
# ---------------------------------------------------------------
print("=" * 60)
print("1. Abruf: Clients (Mandanten)")
print("=" * 60)

response = requests.get(
    url_clients,
    auth=HTTPBasicAuth(username, password)
)
print(f"Status Code: {response.status_code}")
clients_data = response.json()
print(json.dumps(clients_data, indent=2, ensure_ascii=False))

# ---------------------------------------------------------------
# 2. Abruf: Buchungen (Account Postings) für den ersten Client
# ---------------------------------------------------------------
print()
print("=" * 60)
print("2. Abruf: Buchungen (Account Postings) für ersten Client")
print("=" * 60)

# client_id aus dem ersten Eintrag der zurückgegebenen Liste extrahieren
# Antwortformat: Liste von Objekten, jedes mit einem "id"-Feld
try:
    if isinstance(clients_data, list):
        clients_list = clients_data
    elif isinstance(clients_data, dict):
        # Falls die Liste unter einem Schlüssel wie "data" oder "value" liegt
        clients_list = clients_data.get('data') or clients_data.get('value') or []
    else:
        clients_list = []

    if not clients_list:
        raise ValueError("Keine Clients in der Antwort gefunden.")

    first_client = clients_list[0]
    client_id    = first_client.get('id') or first_client.get('client_id')
    client_name  = first_client.get('name', '(unbekannt)')

    if not client_id:
        raise ValueError(f"Kein 'id' oder 'client_id' Feld im ersten Client gefunden: {first_client}")

    print(f"Verwende Client: {client_name} (ID: {client_id})")

    # Buchungsabruf: Konten 1000–2000 (8-stellige Kontonummern: 10000000–20000000)
    fiscal_year   = '20250101'
    account_from  = 10000000   # entspricht Konto 1000 (8-stellig)
    account_to    = 20000000   # entspricht Konto 2000 (8-stellig)
    filter_param  = f'account_number ge {account_from} and account_number le {account_to}'

    url_postings = f'{base_url}/clients/{client_id}/fiscal-years/{fiscal_year}/account-postings'

    print(f"Basis-URL: {url_postings}")
    print(f"Filter:    {filter_param}")
    print()

    # Filter-Parameter manuell mit %20 kodieren (nicht + wie bei params=)
    url_postings_with_filter = f'{url_postings}?filter={quote(filter_param)}'

    response2 = requests.get(
        url_postings_with_filter,
        auth=HTTPBasicAuth(username, password)
    )
    print(f"Tatsächlich gesendete URL: {response2.request.url}")
    print(f"Status Code: {response2.status_code}")
    postings_data = response2.json()
    print(json.dumps(postings_data, indent=2, ensure_ascii=False))

except (IndexError, KeyError, ValueError) as e:
    print(f"Fehler beim Verarbeiten der Clients-Antwort: {e}")

Installation von OCT

OCT Datenbank in der OCT Cloud bereitstellen lassen, Version 2026.07 (🔒 sxProzess)
Storage Account mit AccessKey in der OCT Cloud bereitstellen lassen

Einrichtung der OCT Datenbank und des Storage Accounts

Storage Account als Datenquelle in OCT registrieren
Gateway_DATEVconnect.zip 🔒herunterladen und entpacken
im Storage Account eine Dateifreigabe “python” anlegen
Gateway_DATEVconnect Ordner auf den Storage Account in diese Dateifreigabe kopieren
config.json Datei bearbeiten

Einrichtung der OCT Datenbank für die DATEVconnect Staging Daten

über die Modulverwaltung das Modul GATEWAY-DATEVCONNECT-PYTHON ausführen → erstellt die Tabellen und die Pipeline für den staging Prozess
in der Pipeline GWDC die beiden Steps für Ausführung per Python löschen → Diese werden nicht benötigt, da die Verarbeitung über Container erfolgt.
den Step für die Ausführung über Container mit den passenden Containerparametern 🔒 konfigurieren
Containerstep zum Abruf der Steuerungstabelle aktivieren und Pipeline ausführen (nur diesen einen Step)
Ergebnis im Validierungs-Tab Steuerungstabelle der Produktlinie GWDC in der Factory ZV anschauen und über die abzurufenden Mandanten entscheiden
relevante Mandanten im Step Mandantenaktivierung der Pipeline aktivieren
Containerstep für Staging aktivieren und Pipeline ausführen
Daten im Tab für die Validierung der Gatewaydaten validieren

Mit dem Staging der DATEV Daten ist man jetzt fertig - es folgt die Standardeinrichtung des REWE Modells.

Einrichtung der OCT Datenbank für das REWE Modell

REWE Modul und Konnektor einspielen
- REWE Modell inkl. Modellierung über die OCT Modulverwaltung einspielen (Modul “REWE”) → erstellt die integration.tREWE* und result.tREWE* Tabellen + Views etc. und die Pipeline RESULTLOAD, erweitert Pipeline TRUNCATE und ATR
- Konnektor DATEVconnect_REWE über die OCT Modulverwaltung einspielen (Modul “REWE-KONNEKTOR-GATEWAY-DATEVCONNECT”) → erstellt die Pipeline GWDC
Modellierung einrichten
- In TAB “GWDC” die Steuerungstabelle erneut prüfen und sich die dort eingesetzten Kontenrahmen merken.
- In der Factory “Modellierung (M)”
  - aus der Produktline “Templates Modellierung (TM)” das Produkt “HGB Struktur” in die Produktline “Strukturen (MS)” kopieren (per Drag & Drop)
  - jeden benötigten Kontenrahmen aus Produktline “Templates Modellierung (TM)” in Produktline “Zuordnungen (MZ)” kopieren (per Drag & Drop)
- Pipeline “Modellierung” ausführen
Pipeline für Konnektor konfigurieren
- Die Konfiguration muss nicht angepasst werden, die Pipeline wird einfach ausgeführt. → Diese Pipeline holt die Daten aus staging* nach integration* . Das Ergebnis kann in den Validierungstabellen der Produktline analysiert werden.
Pipeline RESULTLOAD
- Die Konfiguration muss nicht angepasst werden, die Pipeline wird einfach ausgeführt. → Diese Pipeline holt die Daten aus integration* nach result* . Das Ergebnis kann in den Validierungstabellen der Produktline analysiert werden.

Optionale Komponenten

Modul “REWE-ADDON-CORPORATE-PLANNER” und “REWE-ADDON-BUCHUNGSKREISMAPPING” einspielen Doku → Dieses bereitet die Daten aus result* für CP auf. Das Ergebnis kann in den Validierungstabellen der Produktline analysiert werden.
Modul “REWE für Bissantz DeltaMaster” einspielen → Dieses bereitet die Daten aus result* für DeltaMaster auf. Das Ergebnis kann in den Validierungstabellen der Produktline analysiert werden.

Last updated: September 11, 2026