Сериализация и десериализация данных на Python

import pickle
 
# Создаём сложную структуру данных
data = {
    'name': 'Дмитрий',
    'skills': ['Python', 'PostgreSQL', 'Redis'],
    'experience': 5,
    'active': True
}
 
# Сериализуем в байтовую строку
serialized_data = pickle.dumps(data)
print(f"Сериализованные данные: {serialized_data[:30]}...")
 
# Десериализуем обратно
restored_data = pickle.loads(serialized_data)
print(f"Восстановленные данные: {restored_data}")

import json
 
# Та же структура данных
data = {
    'name': 'Дмитрий',
    'skills': ['Python', 'PostgreSQL', 'Redis'],
    'experience': 5,
    'active': True
}
 
# Сериализуем в строку JSON
json_data = json.dumps(data, indent=2)
print(f"JSON данные:\n{json_data}")
 
# Десериализуем обратно
restored_data = json.loads(json_data)
print(f"Восстановленные данные: {restored_data}")

import json
from datetime import datetime
 
class User:
    def __init__(self, name, created_at):
        self.name = name
        self.created_at = created_at
    
    def to_json(self):
        return {
            'name': self.name,
            'created_at': self.created_at.isoformat()
        }
    
    @classmethod
    def from_json(cls, data):
        return cls(
            name=data['name'],
            created_at=datetime.fromisoformat(data['created_at'])
        )
 
# Создаём объект
user = User('Анна', datetime.now())
 
# Сериализуем
user_json = json.dumps(user.to_json())
 
# Десериализуем
user_data = json.loads(user_json)
restored_user = User.from_json(user_data)
 
print(f"Оригинальное имя: {user.name}")
print(f"Восстановленное имя: {restored_user.name}")

pip install pyyaml

import yaml
 
# Конфигурация приложения
config = {
    'database': {
        'host': 'localhost',
        'port': 5432,
        'user': 'admin',
        'password': 'secret'  # В реальном коде используйте безопасные способы хранения
    },
    'api': {
        'enabled': True,
        'rate_limit': 100,
        'log_level': 'DEBUG'
    },
    'features': ['auth', 'reporting', 'export']
}
 
# Сериализуем в YAML
yaml_data = yaml.dump(config, default_flow_style=False)
print(f"YAML данные:\n{yaml_data}")
 
# Десериализуем обратно
restored_config = yaml.safe_load(yaml_data)
print(f"Восстановленная конфигурация: {restored_config}")

import shelve
 
# Открываем "полку" для хранения данных
with shelve.open('data_storage') as storage:
    # Сохраняем различные объекты
    storage['users'] = ['Алексей', 'Мария', 'Сергей']
    storage['config'] = {'debug': True, 'lang': 'ru'}
    storage['counter'] = 42
 
[H2]Файл закрывается автоматически благодаря контекстному менеджеру[/H2]
 
# Позже открываем и используем сохранённые данные
with shelve.open('data_storage') as storage:
    print(f"Пользователи: {storage['users']}")
    print(f"Конфигурация: {storage['config']}")
    storage['counter'] += 1
    print(f"Счетчик: {storage['counter']}")

import marshal
 
# Создаём некоторые данные
code_obj = compile("x = 42; print(x)", "example", "exec")
data = (1, 2, 3, "hello", {"key": "value"})
 
# Сериализуем с помощью marshal
serialized_code = marshal.dumps(code_obj)
serialized_data = marshal.dumps(data)
 
# Десериализуем
loaded_code = marshal.loads(serialized_code)
loaded_data = marshal.loads(serialized_data)
 
print(f"Тип загруженного кода: {type(loaded_code)}")
print(f"Загруженные данные: {loaded_data}")
 
# Запускаем десериализованный код
exec(loaded_code)

import msgpack
 
[H2]Устанавливаем через pip: pip install msgpack[/H2]
 
# Создаём данные
data = {
    'compact': True,
    'schema': 0,
    'payload': ['some', 'data', 123]
}
 
# Сериализуем с msgpack
packed = msgpack.packb(data)
print(f"Размер MessagePack: {len(packed)} байт")
 
