Продвинутая обработка JSON в Python: от основ до мастерства

import json

# Строка JSON
json_string = '{"name": "Alice", "age": 30, "is_developer": true, "skills": ["Python", "SQL", "Docker"]}'

# Преобразование в Python-объект
python_obj = json.loads(json_string)

# Теперь можем работать с данными как с Python-структурой
print(python_obj['name']) # Alice
print(python_obj['skills'][0]) # Python

try:
data = json.loads('{"broken_json": "missing quote}')
except json.JSONDecodeError as e:
print(f"Ошибка декодирования JSON: {e}")
# Можно выполнить альтернативную логику или записать в лог

nested_json = '''
{
"user": {
"profile": {
"contacts": {
"email": "user@example.com",
"phones": ["+1234567890", "+0987654321"]
}
}
}
}
'''

data = json.loads(nested_json)

# Безопасный доступ к вложенным данным
email = data.get("user", {}).get("profile", {}).get("contacts", {}).get("email")
print(email) # user@example.com

# Проверка наличия ключей перед обращением к вложенным элементам
if "user" in data and "profile" in data["user"] and "contacts" in data["user"]["profile"]:
phones = data["user"]["profile"]["contacts"].get("phones", [])
if phones:
print(f"Первый телефон: {phones[0]}")

def safe_get_nested(obj, path, default=None):
"""Безопасно извлекает значение по пути из вложенной структуры."""
keys = path.split('.')
result = obj

try:
for key in keys:
if isinstance(result, dict):
result = result.get(key)
else:
return default
return result
except (KeyError, TypeError, AttributeError):
return default

# Использование
email = safe_get_nested(data, 'user.profile.contacts.email')
first_phone = safe_get_nested(data, 'user.profile.contacts.phones.0')

class User:
def __init__(self, name, age, is_developer=False, skills=None):
self.name = name
self.age = age
self.is_developer = is_developer
self.skills = skills or []

def user_decoder(obj):
if 'name' in obj and 'age' in obj:
return User(obj['name'], obj['age'], obj.get('is_developer', False), obj.get('skills', []))
return obj

# Использование декодера
user_data = json.loads(json_string, object_hook=user_decoder)
print(type(user_data)) # <class '__main__.User'>
print(user_data.name) # Alice
print(user_data.skills) # ['Python', 'SQL', 'Docker']

import json

# Объект Python
python_obj = {
"name": "Bob",
"age": 25,
"is_developer": True,
"languages": ["Python", "JavaScript", "Go"],
"experience": {
"Python": 5,
"JavaScript": 3,
"Go": 1
}
}

# Сериализация в JSON-строку
json_string = json.dumps(python_obj)
print(json_string)
# {"name": "Bob", "age": 25, "is_developer": true, "languages": ["Python", "JavaScript", "Go"], "experience": {"Python": 5, "JavaScript": 3, "Go": 1}}

# Форматированный вывод для лучшей читаемости
formatted_json = json.dumps(python_obj, indent=4, sort_keys=True)
print(formatted_json)

{
"age": 25,
"experience": {
"Go": 1,
"JavaScript": 3,
"Python": 5
},
"is_developer": true,
"languages": [
"Python",
"JavaScript",
"Go"
],
"name": "Bob"
}

data = {"name": "Иван Петров", "city": "Москва"}

# По умолчанию не-ASCII символы экранируются
ascii_json = json.dumps(data)
print(ascii_json) # {"name": "\u0418\u0432\u0430\u043d \u041f\u0435\u0442\u0430\u0440\u043e\u0432", "city": "\u041c\u043e\u0441\u043a\u0432\u0430"}

# Отключение экранирования для читаемости
readable_json = json.dumps(data, ensure_ascii=False)
print(readable_json) # {"name": "Иван Петров", "city": "Москва"}

class Employee:
def __init__(self, name, position, salary):
self.name = name
self.position = position
self.salary = salary

def to_dict(self):
return {
"name": self.name,
"position": self.position,
"salary": self.salary
}

# Функция для сериализации объектов Employee
def employee_serializer(obj):
if isinstance(obj, Employee):
return obj.to_dict()
# Для остальных типов, которые не поддерживаются
raise TypeError(f"Object of type {type(obj)} is not JSON serializable")

employee = Employee("Alice Smith", "Senior Developer", 120000)

# Использование custom serializer
json_employee = json.dumps(employee, default=employee_serializer, indent=2)
print(json_employee)

from datetime import datetime, date
from decimal import Decimal

class CustomEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return {"__datetime__": obj.isoformat()}
elif isinstance(obj, date):
return {"__date__": obj.isoformat()}
elif isinstance(obj, Decimal):
return {"__decimal__": str(obj)}
elif isinstance(obj, Employee):
return {"__employee__": obj.to_dict()}
# Добавляем поддержку других типов
return super().default(obj)

complex_data = {
"employee": employee,
"hire_date": date(2020, 5, 15),
"last_promotion": datetime(2022, 3, 10, 14, 30),
"current_salary": Decimal("120000.50")
}

