Store nested json repsonses in relational database

Question

The goal is to collect market data from a free rest api. The response is in json and size per response is above 1MB. I want to get at least updated data once per minute which means about 24h * 60min * 1MB ~ 1.5GB per day of json data. Since i have to collect data over a few weeks in order to start analysing market patterns i have to store the json responses in a space efficient way --> SQLite is my idea so far. Now the ugly part is dealing with nested (4 levels) json and insert it into a relational database.

json example:

{
    'country-id': 1,
    'name': 'USA',
    'states': [
    {
        'country-id': 1,
        'name': 'Alaska',
        'state-id': 11,
        'cities': [
        {
            'state-id': 11,
            'name': 'Anchorage',
            'city-id': 111,
            'streets':
            [
                {
                    'city-id': 111,
                    'name': 'Ingra St.',
                    'street-id': 1111
                }
            ]
        },
        {
            'state-id': 11,
            'name': 'Fairbanks',
            'city-id': 112,
            'streets': 
            [
                {
                    'city-id': 112,
                    'name': 'Johansen Expy',
                    'street-id': 1121
                }
            ]
        }
        ]
    },
    ]
}

My approach:

from dataclasses import dataclass
from typing import List


@dataclass
class Country:
    def __init__(self, country_json):
        self.country_id: int = country_json['country-id']
        self.name: str = country_json['name']
        self.states: List(State) = [State(state) for state in country_json['states']]


@dataclass
class State:
    def __init__(self, state_json):
        self.country_id: int = state_json['country_id']
        self.name: str = state_json['name']
        self.state_id: int = state_json['state_id']
        self.cities: List(City) = [City(city) for city in state_json['cities']]


@dataclass
class City:
    def __init__(self, city_json):
        self.state_id: int = city_json['state-id']
        self.name: str = city_json['name']
        self.city_id: int = city_json['city-id']
        self.streets: List(Street) = [Street(street) for street in city_json['street']]


@dataclass
class Street:
    def __init__(self, street_json):
        self.city_id: int = street_json['city-id']
        self.name: str = street_json['name']
        self.street_id: int = street_json['street-id']

Next steps would be to add functions to each dataclass that store the values in a SQlite DB. I am really not sure if my general approach is efficient or if there is a better solution for this kind of problem. My idea in general was that if i build it like this i can easily add other steps e.g. data validation.

Answer 1

You don't use typing properly. Parametrizing generic types should be done using brackets notation (aka __getitem__) not parenthesis (aka instantiation).

You also don't use any feature of dataclasses in your defined classes, so you might as well drop that dependency. Or you could use it properly so that Country(**json_data) will build the whole thing; but:

• Keys in the JSON data are not valid python identifiers, you would need to convert them;
• You would need to handle converting children "manually" after the __init__ took place;
• You will need to swap the order in which you define classes due to scope evaluation.

The closest we would come should be something along the lines of:

from dataclasses import dataclass
from typing import List


def convert_keys(dct):
    return {name.replace('-', '_'): value for name, value in dct.items()}


@dataclass
class Street:
    city_id: int
    name: str
    street_id: int


@dataclass
class City:
    state_id: int
    name: str
    city_id: int
    streets: List[Street]

    def __post_init__(self):
        self.streets = [Street(**convert_keys(street)) for street in self.streets]


@dataclass
class State:
    country_id: int
    name: str
    state_id: int
    cities: List[City]

    def __post_init__(self):
        self.cities = [City(**convert_keys(city)) for city in self.cities]


@dataclass
class Country:
    country_id: int
    name: str
    states: List[State]

    def __port_init__(self):
        self.states = [State(**convert_keys(state)) for state in self.states]

Initiate the call using Country(**convert_keys(json_data)). But this solution doesn't necessary feel cleaner than yours.

---

Now as regards to using these classes as a mean to store data into a relational database, we need to examine usage.

The DB API 2.0 tells us that you can expect to be able to:

cursor.execute('INSERT INTO Street VALUES (?, ?, ?)', (city_id, name, street_id))

Which means that we need to:

1. Be able to convert these classes to tuples;
2. Remove the reverse relationship that we had so much troubles parsing properly Country.states, State.cities, and City.streets).

So here we go, trying to patch our approach. We could use dataclasses.astuple to convert or objects to proper parameters for our query, but we would still need a specific parser to recursively traverse nested JSON layers. Coupling that with the invalid identifiers issue, I don't think storing the data into intermediate classes makes much sense. Instead I would rather write them directly in the database:

def parse_country(cursor, json_data):
    country_id = json_data['country-id']
    name = json_data['name']
    cursor.execute('INSERT INTO Country VALUES (?, ?)', (country_id, name))
    for state in json_data['states']:
        parse_state(cursor, state)


def parse_state(cursor, json_data):
    state_id = json_data['state-id']
    name = json_data['name']
    country_id = json_data['country-id']
    cursor.execute('INSERT INTO State VALUES (?, ?, ?)', (state_id, name, country_id))
    for city in json_data['cities']:
        parse_city(cursor, city)


def parse_city(cursor, json_data):
    city_id = json_data['city-id']
    name = json_data['name']
    state_id = json_data['state-id']
    cursor.execute('INSERT INTO City VALUES (?, ?, ?)', (city_id, name, state_id))
    for street in json_data['streets']:
        parse_street(cursor, street)


def parse_street(cursor, json_data):
    street_id = json_data['street-id']
    name = json_data['name']
    city_id = json_data['city-id']
    cursor.execute('INSERT INTO Street VALUES (?, ?, ?)', (street_id, name, city_id))

And that's pretty much your original code except the data is now in DB and not in memory. Usage being:

conn = # create appropriate DB connection here
with conn:
    parse_country(conn.cursor(), json_data)