# Для сравнения сериализуем с JSON
import json
json_data = json.dumps(data).encode('utf8')
print(f"Размер JSON: {len(json_data)} байт")
 
# Десериализуем
unpacked = msgpack.unpackb(packed)
print(f"Распакованные данные: {unpacked}")

[H2]Сначала нужно установить: pip install protobuf[/H2]
 
# Затем создаём файл .proto с описанием структуры данных:
# message Person {
#   required string name = 1;
#   required int32 id = 2;
#   optional string email = 3;
[H2]}[/H2]
 
# Затем компилируем его: protoc --python_out=. person.proto
[H2]Это создаст файл person_pb2.py[/H2]
 
# Теперь можем использовать сгенерированные классы
import person_pb2
 
# Создаём и заполняем объект
person = person_pb2.Person()
person.name = "Иван"
person.id = 123
person.email = "ivan@example.com"
 
# Сериализуем в бинарный формат
serialized = person.SerializeToString()
print(f"Размер сериализованных данных: {len(serialized)} байт")
 
# Десериализуем
new_person = person_pb2.Person()
new_person.ParseFromString(serialized)
print(f"Имя: {new_person.name}, ID: {new_person.id}")

import xml.etree.ElementTree as ET
 
# Создаём XML-структуру
root = ET.Element("data")
items = ET.SubElement(root, "items")
 
for i in range(3):
    item = ET.SubElement(items, "item")
    item.set("id", str(i))
    item.text = f"Item {i}"
 
# Сериализуем в строку
xml_string = ET.tostring(root, encoding='utf8').decode('utf8')
print(f"XML данные:\n{xml_string}")
 
# Десериализуем
tree = ET.ElementTree(ET.fromstring(xml_string))
for item in tree.findall(".//item"):
    print(f"ID: {item.get('id')}, Value: {item.text}")

import toml
 
[H2]Установите через pip: pip install toml[/H2]
 
config = """
[server]
host = "localhost"
port = 8080
 
[database]
user = "admin"
password = "secret"
hosts = ["localhost", "192.168.1.1"]
 
[[employees]]
name = "Иван"
position = "Разработчик"
 
[[employees]]
name = "Мария"
position = "Дизайнер"
"""
 
# Десериализуем
parsed = toml.loads(config)
print(f"Порт сервера: {parsed['server']['port']}")
print(f"Сотрудники: {[e['name'] for e in parsed['employees']]}")
 
# Сериализуем обратно
new_toml = toml.dumps(parsed)
print(f"Сгенерированный TOML:\n{new_toml}")

import pickle
 
class DatabaseConnection:
    def __init__(self, host, user, password):
        self.host = host
        self.user = user
        self.password = password
        # Устанавливаем соединение
        self.connection = f"Connection to {host} as {user}"
    
    def __getstate__(self):
        # Не сериализуем соединение
        state = self.__dict__.copy()
        del state['connection']
        return state
    
    def __setstate__(self, state):
        # Восстанавливаем объект и переустанавливаем соединение
        self.__dict__.update(state)
        self.connection = f"Connection to {self.host} as {self.user}"
 
conn = DatabaseConnection("localhost", "admin", "pass123")
pickled = pickle.dumps(conn)
new_conn = pickle.loads(pickled)
print(new_conn.connection)

import time
import pickle
import json
import yaml
import msgpack
import gzip
import os
 
def benchmark(func, *args, **kwargs):
    start_time = time.time()
    result = func(*args, **kwargs)
    end_time = time.time()
    return result, end_time - start_time
 
# Тестовые данные — словарь с разными типами
test_data = {
    'int': 42,
    'float': 3.14159,
    'str': 'Hello, World!' * 100,
    'list': list(range(1000)),
    'dict': {str(i): i for i in range(100)},
    'bool': True,
    'none': None
}

# Сериализация и замер размера
results = {}
 
# Pickle
pickle_data, pickle_serialize_time = benchmark(pickle.dumps, test_data)
_, pickle_deserialize_time = benchmark(pickle.loads, pickle_data)
results['pickle'] = {
    'serialize_time': pickle_serialize_time,
    'deserialize_time': pickle_deserialize_time,
    'size': len(pickle_data)
}
 
