De-serializing a nested python data class from json

Viewed 42

How to de-serialize a json file to a nested data class in a modular and fault-tolerant way?

@frozen(kw_only=True)
class Address:
    street: str
    city: str

@frozen(kw_only=True)
class Person:
    name: str
    age: int
    address: Address

@frozen
class Persons:
    data: List[Person]

Please read before marking as duplicate

  • This post is made for self-answering
  • Most posts suffer from one (or more) "flaws"
    • they bundle encoding and decoding together (but encoding is usually straight-forward)
    • they don't provide a modular approach (where each subclass de-serializes itself)
    • they are not fault-tolerant (single corrupted entry will fail the entire file)
1 Answers

Each class and subclass should have the following static methods:

  • valid_json_entry(json_entry) -> bool
  • decode(json_entry) -> Optional[<relevent class name (e.g. Person)>]

First step: "jsonification" of the input file:

with open("contacts.json") as fl:
    persons = Persons.decode(
        json.load(fl)
    )

Then, recursively decode subclasses:

    @staticmethod # Person.decode
    def decode(json_entry) -> Optional[Person]:
        
        if not Person.valid_json_entry(json_entry):
            return None

        if address := Address.decode(json_entry["address"]):
            return Person(
                name=json_entry["name"],
                age=json_entry["age"],
                address=address
            )

        return None

Overall, the entire solution looks like this:

from __future__ import annotations

import json
from attrs import frozen, asdict
from typing import List, Optional

@frozen(kw_only=True)
class Address:

    street: str
    city: str

    @staticmethod
    def valid_json_entry(json_entry) -> bool:
        return (
            "street" in json_entry and
            "city" in json_entry
        )

    @staticmethod
    def decode(json_entry) -> Optional[Address]:

        if not Address.valid_json_entry(json_entry):
            return None

        return Address(
            street=json_entry["street"],
            city=json_entry["city"]
        )

@frozen(kw_only=True)
class Person:

    name: str
    age: int
    address: Address

    @staticmethod
    def valid_json_entry(json_entry) -> bool:
        return (
            "name" in json_entry and
            "age" in json_entry and
            "address" in json_entry
        )

    @staticmethod
    def decode(json_entry) -> Optional[Person]:
        
        if not Person.valid_json_entry(json_entry):
            return None

        if address := Address.decode(json_entry["address"]):
            return Person(
                name=json_entry["name"],
                age=json_entry["age"],
                address=address
            )

        return None

@frozen
class Persons:

    data: List[Person]

    def __str__(self) -> str:
        """
        give a human readable, json string
        """
        return json.dumps(
            self.data,
            default=asdict,
            indent=4
        )

    @staticmethod
    def valid_json_entry(json_entry) -> bool:
        return "data" in json_entry

    @staticmethod
    def decode(json_entry) -> Optional[Persons]:
        
        if not Persons.valid_json_entry(json_entry):
            return None

        persons = []
        for item in json_entry["data"]:
            if person := Person.decode(item):
                persons.append(person)
            else:
                # skip malformed json entries
                # don't lose the entire file
                # for a single bad entry
                pass

        return Persons(persons)

with open("contacts.json") as fl:
    persons = Persons.decode(
        json.load(fl)
    )

print(persons)

When we feed the following json:

{
    "data": [
        {
            "name": "moish",
            "age": 7400,
            "address": {
                "street": "BenGurion",
                "city": "TelAviv"
            }
        },
        {
            "name": "uzi",
            "age": 6210,
            "address": {
                "street": "Kaplan",
                "citttty": "Holon"
            }
        }
    ]
}

our program is able to drop (just) the corrupted second entry

$ python3 main.py 
[
    {
        "name": "moish",
        "age": 7400,
        "address": {
            "street": "BenGurion",
            "city": "TelAviv"
        }
    }
]
Related