Работа с глубокими JSON: от получения значения до маппинга в модели

Расширенные примеры работы с вложенным JSON

Пример 1: Обработка большого JSON с помощью потоковой загрузки

Когда JSON-файл занимает гигабайты, обычный json.load() может потребовать слишком много памяти. Используйте ijson для потокового парсинга.

import ijson

with open('huge.json', 'r', encoding='utf-8') as f:
    parser = ijson.parse(f)
    for prefix, event, value in parser:
        if prefix == 'items.item.id' and event == 'number':
            print(f"Found ID: {value}")

Результат выводит все идентификаторы по мере поступления данных, не загружая весь файл.

Пример 2: Объединение нескольких вложенных JSON-объектов

Часто нужно объединить два JSON-документа с разной структурой. Используйте рекурсивное слияние словарей.

def deep_merge(dict1, dict2):
    result = dict1.copy()
    for key, value in dict2.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = deep_merge(result[key], value)
        else:
            result[key] = value
    return result

a = {"user": {"name": "Иван", "contact": {"phone": "123"}}}
b = {"user": {"contact": {"email": "ivan@example.com"}, "age": 30}}
merged = deep_merge(a, b)
print(merged)
# {'user': {'name': 'Иван', 'contact': {'phone': '123', 'email': 'ivan@example.com'}, 'age': 30}}

Результат - один словарь с объединёнными данными. Внимание: значения-не-словари заменяются, а не сливаются.

Пример 3: Фильтрация вложенного JSON с помощью рекурсивной функции

Извлечь все значения, соответствующие определённому условию, независимо от глубины.

def find_all(data, condition):
    results = []
    if condition(data):
        results.append(data)
    if isinstance(data, dict):
        for v in data.values():
            results.extend(find_all(v, condition))
    elif isinstance(data, list):
        for item in data:
            results.extend(find_all(item, condition))
    return results

data = [
    {"type": "fruit", "value": 5},
    {"type": "vegetable", "value": 3},
    {"nested": [{"type": "fruit", "value": 8}]}
]
fruits = find_all(data, lambda x: isinstance(x, dict) and x.get("type") == "fruit")
print(fruits)  # [{'type': 'fruit', 'value': 5}, {'type': 'fruit', 'value': 8}]

Функция обходит все вложенные списки и словари, собирая подходящие элементы.

Пример 4: Использование json.loads с object_hook для преобразования ключей

Параметр object_hook позволяет кастомизировать десериализацию каждого JSON-объекта. Например, привести все ключи к нижнему регистру.

import json

def to_lower(dct):
    return {k.lower(): v for k, v in dct.items()}

json_str = '{"UserName": "Петр", "UserAge": 40}'
data = json.loads(json_str, object_hook=to_lower)
print(data)  # {'username': 'Петр', 'userage': 40}

Этот подход полезен для нормализации данных из разных источников.

Пример 5: Работа с вложенными JSON в асинхронном коде

Для асинхронного парсинга используйте aiojson или aiofiles с обычным json.

import aiofiles
import json

async def read_json(path):
    async with aiofiles.open(path, 'r') as f:
        content = await f.read()
        return json.loads(content)
# Пример вызова: data = await read_json('data.json')

Асинхронное чтение не блокирует цикл событий, что важно для высоконагруженных приложений.

Пример 6: Сериализация сложных объектов с вложенными dataclass в JSON

Используйте dataclasses с кастомным кодировщиком.

import json
from dataclasses import dataclass, asdict

@dataclass
class Address:
    city: str
    street: str

@dataclass
class Person:
    name: str
    address: Address

def custom_encoder(obj):
    if hasattr(obj, '__dict__'):
        return asdict(obj)
    raise TypeError

person = Person("Анна", Address("Москва", "Тверская"))
json_str = json.dumps(person, default=custom_encoder, ensure_ascii=False)
print(json_str)  # {"name": "Анна", "address": {"city": "Москва", "street": "Тверская"}}

Метод asdict рекурсивно преобразует dataclass в словарь.

Пример 7: Валидация вложенного JSON с помощью JSON Schema

Библиотека jsonschema проверяет соответствие заданной схеме.

import json
from jsonschema import validate, ValidationError

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "address": {
            "type": "object",
            "properties": {
                "city": {"type": "string"},
                "zip": {"type": "string", "pattern": "^\\d{5}$"}
            },
            "required": ["city"]
        }
    },
    "required": ["name"]
}

data = {"name": "Иван", "address": {"city": "Москва", "zip": "12345"}}
try:
    validate(instance=data, schema=schema)
    print("Валидация пройдена")
except ValidationError as e:
    print(e.message)

Результат - сообщение об ошибке, если данные не соответствуют схеме.