# JSON
json_data, json_serialize_time = benchmark(lambda d: json.dumps(d).encode('utf-8'), test_data)
_, json_deserialize_time = benchmark(lambda d: json.loads(d.decode('utf-8')), json_data)
results['json'] = {
    'serialize_time': json_serialize_time,
    'deserialize_time': json_deserialize_time,
    'size': len(json_data)
}
 
# YAML
yaml_data, yaml_serialize_time = benchmark(lambda d: yaml.dump(d).encode('utf-8'), test_data)
_, yaml_deserialize_time = benchmark(lambda d: yaml.safe_load(d.decode('utf-8')), yaml_data)
results['yaml'] = {
    'serialize_time': yaml_serialize_time,
    'deserialize_time': yaml_deserialize_time,
    'size': len(yaml_data)
}
 
# MessagePack
msgpack_data, msgpack_serialize_time = benchmark(msgpack.packb, test_data)
_, msgpack_deserialize_time = benchmark(msgpack.unpackb, msgpack_data)
results['msgpack'] = {
    'serialize_time': msgpack_serialize_time,
    'deserialize_time': msgpack_deserialize_time,
    'size': len(msgpack_data)
}
 
# Вывод результатов
for format_name, metrics in results.items():
    print(f"{format_name.ljust(10)}: {metrics['size']} bytes, "
          f"ser: {metrics['serialize_time']:.6f}s, "
          f"deser: {metrics['deserialize_time']:.6f}s")

pickle    : 10254 bytes, ser: 0.000320s, deser: 0.000215s
json      : 17842 bytes, ser: 0.001245s, deser: 0.000987s
yaml      : 18975 bytes, ser: 0.006521s, deser: 0.008745s
msgpack   : 9854 bytes, ser: 0.000412s, deser: 0.000654s

# Бенчмарк разных протоколов pickle
for protocol in range(6):  # От 0 до 5
    data, serialize_time = benchmark(pickle.dumps, test_data, protocol=protocol)
    _, deserialize_time = benchmark(pickle.loads, data)
    print(f"Pickle protocol {protocol}: {len(data)} bytes, "
          f"ser: {serialize_time:.6f}s, "
          f"deser: {deserialize_time:.6f}s")

def incremental_serialize(data_generator, filename, batch_size=1000):
    """Сериализует данные из генератора порциями"""
    count = 0
    with open(filename, 'wb') as f:
        batch = []
        for item in data_generator:
            batch.append(item)
            count += 1
            if len(batch) >= batch_size:
                pickle.dump(batch, f)
                batch = []
        if batch:  # Не забываем последнюю порцию
            pickle.dump(batch, f)
    return count
 
def incremental_deserialize(filename):
    """Десериализует данные порциями"""
    with open(filename, 'rb') as f:
        while True:
            try:
                batch = pickle.load(f)
                for item in batch:
                    yield item
            except EOFError:
                break
 
# Пример использования
def generate_big_data(n):
    for i in range(n):
        yield {'id': i, 'value': f'data_{i}'}
 
# Сериализуем миллион объектов порциями по 1000
count = incremental_serialize(generate_big_data(1_000_000), 'big_data.pkl')
print(f"Сериализовано {count} объектов")
 
# Десериализуем и используем только первые 10
for i, item in enumerate(incremental_deserialize('big_data.pkl')):
    print(item)
    if i >= 9:
        break

def compress_serialized(data, compression_level=9):
    """Сжимает сериализованные данные с помощью gzip"""
    return gzip.compress(data, compression_level)
 
def decompress_serialized(compressed_data):
    """Распаковывает сжатые данные"""
    return gzip.decompress(compressed_data)
 
# Сериализуем, сжимаем и сравниваем размеры
serialized = pickle.dumps(test_data)
compressed = compress_serialized(serialized)
 
