SC ReconciliationReconciliation Automation
  1. Home
  2. Architecture
  3. EDI 810 vs 850 Schema Mapping
  4. How to Map EDI 810 Invoices to Internal PO Schemas

How to Map EDI 810 Invoices to Internal PO Schemas

↑ Part of EDI 810 vs 850 Schema Mapping.

When an EDI 810 invoice lands in your pipeline, the mapping layer must resolve it against the originating 850 purchase order, normalize line-level quantities and pricing into a single canonical record, and commit the result to your ERP without ever writing a duplicate financial row. This page is the concrete procedure for that mapping step: deterministic X12 parsing, a strict target schema, segment-to-field transformation, and an idempotent commit. It assumes a Python ETL stack with Pydantic v2 and transactional Postgres writes, and it sits underneath the broader EDI 810 vs 850 Schema Mapping contract that the downstream match engine depends on.

Operational Trigger Signals

Build this mapping layer the moment any of these measurable conditions appear in your invoice stream — they all indicate that naive string parsing or direct ERP loads are about to leak financial variance:

  1. Trading-partner volume crosses ~50 invoices/day and manual AP keying can no longer keep pace with the 810 arrival rate.
  2. po_number lives in two segments (BIG04 and REF*PK) and partners populate them inconsistently, producing po_join_mismatch events.
  3. Line-item drift exceeds tolerance — abs(invoiced_qty - po_qty) / po_qty > 0.02 or unit-price delta > 0.05 on more than 1% of lines, signalling partial shipments or backorder splits.
  4. Duplicate submissions appear — the same (supplier_id, invoice_number, currency_code) arrives twice via VAN redelivery or AS2 retry.
  5. Mixed-currency invoices require FX normalization before settlement; see Multi-Currency Reconciliation Frameworks.
  6. Schema drift from a partner introduces unexpected NTE or AMT loops that break positional IT1 extraction.

Step-by-Step Implementation

Step 1 — Parse and normalize the X12 envelope

EDI 810 files use tilde (~) as segment terminators and asterisk (*) as element delimiters. Do not rely on naive str.split; implement a stateful parser that respects hierarchical loops (IT1, PID, AMT) and tolerates carriage returns or malformed line breaks. The parser below normalizes whitespace, isolates segments, groups them by identifier, and logs anomalies for downstream triage.

EDI 810 mapping pipeline, from raw payload to idempotent commit A left-to-right pipeline for mapping an EDI 810 invoice. The raw X12 810 payload feeds a stateful, loop-aware segment parser that emits a dict of segment id to element arrays. That feeds Pydantic v2 schema validation, which coerces every monetary value to fixed-point Decimal and builds an InvoiceHeader with line items. Validated records reach the tolerance gate, which checks the quantity and relative-price bands. Lines within both bands branch up to an idempotent ERP commit keyed on supplier, invoice number, and currency; lines that breach tolerance branch down to a tagged exception queue or dead-letter queue carrying the raw payload pointer and a failure reason for replay. Raw 810 payload X12 envelope ISA·GS·ST … SE·IEA Segment parser stateful · loop-aware dict[seg_id] → elems Schema validation Pydantic v2 · Decimal InvoiceHeader + lines Tolerance gate qty & price bands |Δq|≤τ · |Δp/p|≤τ ERP commit idempotent upsert key (supplier,inv,ccy) Exception queue / DLQ tagged · replayable raw_payload_uri + reason within bands breach → queue
PYTHON
import logging
from typing import Dict, List

logger = logging.getLogger("edi810.parser")

def parse_edi_810(raw_payload: str) -> Dict[str, List[List[str]]]:
    """
    Split a raw EDI 810 payload into a dict of segment_id -> list of element arrays.
    Handles ~ terminators, normalizes line breaks, and preserves element order.
    """
    normalized = raw_payload.replace("\r\n", "\n").replace("\r", "\n")
    segments = [seg.strip() for seg in normalized.split("~") if seg.strip()]

    parsed: Dict[str, List[List[str]]] = {}
    for seg in segments:
        parts = seg.split("*")
        seg_id = parts[0]
        parsed.setdefault(seg_id, []).append(parts)

    # Envelope sanity: ISA must equal IEA, ST must equal SE.
    if len(parsed.get("ISA", [])) != len(parsed.get("IEA", [])):
        logger.warning("edi_envelope_mismatch isa=%s iea=%s",
                       len(parsed.get("ISA", [])), len(parsed.get("IEA", [])))
    logger.info("edi810_parsed segments=%d distinct_ids=%d",
                len(segments), len(parsed))
    return parsed

