Модуль Pickle в Python: сериализация объектов для хранения данных

pickle.dump(obj, file, protocol=None, *, fix_imports=True, buffer_callback=None) # сериализует объект и записывает результат в файловый объект.
pickle.load(file, *, fix_imports=True, encoding="ASCII", errors="strict", buffers=None) # считывает сериализованный объект из файла и восстанавливает его.
pickle.dumps(obj, protocol=None, *, fix_imports=True, buffer_callback=None) # сериализует объект и возвращает байтовую строку.
pickle.loads(bytes_object, *, fix_imports=True, encoding="ASCII", errors="strict", buffers=None) # десериализует объект из байтовой строки.

import pickle

# Сериализация объекта в файл
data = {'name': 'Alice', 'scores': [98, 97, 95], 'active': True}
with open('data.pickle', 'wb') as file:
pickle.dump(data, file, protocol=pickle.HIGHEST_PROTOCOL)

# Десериализация объекта из файла
with open('data.pickle', 'rb') as file:
loaded_data = pickle.load(file)
print(loaded_data) # {'name': 'Alice', 'scores': [98, 97, 95], 'active': True}

# Сериализация в байтовую строку
serialized_data = pickle.dumps(data)
print(type(serialized_data)) # <class 'bytes'>

# Десериализация из байтовой строки
deserialized_data = pickle.loads(serialized_data)
print(deserialized_data) # {'name': 'Alice', 'scores': [98, 97, 95], 'active': True}

import pickle
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification

# Генерируем синтетический датасет для классификации
X, y = make_classification(n_samples=1000, n_features=20, random_state=42)

# Создаем и обучаем модель
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X, y)

# Сохраняем обученную модель на диск
with open('random_forest_model.pkl', 'wb') as file:
pickle.dump(model, file)

# Позже в другом сеансе или программе:
with open('random_forest_model.pkl', 'rb') as file:
loaded_model = pickle.load(file)

# Используем загруженную модель для предсказания
predictions = loaded_model.predict(X[:5])
print(predictions)

import pickle
import random

class GameState:
def __init__(self):
self.player_position = {'x': 0, 'y': 0}
self.score = 0
self.inventory = []
self.game_level = 1
self.enemies_defeated = 0

def move_player(self, dx, dy):
self.player_position['x'] += dx
self.player_position['y'] += dy

def add_to_inventory(self, item):
self.inventory.append(item)

def increase_score(self, points):
self.score += points

def __str__(self):
return f"Position: {self.player_position}, Score: {self.score}, Level: {self.game_level}"

# Создаем состояние игры и производим некоторые действия
game = GameState()
game.move_player(5, 10)
game.add_to_inventory("Sword")
game.add_to_inventory("Shield")
game.increase_score(100)
print(f"Current game state: {game}")

# Сохраняем состояние игры
with open('game_save.pkl', 'wb') as file:
pickle.dump(game, file)
print("Game saved!")

# Загружаем состояние игры
with open('game_save.pkl', 'rb') as file:
loaded_game = pickle.load(file)
print(f"Loaded game state: {loaded_game}")

import pickle
import time
import os
import hashlib

def expensive_computation(n):
"""Функция, имитирующая длительные вычисления."""
print(f"Computing result for n={n}...")
time.sleep(3) # Имитация долгой работы
return sum(i**2 for i in range(n))

def cached_computation(n, cache_file="computation_cache.pkl"):
# Создаем хэш для аргументов функции
args_hash = hashlib.md5(str(n).encode()).hexdigest()
cache_key = f"expensive_computation_{args_hash}"

# Проверяем, существует ли кэш
if os.path.exists(cache_file):
with open(cache_file, 'rb') as f:
try:
cache = pickle.load(f)
if cache_key in cache:
print(f"Found cached result for n={n}")
return cache[cache_key]
except (pickle.PickleError, EOFError):
# Если есть проблемы с кэшем, создаем новый
cache = {}
else:
cache = {}

# Вычисляем результат и сохраняем в кэше
result = expensive_computation(n)
cache[cache_key] = result

with open(cache_file, 'wb') as f:
pickle.dump(cache, f)

return result

# Первый вызов – вычисление будет выполнено
result1 = cached_computation(1000)
print(f"Result: {result1}")

# Второй вызов с теми же аргументами – результат будет взят из кэша
result2 = cached_computation(1000)
print(f"Result: {result2}")

# Вызов с другими аргументами – вычисление будет выполнено заново
result3 = cached_computation(2000)
print(f"Result: {result3}")

import pickle

class Person:
species = "Human" # Переменная класса

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

def add_skill(self, skill):
self.skills.append(skill)

def greet(self):
return f"Hello, my name is {self.name}!"

# Создаем экземпляр класса
alice = Person("Alice", 30)
alice.add_skill("Python")
alice.add_skill("Data Analysis")

# Сериализуем объект
with open('person.pkl', 'wb') as file:
pickle.dump(alice, file)