print(f"Оригинальный размер: {len(serialized)} байт")
print(f"Сжатый размер: {len(compressed)} байт")
print(f"Степень сжатия: {len(serialized) / len(compressed):.2f}x")
 
# Распаковываем и десериализуем
decompressed = decompress_serialized(compressed)
restored_data = pickle.loads(decompressed)
print(f"Восстановленные данные идентичны оригиналу: {restored_data == test_data}")

import concurrent.futures
import asyncio
 
def serialize_in_thread(data):
    """Сериализует данные в отдельном потоке"""
    with concurrent.futures.ThreadPoolExecutor() as executor:
        future = executor.submit(pickle.dumps, data)
        return future.result()
 
async def serialize_async(data):
    """Асинхронно сериализует данные, не блокируя event loop"""
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, pickle.dumps, data)
 
# Пример использования в асинхронном коде
async def main():
    data = {'key': 'value', 'list': list(range(10000))}
    
    # Асинхронная сериализация
    serialized = await serialize_async(data)
    print(f"Async сериализация: {len(serialized)} байт")
    
    # Обработка нескольких задач параллельно
    tasks = [serialize_async(data) for _ in range(5)]
    results = await asyncio.gather(*tasks)
    print(f"Завершено {len(results)} задач параллельной сериализации")
 
# asyncio.run(main())  # Раскомментируйте для запуска

def memoized_serialize(obj, memo=None):
    """Сериализует объект с мемоизацией повторяющихся элементов"""
    if memo is None:
        memo = {}
    
    obj_id = id(obj)
    if obj_id in memo:
        return memo[obj_id]
    
    if isinstance(obj, dict):
        result = {memoized_serialize(k, memo): memoized_serialize(v, memo) 
                 for k, v in obj.items()}
    elif isinstance(obj, list):
        result = [memoized_serialize(item, memo) for item in obj]
    else:
        # Для простых типов просто возвращаем значение
        result = obj
    
    memo[obj_id] = result
    return result

import pickle
import base64
import os
 
# Вредоносная полезная нагрузка
class Evil:
    def __reduce__(self):
        # Этот код выполнится при десериализации
        return (os.system, ('echo "Я мог бы выполнить любую команду"',))
 
# Создаём вредоносные данные
malicious_data = pickle.dumps(Evil())
[H2]В реальной атаке эти данные могли бы прийти откуда угодно[/H2]
 
# Жертва десериализует данные
try:
    pickle.loads(malicious_data)
except Exception as e:
    print(f"Предотвращена атака: {e}")

import hmac
import pickle
 
SECRET_KEY = b'ваш-секретный-ключ'
 
def secure_dumps(obj):
    """Сериализует и подписывает объект"""
    pickled = pickle.dumps(obj)
    signature = hmac.new(SECRET_KEY, pickled, 'sha256').digest()
    return signature + pickled
 
def secure_loads(data):
    """Проверяет подпись и десериализует объект"""
    signature, pickled = data[:32], data[32:]
    expected_sig = hmac.new(SECRET_KEY, pickled, 'sha256').digest()
    if not hmac.compare_digest(signature, expected_sig):
        raise ValueError("Некорректная подпись данных - возможна атака!")
    return pickle.loads(pickled)

class RestrictedUnpickler(pickle.Unpickler):
    def find_class(self, module, name):
        # Разрешаем только безопасные классы
        if module == "builtins" and name in ("int", "float", "str", "list", "dict", "tuple"):
            return getattr(__import__(module), name)
        raise pickle.UnpicklingError(f"Запрещено десерилизовать {module}.{name}")
 
def safe_loads(data):
    return RestrictedUnpickler(io.BytesIO(data)).load()

import numpy as np
 
def serialize_numpy_array(arr):
    """Эффективная сериализация numpy массивов"""
    # Сохраняем форму, тип данных и содержимое
    buffer = io.BytesIO()
    np.savez_compressed(buffer, data=arr)
    buffer.seek(0)
    return buffer.read()
 
def deserialize_numpy_array(data):
    """Десерилизация numpy массивов"""
    buffer = io.BytesIO(data)
    loaded = np.load(buffer)
    return loaded['data']

