Практическое руководство: как сериализовать данные в Python

Расширенные примеры и нестандартные сценарии

Здесь приведены примеры, которые выходят за рамки базового использования, но могут оказаться полезными в реальных проектах.

Кастомный JSON-кодировщик для дат, десятичных чисел и dataclass

from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
import json

@dataclass
class Product:
    name: str
    price: Decimal
    created: datetime

class CustomEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.isoformat()
        if isinstance(obj, Decimal):
            return float(obj)
        if isinstance(obj, Product):
            return {'name': obj.name, 'price': obj.price, 'created': obj.created}
        return super().default(obj)

product = Product(name="Laptop", price=Decimal('1299.99'), created=datetime.now())
json_str = json.dumps(product, cls=CustomEncoder, indent=2, ensure_ascii=False)
print(json_str)

{
  "name": "Laptop",
  "price": 1299.99,
  "created": "2025-03-28T12:34:56.789123"
}

Пояснение: класс CustomEncoder переопределяет метод default, который вызывается для объектов, не поддерживаемых стандартным кодировщиком. В примере обработаны три типа. Для dataclass можно также использовать библиотеку dataclasses_json.

Потоковая обработка большого XML-файла с iterparse

import xml.etree.ElementTree as ET

# Предположим, есть большой файл large.xml с тысячами элементов 
def process_items(filepath):
    for event, elem in ET.iterparse(filepath, events=('end',)):
        if elem.tag == 'item':
            # извлекаем данные
            name = elem.findtext('name')
            price = elem.findtext('price')
            print(f"Item: {name}, Price: {price}")
            # очищаем обработанный элемент, чтобы не накапливать память
            elem.clear()
        elif elem.tag == 'root':
            break

# Использование: process_items('large.xml')

Пояснение: функция iterparse генерирует события в процессе разбора, не загружая весь документ в память. После обработки элемента вызов elem.clear() освобождает память. Это важно для файлов размером сотни мегабайт и более.

Проблема: при неверном использовании (например, отсутствие clear) память может расти. Также необходимо обрабатывать события в правильном порядке.

YAML с пользовательскими тегами для безопасной десериализации

import yaml
from datetime import datetime

def datetime_representer(dumper, data):
    return dumper.represent_scalar('tag:yaml.org,2002:timestamp', data.isoformat())

yaml.add_representer(datetime, datetime_representer)

data = {'event': 'meeting', 'timestamp': datetime.now()}
yaml_str = yaml.dump(data, allow_unicode=True)
print(yaml_str)

event: meeting
timestamp: 2025-03-28T12:34:56.789123

Загрузка с кастомным конструктором (безопасно через SafeLoader):

def datetime_constructor(loader, node):
    value = loader.construct_scalar(node)
    return datetime.fromisoformat(value)

yaml.add_constructor('tag:yaml.org,2002:timestamp', datetime_constructor, Loader=yaml.SafeLoader)
loaded = yaml.safe_load(yaml_str)
print(loaded)

{'event': 'meeting', 'timestamp': datetime.datetime(2025, 3, 28, 12, 34, 56, 789123)}

Пояснение: добавление собственных representer и constructor расширяет возможности сериализации без потери безопасности. Всегда используйте SafeLoader для предотвращения выполнения кода.

Передача сложных объектов через socket с помощью pickle

import pickle
import socket

# Сервер (слушает порт)
def server():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(('localhost', 12345))
    s.listen(1)
    conn, addr = s.accept()
    data = conn.recv(4096)
    obj = pickle.loads(data)
    conn.send(pickle.dumps({'result': obj['x'] + obj['y']}))
    conn.close()

# Клиент
def client():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect(('localhost', 12345))
    obj = {'x': 10, 'y': 20, 'func': lambda: None}  # лямбда сериализуема
    s.send(pickle.dumps(obj))
    response = s.recv(4096)
    print(pickle.loads(response))
    s.close()

Пояснение: pickle может передавать даже лямбда-функции, что удобно для распределенных вычислений (например, с multiprocessing), но небезопасно в открытом Интернете. В реальном коде следует тщательно контролировать источник данных.

Проблема: если объект содержит ссылки на внешние ресурсы (например, открытый файл), pickle может их не корректно восстановить.

Использование JSON Lines для хранения записей в логах

import json

logs = [
    {"timestamp": "2025-03-28T10:00:00", "level": "INFO", "msg": "Started"},
    {"timestamp": "2025-03-28T10:01:00", "level": "ERROR", "msg": "Timeout"}
]

# Запись
with open("logs.jsonl", "w", encoding="utf-8") as f:
    for entry in logs:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")

# Чтение
with open("logs.jsonl", "r", encoding="utf-8") as f:
    for line in f:
        entry = json.loads(line.strip())
        print(entry)

{'timestamp': '2025-03-28T10:00:00', 'level': 'INFO', 'msg': 'Started'}
{'timestamp': '2025-03-28T10:01:00', 'level': 'ERROR', 'msg': 'Timeout'}

Пояснение: JSON Lines (каждая строка - отдельный JSON-объект) позволяет читать и дописывать файл без загрузки всего содержимого в память. Удобно для логов, потоковой передачи данных.