Configuration hardening for this step:

  • Set a maximum segment-length guard to reject truncated transmissions before they reach the mapper.
  • Enable strict loop-order validation if partners consistently violate X12 4010/5010 loop sequencing.
  • Archive the raw payload to an encrypted, immutable store before parsing, per the controls in Data Security Boundaries for Procurement Systems.

Step 2 — Define the strict target schema

Model the internal PO/invoice record with Pydantic v2 so type coercion errors and procurement-rule violations surface before reconciliation, not after settlement. The same validation discipline is covered in depth under Schema Validation Using Pydantic.

PYTHON
from pydantic import BaseModel, Field, field_validator
from decimal import Decimal, ROUND_HALF_UP
from datetime import date
from typing import Optional, List

class InvoiceLineItem(BaseModel):
    line_number: int = Field(ge=1)
    supplier_sku: str = Field(min_length=1)
    internal_sku: Optional[str] = None
    quantity_invoiced: Decimal = Field(ge=0, decimal_places=2)
    unit_price: Decimal = Field(ge=0, decimal_places=4)
    uom: str = Field(min_length=2, max_length=3, pattern="^[A-Z]{2,3}$")
    extended_amount: Optional[Decimal] = None

    @field_validator("extended_amount", mode="before")
    @classmethod
    def calc_extended(cls, v: Optional[Decimal], info) -> Decimal:
        if v is not None:
            return v.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
        qty = info.data.get("quantity_invoiced")
        price = info.data.get("unit_price")
        if qty and price:
            return (qty * price).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
        raise ValueError("extended_amount requires quantity_invoiced and unit_price")

class InvoiceHeader(BaseModel):
    invoice_number: str = Field(min_length=1, max_length=20)
    po_number: str = Field(min_length=1, max_length=20)
    invoice_date: date
    supplier_id: str
    currency_code: str = Field(min_length=3, max_length=3)
    line_items: List[InvoiceLineItem]
    total_amount: Decimal = Field(ge=0, decimal_places=2)

Carry every monetary value as a fixed-point Decimal from the parser through the tolerance check — floating-point representation error can push an otherwise-clean line just outside a tight price window and raise a false breach.

Step 3 — Transform segments into schema fields

Map X12 segments to model fields with a deterministic traversal. Each rule is positional and auditable:

X12 Segment Target Field Transformation Rule
BIG (02) invoice_number Strip whitespace. Validate against the duplicate index.
BIG (03) invoice_date Parse YYYYMMDD. Reject if future-dated beyond T+3 business days.
REF*IA / BIG (04) po_number Resolve with precedence BIG04 → REF*PK. Cross-reference the active PO table; fail if status not in OPEN/PARTIAL.
IT1 (01) line_number Cast to int. Validate sequential ordering.
IT1 (02) quantity_invoiced Cast to Decimal. ROUND_HALF_UP to 2 places.
IT1 (04) unit_price Cast to Decimal. ROUND_HALF_UP to 4 places.
IT1 (06) uom Uppercase. Map supplier UOM to internal base unit (e.g., CS → EA via conversion factor).
TDS (01) total_amount Compare against SUM(line.extended_amount) within ±0.05 rounding tolerance.

Use the product identifier (IT106/IT107) as the authoritative line key after alias resolution, and treat IT101 only as a fallback ordinal — sequence-only alignment breaks the instant a supplier reorders, splits, or consolidates invoice lines.

Step 4 — Apply the tolerance gate before commit

A line is eligible for automatic settlement only when both quantity and relative price fall inside their windows. The exact tolerance calibration is owned by Setting Quantity and Price Tolerance Windows, but the gate this mapper enforces is:

∣qinv−qpo∣≤τqtyand∣pinv−ppoppo∣≤τprice\left| q_{inv} - q_{po} \right| \le \tau_{qty} \quad\text{and}\quad \left| \frac{p_{inv} - p_{po}}{p_{po}} \right| \le \tau_{price}

Lines that satisfy both inequalities post to ERP; everything else routes to a tagged exception queue rather than auto-rejecting.

Step 5 — Commit idempotently

Once validation and the tolerance gate pass, execute an idempotent upsert inside a transaction with explicit rollback. The composite conflict key is what makes a redelivered 810 safe.

PYTHON
import logging
from psycopg2.extras import execute_values

logger = logging.getLogger("edi810.commit")