from multiprocessing import Pool
import pickle
 
def parallel_serialize(data_list, processes=4):
    """Параллельная сериализация списка объектов"""
    with Pool(processes) as pool:
        return pool.map(pickle.dumps, data_list)
 
# Пример использования
large_data = [{"id": i, "data": list(range(1000))} for i in range(100)]
serialized = parallel_serialize(large_data)

import pickle
import os
from contextlib import contextmanager
 
@contextmanager
def saved_state(filename, default=None):
    """Контекстный менеджер для сохранения состояния приложения."""
    state = default
    if os.path.exists(filename):
        try:
            with open(filename, 'rb') as f:
                state = pickle.load(f)
        except Exception as e:
            print(f"Ошибка загрузки состояния: {e}")
    
    try:
        yield state
    finally:
        with open(filename, 'wb') as f:
            pickle.dump(state, f)
 
# Пример использования
with saved_state('app_state.pkl', default={}) as state:
    state['counter'] = state.get('counter', 0) + 1
    print(f"Приложение запущено {state['counter']} раз")
    
    # Работа с состоянием
    if 'last_run' not in state:
        state['last_run'] = []
    state['last_run'].append(time.time())

import json
from datetime import datetime, date
import uuid
 
class ComplexJSONEncoder(json.JSONEncoder):
    """Расширенный JSON энкодер для работы со сложными типами."""
    def default(self, obj):
        if isinstance(obj, datetime):
            return {"_type": "datetime", "iso": obj.isoformat()}
        elif isinstance(obj, date):
            return {"_type": "date", "iso": obj.isoformat()}
        elif isinstance(obj, uuid.UUID):
            return {"_type": "uuid", "hex": obj.hex}
        elif hasattr(obj, "to_json"):
            # Для объектов, реализующих метод to_json
            return obj.to_json()
        return super().default(obj)
 
def complex_json_decoder(json_dict):
    """Декодер для восстановления спецальных типов из JSON."""
    if "_type" in json_dict:
        obj_type = json_dict["_type"]
        if obj_type == "datetime":
            return datetime.fromisoformat(json_dict["iso"])
        elif obj_type == "date":
            return date.fromisoformat(json_dict["iso"])
        elif obj_type == "uuid":
            return uuid.UUID(json_dict["hex"])
    return json_dict
 
# Пример использования
data = {
    "id": uuid.uuid4(),
    "created_at": datetime.now(),
    "expires_on": date.today()
}
 
# Сериализация
json_str = json.dumps(data, cls=ComplexJSONEncoder, indent=2)
print(json_str)
 
# Десериализация
restored = json.loads(json_str, object_hook=complex_json_decoder)
print(f"Восстановленный UUID: {restored['id']}")
print(f"Восстановленная дата: {restored['expires_on']}")

import json
 
def process_large_json_file(input_file, output_file, transform_func):
    """Потоковая обработка большого JSON-файла с преобразованием каждого объекта."""
    with open(input_file, 'r') as fin, open(output_file, 'w') as fout:
        # Читаем открывающую скобку
        assert fin.read(1) == '['
        
        # Запишем открывающую скобку в выходной файл
        fout.write('[')
        
        first_item = True
        buffer = ''
        bracket_count = 0
        in_string = False
        escape_next = False
        
        while True:
            char = fin.read(1)
            if not char:  # Конец файла
                break
                
            buffer += char
            
            # Отслеживаем, находимся ли мы внутри строки
            if char == '"' and not escape_next:
                in_string = not in_string
            elif char == '\\' and in_string:
                escape_next = True
                continue
            
            if not in_string:
                if char == '{':
                    bracket_count += 1
                elif char == '}':
                    bracket_count -= 1
                    
                    # Если закрыли объект верхнего уровня
                    if bracket_count == 0:
                        # Обработаем объект
                        obj = json.loads(buffer)
                        transformed = transform_func(obj)
                        
                        # Запишем разделитель, если это не первый объект
                        if not first_item:
                            fout.write(',')
                        first_item = False
                        
                        # Запишем преобразованный объект
                        fout.write(json.dumps(transformed))
                        buffer = ''
            
            escape_next = False
            
        # Закроем массив
        fout.write(']')
 