json_complex = json.dumps(complex_data, cls=CustomEncoder, indent=2)
print(json_complex)

import json

# Открываем файл и загружаем JSON
with open('data.json', 'r', encoding='utf-8') as file:
data = json.load(file)

# Теперь с данными можно работать как с обычным объектом Python
print(data['name'])

import json

data = {
'name': 'Alice',
'age': 30,
'skills': ['Python', 'SQL', 'Docker'],
'is_active': True
}

# Сохраняем в файл с форматированием
with open('output.json', 'w', encoding='utf-8') as file:
json.dump(data, file, indent=4, ensure_ascii=False)

import ijson

# Обрабатываем большой файл построчно без загрузки в память целиком
with open('large_data.json', 'rb') as file:
# Получаем итератор по всем элементам в массиве "items"
items = ijson.items(file, 'items.item')

# Обрабатываем каждый элемент отдельно
for item in items:
process_item(item) # Какая-то функция обработки

def process_large_json(filename, chunk_size=1000):
"""Обработка большого JSON-файла порциями."""
with open(filename, 'r', encoding='utf-8') as f:
# Предполагаем, что корневой элемент – массив
data = json.load(f)

total_items = len(data)

# Обрабатываем по частям
for i in range(0, total_items, chunk_size):
chunk = data[i:i + chunk_size]
process_chunk(chunk) # Функция обработки порции данных

# Для освобождения памяти
del chunk

# Освобождаем память
del data

import json
import os

def load_json_safely(filepath, default=None):
"""Безопасная загрузка JSON из файла с обработкой ошибок."""
try:
if not os.path.exists(filepath):
return default

with open(filepath, 'r', encoding='utf-8') as f:
return json.load(f)

except json.JSONDecodeError:
print(f"Ошибка: {filepath} содержит некорректный JSON")
return default

except (IOError, PermissionError) as e:
print(f"Ошибка доступа к файлу {filepath}: {e}")
return default

# Использование
config = load_json_safely('config.json', default={})

import json
import os
import tempfile
import shutil

def atomic_json_dump(data, filepath, **kwargs):
"""Атомарная запись в JSON-файл для предотвращения повреждения при сбоях."""
# Создаем временный файл в той же директории
directory = os.path.dirname(os.path.abspath(filepath))
fd, temp_path = tempfile.mkstemp(dir=directory)

try:
with os.fdopen(fd, 'w', encoding='utf-8') as f:
json.dump(data, f, **kwargs)
f.flush()
os.fsync(f.fileno())

# Атомарная замена на POSIX-системах
os.replace(temp_path, filepath)
except:
# Удаляем временный файл в случае ошибки
os.unlink(temp_path)
raise

# Использование
data = {"config": "value", "updated_at": "2023-07-25"}
atomic_json_dump(data, 'config.json', indent=2)

import json
import ujson
import orjson
import time

data = {
"records": [{"id": i, "value": f"test{i}"} for i in range(100000)]
}

# Тестирование скорости сериализации
start = time.time()
json_str = json.dumps(data)
print(f"json: {time.time() – start:.4f}s")

start = time.time()
ujson_str = ujson.dumps(data)
print(f"ujson: {time.time() – start:.4f}s")

start = time.time()
orjson_str = orjson.dumps(data)
print(f"orjson: {time.time() – start:.4f}s")

# Тестирование скорости десериализации
start = time.time()
_ = json.loads(json_str)
print(f"json loads: {time.time() – start:.4f}s")

start = time.time()
_ = ujson.loads(ujson_str)
print(f"ujson loads: {time.time() – start:.4f}s")

start = time.time()
_ = orjson.loads(orjson_str)
print(f"orjson loads: {time.time() – start:.4f}s")

from jsonschema import validate, ValidationError

# Определяем схему JSON
user_schema = {
"type": "object",
"required": ["username", "email"],
"properties": {
"username": {"type": "string", "minLength": 3},
"email": {"type": "string", "format": "email"},
"age": {"type": "integer", "minimum": 18},
"roles": {
"type": "array",
"items": {"type": "string", "enum": ["admin", "user", "editor"]}
}
}
}

# Данные для проверки
user_data = {
"username": "john_doe",
"email": "john@example.com",
"age": 25,
"roles": ["user", "editor"]
}

try:
validate(instance=user_data, schema=user_schema)
print("Валидация успешна")
except ValidationError as e:
print(f"Ошибка валидации: {e}")

from jsonpath_ng import parse

# Сложная структура JSON
data = {
"store": {
"books": [
{
"title": "Война и мир",
"author": "Лев Толстой",
"price": 12.99
},
{
"title": "Гарри Поттер",
"author": "Дж. К. Роулинг",
"price": 8.99
}
],
"bicycles": [
{
"model": "Mountain Bike",
"price": 399.99
}
]
}
}

