feat(backend): Pydantic v2 models for 837P claims

2026-06-19 15:39:08 -06:00
parent 8475e672a7
commit aa801805ce
2 changed files with 244 additions and 0 deletions
@@ -0,0 +1,142 @@
+"""Pydantic v2 models for parsed 837P claims.
+
+Models mirror the X12 837P structure (Loop 2000A → 2000B → 2300 → 2400) but
+flattened to one ``ClaimOutput`` per CLM segment, suitable for JSON serialization
+and consumption by the Cyclone frontend.
+"""
+
+from __future__ import annotations
+
+from datetime import date
+from decimal import Decimal
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, model_serializer
+
+
+class _Base(BaseModel):
+    model_config = ConfigDict(extra="ignore", str_strip_whitespace=True)
+
+    @model_serializer(mode="wrap")
+    def _serialize(self, handler):  # type: ignore[no-untyped-def]
+        data = handler(self)
+        for key, value in data.items():
+            if isinstance(value, date):
+                data[key] = value.isoformat()
+        return data
+
+
+class Address(_Base):
+    line1: str
+    line2: str | None = None
+    city: str
+    state: str
+    zip: str
+
+
+class BillingProvider(_Base):
+    name: str
+    npi: str
+    tax_id: str | None = None
+    address: Address | None = None
+
+
+class Subscriber(_Base):
+    first_name: str
+    last_name: str
+    member_id: str
+    dob: date | None = None
+    gender: Literal["M", "F", "U"] | None = None
+    address: Address | None = None
+
+
+class Payer(_Base):
+    name: str
+    id: str
+
+
+class ClaimHeader(_Base):
+    claim_id: str
+    total_charge: Decimal
+    place_of_service: str | None = None
+    frequency_code: str | None = None
+    provider_signature: str | None = None
+    assignment: str | None = None
+    release_of_info: str | None = None
+    prior_auth: str | None = None
+
+
+class Diagnosis(_Base):
+    code: str
+    qualifier: str | None = None
+
+
+class Procedure(_Base):
+    qualifier: str
+    code: str
+    modifiers: list[str] = Field(default_factory=list)
+
+
+class ServiceLine(_Base):
+    line_number: int
+    procedure: Procedure
+    charge: Decimal
+    unit_type: str | None = None
+    units: Decimal | None = None
+    place_of_service: str | None = None
+    service_date: date | None = None
+    provider_reference: str | None = None
+
+
+class ValidationIssue(_Base):
+    rule: str
+    severity: Literal["error", "warning"]
+    message: str
+    segment_index: int | None = None
+
+
+class ValidationReport(_Base):
+    passed: bool
+    errors: list[ValidationIssue] = Field(default_factory=list)
+    warnings: list[ValidationIssue] = Field(default_factory=list)
+
+
+class Envelope(_Base):
+    sender_id: str
+    receiver_id: str
+    control_number: str
+    transaction_date: date
+    transaction_time: str | None = None
+    implementation_guide: str | None = None
+
+
+class BatchSummary(_Base):
+    input_file: str
+    control_number: str | None = None
+    transaction_date: date | None = None
+    total_claims: int = 0
+    passed: int = 0
+    failed: int = 0
+    failed_claim_ids: list[str] = Field(default_factory=list)
+    issues_by_rule: dict[str, int] = Field(default_factory=dict)
+    output_dir: str | None = None
+
+
+class ClaimOutput(_Base):
+    claim_id: str
+    control_number: str
+    transaction_date: date
+    billing_provider: BillingProvider
+    subscriber: Subscriber
+    payer: Payer
+    claim: ClaimHeader
+    diagnoses: list[Diagnosis] = Field(default_factory=list)
+    service_lines: list[ServiceLine] = Field(default_factory=list)
+    validation: ValidationReport
+    raw_segments: list[list[str]] = Field(default_factory=list)
+
+
+class ParseResult(_Base):
+    envelope: Envelope | None = None
+    claims: list[ClaimOutput] = Field(default_factory=list)
+    summary: BatchSummary
@@ -0,0 +1,102 @@
+import json
+from datetime import date
+from decimal import Decimal
+
+import pytest
+from pydantic import ValidationError as PydanticValidationError
+
+from cyclone.parsers.models import (
+    Address,
+    BillingProvider,
+    ClaimHeader,
+    ClaimOutput,
+    Diagnosis,
+    Envelope,
+    Payer,
+    Procedure,
+    ServiceLine,
+    Subscriber,
+    ValidationIssue,
+    ValidationReport,
+)
+
+
+def test_address_round_trip():
+    a = Address(line1="1100 East Main St", line2="Suite A", city="Montrose", state="CO", zip="81401")
+    d = a.model_dump()
+    assert d["line1"] == "1100 East Main St"
+    assert d["line2"] == "Suite A"
+    assert Address.model_validate(d) == a
+
+
+def test_address_line2_optional():
+    a = Address(line1="1 Main", city="X", state="CO", zip="80000")
+    assert a.line2 is None
+
+
+def test_subscriber_splits_name():
+    s = Subscriber(first_name="John", last_name="Doe", member_id="ABC", dob=date(1980, 1, 1), gender="M", address=Address(line1="x", city="x", state="CO", zip="80000"))
+    assert s.first_name == "John"
+    assert s.gender == "M"
+
+
+def test_service_line_units_are_decimal():
+    sl = ServiceLine(
+        line_number=1,
+        procedure=Procedure(qualifier="HC", code="99213", modifiers=[]),
+        charge=Decimal("100.00"),
+        unit_type="UN",
+        units=Decimal("1.0"),
+    )
+    assert sl.units == Decimal("1.0")
+
+
+def test_claim_header_total_charge_decimal():
+    h = ClaimHeader(claim_id="C1", total_charge=Decimal("151.72"), place_of_service="12", frequency_code="1", provider_signature="Y", assignment="Y", release_of_info="Y", prior_auth="3173")
+    assert h.total_charge == Decimal("151.72")
+
+
+def test_validation_issue_strict_severity():
+    issue = ValidationIssue(rule="R001", severity="error", message="bad")
+    assert issue.severity == "error"
+    with pytest.raises(PydanticValidationError):
+        ValidationIssue(rule="R001", severity="WTF", message="bad")
+
+
+def test_validation_report_passed_true_when_no_errors():
+    r = ValidationReport(passed=True, errors=[], warnings=[])
+    assert r.passed is True
+
+
+def test_envelope_serializes_to_iso_date():
+    e = Envelope(
+        sender_id="11525703",
+        receiver_id="COMEDASSISTPROG",
+        control_number="991102977",
+        transaction_date=date(2026, 6, 11),
+        transaction_time="081417",
+        implementation_guide="005010X222A1",
+    )
+    d = e.model_dump()
+    assert d["transaction_date"] == "2026-06-11"
+
+
+def test_claim_output_full_round_trip():
+    claim = ClaimOutput(
+        claim_id="C1",
+        control_number="991102977",
+        transaction_date=date(2026, 6, 11),
+        billing_provider=BillingProvider(name="X", npi="1234567890", tax_id="123456789", address=Address(line1="1", city="x", state="CO", zip="80000")),
+        subscriber=Subscriber(first_name="J", last_name="D", member_id="M", dob=date(1980, 1, 1), gender="M", address=Address(line1="1", city="x", state="CO", zip="80000")),
+        payer=Payer(name="P", id="SKCO0"),
+        claim=ClaimHeader(claim_id="C1", total_charge=Decimal("100"), place_of_service="12", frequency_code="1", provider_signature="Y", assignment="Y", release_of_info="Y"),
+        diagnoses=[Diagnosis(code="R69", qualifier="ABK")],
+        service_lines=[],
+        validation=ValidationReport(passed=True, errors=[], warnings=[]),
+        raw_segments=[["CLM", "C1", "100"]],
+    )
+    blob = json.loads(claim.model_dump_json())
+    assert blob["claim_id"] == "C1"
+    assert blob["billing_provider"]["npi"] == "1234567890"
+    assert blob["raw_segments"] == [["CLM", "C1", "100"]]
+    assert ClaimOutput.model_validate(blob).claim_id == "C1"