# Пример использования
def transform_user(user):
    """Пример функции преобразования."""
    # Добавим вычисляемое поле
    if 'first_name' in user and 'last_name' in user:
        user['full_name'] = f"{user['first_name']} {user['last_name']}"
    return user
 
# Обработка файла с миллионами пользователей
process_large_json_file('users.json', 'transformed_users.json', transform_user)

class VersionedModel:
    __schema_version__ = 1
    
    def to_dict(self):
        """Преобразует объект в словарь для сериализации."""
        data = {
            "__schema_version__": self.__schema_version__,
            # Данные конкретной версии
        }
        # Заполняем данные в соответсвии с текущей версией
        data.update(self._to_dict_v1())
        return data
    
    def _to_dict_v1(self):
        """Реализация сериализации для версии 1."""
        raise NotImplementedError()
    
    @classmethod
    def from_dict(cls, data):
        """Создаёт объект из словаря с учётом версии."""
        version = data.get("__schema_version__", 1)
        
        if version == 1:
            obj = cls._from_dict_v1(data)
        else:
            raise ValueError(f"Неизвестная версия схемы: {version}")
        
        return obj
    
    @classmethod
    def _from_dict_v1(cls, data):
        """Реализация десериализации для версии 1."""
        raise NotImplementedError()
 
# Пример использования
class User(VersionedModel):
    __schema_version__ = 2
    
    def __init__(self, username, email=None, display_name=None):
        self.username = username
        self.email = email
        self.display_name = display_name
    
    def _to_dict_v1(self):
        """Первая версия формата."""
        return {
            "username": self.username,
            "email": self.email
        }
    
    def _to_dict_v2(self):
        """Вторая версия добавляет display_name."""
        data = self._to_dict_v1()
        data["display_name"] = self.display_name
        return data
    
    def to_dict(self):
        """Переопределяем для поддержки новой версии."""
        data = {
            "__schema_version__": self.__schema_version__,
        }
        # Используем последнюю версию сериализации
        data.update(self._to_dict_v2())
        return data
    
    @classmethod
    def _from_dict_v1(cls, data):
        """Десериализация из формата v1."""
        return cls(
            username=data["username"],
            email=data["email"]
        )
    
    @classmethod
    def _from_dict_v2(cls, data):
        """Десериализация из формата v2."""
        user = cls._from_dict_v1(data)
        user.display_name = data.get("display_name")
        return user
    
    @classmethod
    def from_dict(cls, data):
        """Переопределяем для поддержки разных версий."""
        version = data.get("__schema_version__", 1)
        
        if version == 1:
            return cls._from_dict_v1(data)
        elif version == 2:
            return cls._from_dict_v2(data)
        else:
            raise ValueError(f"Неизвестная версия схемы: {version}")

from rest_framework import serializers
from myapp.models import User
 
class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['id', 'username', 'email', 'is_active']
        read_only_fields = ['id']

def deserialize_user_data(json_data):
    try:
        data = json.loads(json_data)
    except json.JSONDecodeError:
        raise ValueError("Некорректный формат JSON")
    
    # Проверяем обязательные поля
    required_fields = ['username', 'email']
    for field in required_fields:
        if field not in data:
            raise ValueError(f"Отсутствует обязательное поле: {field}")
    
    # Валидируем email
    email = data['email']
    if not re.match(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$', email):
        raise ValueError(f"Некорректный email: {email}")
    
    return data

from pydantic import BaseModel, EmailStr, validator
 
class UserData(BaseModel):
    username: str
    email: EmailStr
    age: int = None  # Опциональное поле
    
    @validator('username')
    def username_must_be_valid(cls, v):
        if len(v) < 3:
            raise ValueError('Имя пользователя должно содержать не менее 3 символов')
        return v
 
# Десериализация с автоматической валидацией
try:
    user = UserData.parse_raw('{"username": "al", "email": "not-an-email"}')
except Exception as e:
    print(f"Ошибка валидации: {e}")