How can I safely normalize nested coverages into a flat dictionary while keeping from_dict() and to_dict() idempotent?

I am normalizing property coverage data that can arrive in either a nested or legacy flat format. The goal is to convert both formats into a single coverages structure before creating the model objects.

The normalization works in most cases, but I am concerned about keeping the conversion consistent when the data is deserialized with from_dict() and later serialized again with to_dict().

def expand_premises_coverages(premise: dict) -> list[dict]:
    """Flatten a nested premise (``{...shared, "coverages": [...]}``) into one
    row per coverage; return ``[premise]`` unchanged when it has no nested
    coverages list, so callers work with both the flat and nested input shapes.
    """
    if not isinstance(premise, dict):
        return []
    coverages = premise.get("coverages")
    if not isinstance(coverages, list) or not coverages:
        return [premise]
    shared = {k: v for k, v in premise.items() if k != "coverages"}
    rows = [{**shared, **cov} for cov in coverages if isinstance(cov, dict)]
    return rows or [premise]


def normalize_property_coverages(propertydata: dict | None) -> None:
    """Rewrite ``propertydata`` so all coverages live under a single flat
    ``coverages`` list.

    For nested inbound (``premisesN`` with a ``coverages: [...]`` list) each
    coverage is emitted as its own row inheriting the parent's identity fields
    (premises_number, building_number, street_address, bldg_desc, etc.).
    For legacy flat inbound (``premisesN`` already one-subject-per-row) the
    rows are promoted 1:1 into ``coverages``. Either way ``propertydata`` ends
    up with a canonical ``coverages`` list that every downstream reader (cp7,
    gl7, submit_validation, …) can iterate uniformly – no per-reader adapter
    needed. Idempotent when ``coverages`` is already a populated list.
    """
    if not isinstance(propertydata, dict):
        return
    existing = propertydata.get("coverages")
    if isinstance(existing, list) and existing:
        return
    if isinstance(existing, dict) and existing:
        # Legacy coverageN-keyed dict – flatten to a list.
        propertydata["coverages"] = [
            row for row in existing.values() if isinstance(row, dict)
        ]
        propertydata.pop("premises", None)
        propertydata.pop("premises_information", None)
        return
    premises = (
        propertydata.get("premises")
        or propertydata.get("premises_information")
    )
    if not premises:
        return
    if isinstance(premises, dict):
        entries = [row for row in premises.values() if isinstance(row, dict)]
    elif isinstance(premises, list):
        entries = [row for row in premises if isinstance(row, dict)]
    else:
        return
    flat: list[dict] = []
    for entry in entries:
        flat.extend(expand_premises_coverages(entry))
    if not flat:
        return
    propertydata["coverages"] = flat
    # Drop the legacy premises group so downstream sees a single source of truth.
    propertydata.pop("premises", None)
    propertydata.pop("premises_information", None)


    @classmethod
    def from_dict(cls, data: dict | None) -> PropertyData:
        data = data or {}
        normalize_property_coverages(data)
        coverages_block = (
            data.get("coverages")
            or data.get("premises")
            or data.get("premises_information")
            or {}
        )
        rows: list[PremisesRow] = []
        if isinstance(coverages_block, dict):
            for _key, row in coverages_block.items():
                if isinstance(row, dict):
                    rows.append(PremisesRow.from_dict(row))
        elif isinstance(coverages_block, list):
            for row in coverages_block:
                if isinstance(row, dict):
                    rows.append(PremisesRow.from_dict(row))
        return cls(
            premises=rows,
            additional_coverages=dict(
                data.get("additionalCoverages")
                or data.get("additional_coverages")
                or {}
            ),
            raw=dict(data),
        )

    def to_dict(self) -> dict:
        """Emit the normalized ``coverages`` list and drop legacy premises keys."""
        out = dict(self.raw) if self.raw else {}
        out["coverages"] = [row.to_dict() for row in self.premises]
        out.pop("premises", None)
        out.pop("premises_information", None)
        if self.additional_coverages:
            out["additionalCoverages"] = self.additional_coverages
        return out

Вернуться на верх