# Десериализуем объект
with open('person.pkl', 'rb') as file:
loaded_alice = pickle.load(file)

# Проверяем атрибуты и методы восстановленного объекта
print(loaded_alice.name) # Alice
print(loaded_alice.age) # 30
print(loaded_alice.skills) # ['Python', 'Data Analysis']
print(loaded_alice.species) # Human
print(loaded_alice.greet()) # Hello, my name is Alice!

import pickle
import datetime

class DatabaseConnection:
def __init__(self, host, user, password):
self.host = host
self.user = user
self.password = password
self.connection = None
self.connect()

def connect(self):
# Имитация подключения к базе данных
print(f"Connecting to database at {self.host}...")
self.connection = f"Connection to {self.host} established at {datetime.datetime.now()}"
self.connected_at = datetime.datetime.now()

def query(self, sql):
if not self.connection:
self.connect()
return f"Executing '{sql}' on {self.host}"

def __getstate__(self):
# Не сохраняем соединение и время подключения, только параметры
state = self.__dict__.copy()
del state['connection']
del state['connected_at']
return state

def __setstate__(self, state):
self.__dict__.update(state)
# Восстанавливаем соединение при десериализации
self.connection = None
self.connect()

# Создаем объект подключения к БД
db = DatabaseConnection("db.example.com", "user123", "password123")
print(db.query("SELECT * FROM users"))

# Сериализуем объект
with open('db_connection.pkl', 'wb') as file:
pickle.dump(db, file)

print("Database connection serialized")

# Десериализуем объект
with open('db_connection.pkl', 'rb') as file:
loaded_db = pickle.load(file)

# Проверяем восстановленное подключение
print(loaded_db.query("SELECT count(*) FROM products"))

import pickle

# Создаем сложную вложенную структуру данных
team_data = {
"name": "Data Science Team",
"members": [
Person("Bob", 35, ["Python", "Machine Learning", "Statistics"]),
Person("Carol", 28, ["Data Visualization", "SQL", "R"])
],
"projects": {
"customer_segmentation": {
"deadline": datetime.date(2023, 12, 31),
"priority": "high",
"resources": ["database", "compute_cluster"]
},
"recommendation_engine": {
"deadline": datetime.date(2024, 3, 15),
"priority": "medium",
"resources": ["gpu_server"]
}
},
"performance_metrics": [
[95, 87, 92], # Q1 scores
[88, 91, 94] # Q2 scores
]
}

# Сериализуем структуру
with open('team_data.pkl', 'wb') as file:
pickle.dump(team_data, file)

# Десериализуем структуру
with open('team_data.pkl', 'rb') as file:
loaded_team_data = pickle.load(file)

# Проверяем вложенные данные
print(f"Team: {loaded_team_data['name']}")
print(f"First member: {loaded_team_data['members'][0].name}")
print(f"First member skills: {loaded_team_data['members'][0].skills}")
print(f"Project deadline: {loaded_team_data['projects']['customer_segmentation']['deadline']}")

import pickle
import os

class MaliciousPayload:
def __reduce__(self):
# При десериализации будет выполнена команда system
cmd = "echo 'This could be a dangerous command' > warning.txt"
return os.system, (cmd,)

# Создаем вредоносную нагрузку
malicious = MaliciousPayload()

# Сериализуем ее
with open('malicious.pkl', 'wb') as file:
pickle.dump(malicious, file)

# При десериализации будет выполнена команда system!
with open('malicious.pkl', 'rb') as file:
pickle.load(file)

import pickle
import hmac
import hashlib

def secure_pickle_dump(obj, filename, key):
"""Безопасно сериализует объект с добавлением HMAC."""
pickled_data = pickle.dumps(obj)
digest = hmac.new(key, pickled_data, hashlib.sha256).digest()

with open(filename, 'wb') as file:
file.write(digest + pickled_data)

def secure_pickle_load(filename, key):
"""Безопасно загружает объект с проверкой HMAC."""
with open(filename, 'rb') as file:
data = file.read()

digest_size = 32 # SHA-256 produces 32 bytes
stored_digest = data[:digest_size]
pickled_data = data[digest_size:]

calculated_digest = hmac.new(key, pickled_data, hashlib.sha256).digest()

if not hmac.compare_digest(stored_digest, calculated_digest):
raise ValueError("Данные могли быть подделаны!")

return pickle.loads(pickled_data)

# Пример использования
secret_key = b'very-secret-key-not-for-production'
data = {"username": "alice", "permissions": ["read", "write"]}

# Сохраняем данные с защитой целостности
secure_pickle_dump(data, 'secure_data.pkl', secret_key)

# Загружаем с проверкой целостности
try:
loaded_data = secure_pickle_load('secure_data.pkl', secret_key)
print("Данные успешно загружены:", loaded_data)
except ValueError as e:
print("Ошибка:", str(e))