def commit_reconciliation(invoice: "InvoiceHeader", db_conn) -> None:
    """Upsert header + lines in one transaction; idempotent on (supplier, invoice, currency)."""
    with db_conn.cursor() as cur:
        try:
            cur.execute(
                """INSERT INTO invoices (invoice_number, po_number, invoice_date,
                   supplier_id, currency_code, total_amount, status)
                   VALUES (%s, %s, %s, %s, %s, %s, 'RECONCILED')
                   ON CONFLICT (supplier_id, invoice_number, currency_code)
                   DO UPDATE SET status = EXCLUDED.status""",
                (invoice.invoice_number, invoice.po_number, invoice.invoice_date,
                 invoice.supplier_id, invoice.currency_code, invoice.total_amount),
            )
            line_data = [
                (invoice.invoice_number, item.line_number, item.supplier_sku,
                 item.internal_sku, item.quantity_invoiced, item.unit_price,
                 item.uom, item.extended_amount)
                for item in invoice.line_items
            ]
            execute_values(cur,
                """INSERT INTO invoice_lines (invoice_number, line_number, supplier_sku,
                   internal_sku, qty_invoiced, unit_price, uom, extended_amount)
                   VALUES %s ON CONFLICT DO NOTHING""",
                line_data,
            )
            db_conn.commit()
            logger.info("invoice_committed invoice=%s lines=%d",
                        invoice.invoice_number, len(line_data))
        except Exception as e:
            db_conn.rollback()
            logger.error("commit_failed invoice=%s error=%s", invoice.invoice_number, e)
            raise RuntimeError(f"Transaction failed for invoice {invoice.invoice_number}") from e

Configuration Reference

Parameter Accepted values Default Notes
segment_terminator single char ~ Some partners use \n; read from ISA16 when present.
element_delimiter single char * Taken from the ISA fixed-width header.
qty_tolerance_abs (τ_qty) 0–N units 0 Absolute unit slack before exception routing.
price_tolerance_pct (τ_price) 0.00–0.15 0.05 Relative unit-price band; >0.15 should hard-fail.
total_rounding_tolerance Decimal 0.05 TDS vs summed extended_amount.
invoice_date_future_window business days 3 Reject invoice_date beyond T+3.
manual_review_ceiling 0.00–1.00 0.15 Variance above this never auto-settles.
idempotency_key tuple (supplier_id, invoice_number, currency_code) Composite unique index.

Debugging & Recovery

Reconciliation pipelines fail predictably. Route each failure to a tagged dead-letter queue (DLQ) entry carrying the raw payload pointer, the failing segment, and the resolution owner.

  1. PO_NOT_FOUND / stale PO — SELECT status, last_updated FROM purchase_orders WHERE po_number = ?. If CLOSED, route to AP_HOLD; if DRAFT, fire the PO-activation webhook. Log {"error": "PO_NOT_FOUND", "po": po_number, "invoice": invoice_number}.
  2. QTY_PRICE_VARIANCE — when abs(invoiced_qty - po_qty) / po_qty > 0.02 or abs(price_diff) > 0.05, flag the line for manual review (never auto-reject under the manual_review_ceiling). Compare IT1 elements against the 850 baseline to spot partial shipments or backorder splits.
  3. DUPLICATE_INVOICE — the composite unique index raises a conflict; return HTTP 409, log the attempt, and do not process. A pre-commit check against a Redis cache of recently seen (supplier_id, invoice_number, currency_code) keys suppresses VAN/AS2 retries before they hit the database.
  4. SCHEMA_DRIFT — unexpected NTE or AMT loops break positional IT1 parsing. Temporarily relax loop-order enforcement, fall back to keyed regex extraction for critical fields, and watch for segment-count deviations.

Audit-trail fields to persist on every record: raw_payload_uri, parsed_at, mapper_version, tolerance_profile_id, dlq_reason, and the final status transition. Confirm the layer is healthy by tracking the monitoring metrics edi_segment_anomalies_total, po_join_mismatch_total, invoice_duplicate_suppressed_total, and DLQ depth; alert when DLQ depth grows faster than the drain rate.

Final validation checklist

FAQ

Should I match invoice lines by IT101 sequence or product ID?

Use the product identifier (IT106/IT107) as the authoritative line key after alias resolution, and treat IT101 as a fallback ordinal only when product IDs are missing. Sequence-only alignment fails the moment a supplier reorders, splits, or consolidates lines — which is routine on partial shipments.

How do I stop a redelivered 810 from paying a supplier twice?

Persist an idempotency key of (supplier_id, invoice_number, currency_code) (extend with line_sequence for line-level safety) and enforce it as a composite unique index before the ERP post. A VAN redelivery or AS2 retry then maps to the same key and is suppressed at the commit boundary, keeping an at-least-once transport effectively exactly-once.

What belongs in this mapping layer versus the match engine?

This layer only parses, normalizes both documents into the canonical schema, and validates structure plus the tolerance gate; it does not adjudicate disputes. Tier selection and exception resolution belong to the downstream match engine, which keeps the mapper stateless and fully replayable from the archived raw payload.