# Запросы с использованием JSONPath
# Найти все книги с ценой меньше 10
jsonpath_expr = parse('$.store.books[?(@.price < 10)]')
results = [match.value for match in jsonpath_expr.find(data)]
print("Книги дешевле 10:")
for book in results:
print(f" – {book['title']} (${book['price']})")

# Найти все цены в магазине
price_expr = parse('$..price')
prices = [match.value for match in price_expr.find(data)]
print(f"Все цены: {prices}")

from jsonpatch import JsonPatch, make_patch

# Исходный документ
original = {
"name": "John",
"address": {
"city": "New York",
"zipcode": "10001"
},
"phones": ["212-555-1234", "646-555-4567"]
}

# Обновлённый документ
updated = {
"name": "John",
"address": {
"city": "San Francisco",
"zipcode": "94107"
},
"phones": ["212-555-1234", "415-555-6789"]
}

# Создаём патч – разницу между документами
patch = make_patch(original, updated)
print("JSON Patch:")
print(patch)

# Применяем патч к оригинальному документу
patched = patch.apply(original)
print("\nПосле применения патча:")
print(patched)

# Создание патча вручную
manual_patch = JsonPatch([
{"op": "replace", "path": "/address/city", "value": "Los Angeles"},
{"op": "replace", "path": "/address/zipcode", "value": "90001"},
{"op": "replace", "path": "/phones/1", "value": "213-555-8910"}
])

# Применяем ручной патч
new_patched = manual_patch.apply(patched)
print("\nПосле применения ручного патча:")
print(new_patched)

# Потоковая генерация JSON (построчный вывод)
def stream_json_records(records_generator, output_file):
"""Записывает записи в файл JSON построчно без загрузки всех в память."""
with open(output_file, 'w', encoding='utf-8') as f:
f.write('[\n') # Начало JSON-массива

first = True
for record in records_generator:
if not first:
f.write(',\n')
else:
first = False

json_record = json.dumps(record, ensure_ascii=False)
f.write(' ' + json_record)

f.write('\n]') # Конец JSON-массива

# Генератор записей (пример)
def generate_sample_records(count):
"""Генерирует тестовые записи без загрузки всех в память."""
for i in range(count):
yield {
"id": i,
"name": f"Record {i}",
"value": i * 10
}

# Использование
stream_json_records(generate_sample_records(1000000), 'large_output.json')

from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import json

Base = declarative_base()

class Product(Base):
__tablename__ = 'products'

id = Column(Integer, primary_key=True)
name = Column(String)
price = Column(Float)
category = Column(String)

def to_dict(self):
return {
'id': self.id,
'name': self.name,
'price': self.price,
'category': self.category
}

@staticmethod
def from_dict(data):
return Product(
id=data.get('id'),
name=data.get('name'),
price=data.get('price'),
category=data.get('category')
)

# Создание ORM-сериализатора
class ORMJSONEncoder(json.JSONEncoder):
def default(self, obj):
if hasattr(obj, 'to_dict'):
return obj.to_dict()
return super().default(obj)

# Пример использования
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# Создаем продукты
products = [
Product(name='Laptop', price=999.99, category='Electronics'),
Product(name='Headphones', price=129.99, category='Audio')
]
session.add_all(products)
session.commit()

# Получаем продукты и сериализуем
all_products = session.query(Product).all()
json_products = json.dumps(all_products, cls=ORMJSONEncoder, indent=2)
print(json_products)

# Десериализация в объекты ORM
json_data = '''
{
"name": "Smartphone",
"price": 799.99,
"category": "Electronics"
}
'''
new_product = Product.from_dict(json.loads(json_data))
session.add(new_product)
session.commit()

print(f"New product ID: {new_product.id}")

from dataclasses import dataclass
from typing import List, Optional
import json

# Класс с использованием __slots__ для экономии памяти
class CompactUser:
__slots__ = ('id', 'name', 'email')

def __init__(self, id: int, name: str, email: str):
self.id = id
self.name = name
self.email = email

def to_dict(self):
return {
'id': self.id,
'name': self.name,
'email': self.email
}

# Альтернатива с использованием dataclass
@dataclass
class User:
id: int
name: str
email: str
active: bool = True
roles: Optional[List[str]] = None

# Создаем сериализатор для dataclass
def dataclass_json_serializer(obj):
if hasattr(obj, '__dataclass_fields__'):
return {field: getattr(obj, field) for field in obj.__dataclass_fields__}
elif hasattr(obj, 'to_dict'):
return obj.to_dict()
raise TypeError(f"Object of type {type(obj)} is not JSON serializable")

# Пример использования
compact_user = CompactUser(1, "John Doe", "john@example.com")
user = User(2, "Jane Smith", "jane@example.com", roles=["admin", "user"])

# Сериализация
json_compact = json.dumps(compact_user, default=dataclass_json_serializer)
json_user = json.dumps(user, default=dataclass_json_serializer)

print(f"CompactUser: {json_compact}")
print(f"User: {json_user}")