Python Pickle模块:数据存储

import pickle

# 准备要保存的数据
user_data = {
    '姓名': '张三',
    '年龄': 28,
    '技能': ['Python', 'JavaScript', 'SQL'],
    '注册时间': '2023-01-15'
}

# 保存到文件
with open('user_data.pkl', 'wb') as file:
    pickle.dump(user_data, file)

print("数据保存完成")

import pickle

# 从文件读取数据
with open('user_data.pkl', 'rb') as file:
    loaded_data = pickle.load(file)

print("读取的数据:")
print(loaded_data)

读取的数据:
{'姓名': '张三', '年龄': 28, '技能': ['Python', 'JavaScript', 'SQL'], '注册时间': '2023-01-15'}

import pickle

# 各种数据类型的示例
sample_data = {
    '列表': [1, 2, 3, 4, 5],
    '字典': {'a': 1, 'b': 2},
    '元组': (10, 20, 30),
    '集合': {1, 2, 3, 4, 5},
    '字符串': "Hello, World!",
    '数字': 3.14159,
    '布尔值': True,
    '空值': None
}

# 保存数据
with open('all_data.pkl', 'wb') as file:
    pickle.dump(sample_data, file)

# 读取数据
with open('all_data.pkl', 'rb') as file:
    restored_data = pickle.load(file)

print("恢复的数据类型:")
for key, value in restored_data.items():
    print(f"{key}: {value} (类型: {type(value).__name__})")

import pickle

# 创建数据
shopping_list = ['苹果', '香蕉', '牛奶', '面包']

# 转换为字节数据
bytes_data = pickle.dumps(shopping_list)
print(f"字节数据长度: {len(bytes_data)} 字节")
print(f"字节数据: {bytes_data}")

# 从字节数据恢复
restored_list = pickle.loads(bytes_data)
print(f"恢复的列表: {restored_list}")

import pickle
import socket

def send_data_over_network(data, host='localhost', port=8888):
    """模拟网络发送数据"""
    # 序列化为字节
    serialized_data = pickle.dumps(data)
    
    # 这里模拟网络发送（实际使用时需要socket）
    print(f"准备发送 {len(serialized_data)} 字节数据")
    
    # 模拟接收端
    received_data = pickle.loads(serialized_data)
    return received_data

# 测试网络传输
test_data = {
    '消息类型': '用户登录',
    '用户名': '李四',
    '时间戳': '2024-01-20 10:30:00'
}

result = send_data_over_network(test_data)
print(f"接收到的数据: {result}")

import pickle

class Student:
    def __init__(self, name, age, grades):
        self.name = name
        self.age = age
        self.grades = grades
    
    def display_info(self):
        return f"学生: {self.name}, 年龄: {self.age}, 成绩: {self.grades}"

# 创建学生对象
student1 = Student("王小明", 18, {"数学": 90, "语文": 85, "英语": 92})

# 保存对象
with open('student.pkl', 'wb') as file:
    pickle.dump(student1, file)

# 读取对象
with open('student.pkl', 'rb') as file:
    loaded_student = pickle.load(file)

print(loaded_student.display_info())

import pickle
from datetime import datetime

class User:
    def __init__(self, username, email, created_at=None):
        self.username = username
        self.email = email
        self.created_at = created_at or datetime.now()
        self._password = None  # 敏感数据
    
    def set_password(self, password):
        self._password = password
    
    def __getstate__(self):
        """自定义序列化内容"""
        state = self.__dict__.copy()
        # 不保存密码
        if '_password' in state:
            del state['_password']
        return state
    
    def __setstate__(self, state):
        """自定义反序列化内容"""
        self.__dict__.update(state)
        # 恢复时设置默认值
        if '_password' not in self.__dict__:
            self._password = None
    
    def __str__(self):
        return f"用户: {self.username}, 邮箱: {self.email}, 注册时间: {self.created_at}"

# 创建用户
user = User("testuser", "test@example.com")
user.set_password("secret123")

# 序列化
with open('user.pkl', 'wb') as file:
    pickle.dump(user, file)

# 反序列化
with open('user.pkl', 'rb') as file:
    restored_user = pickle.load(file)

print(restored_user)
print(f"密码字段: {hasattr(restored_user, '_password')}")

import pickle

data = list(range(10000))

# 测试不同协议版本
protocols = [0, 1, 2, 3, 4, 5]
available_protocols = []

for protocol in protocols:
    try:
        # 尝试使用该协议序列化
        bytes_data = pickle.dumps(data, protocol=protocol)
        size = len(bytes_data)
        available_protocols.append((protocol, size))
        print(f"协议 {protocol}: {size} 字节")
    except Exception as e:
        print(f"协议 {protocol} 不可用: {e}")

# 使用最高效的协议
if available_protocols:
    best_protocol = max(available_protocols, key=lambda x: x[0])[0]
    print(f"\n推荐使用协议版本: {best_protocol}")
    print(f"最高协议版本: {pickle.HIGHEST_PROTOCOL}")
    print(f"默认协议版本: {pickle.DEFAULT_PROTOCOL}")

import pickle
import os

class AppConfig:
    def __init__(self):
        self.theme = "dark"
        self.language = "zh-CN"
        self.font_size = 14
        self.recent_files = []
    
    def save_config(self, filename='config.pkl'):
        """保存配置到文件"""
        with open(filename, 'wb') as file:
            pickle.dump(self, file)
        print("配置已保存")
    
    @classmethod
    def load_config(cls, filename='config.pkl'):
        """从文件加载配置"""
        if os.path.exists(filename):
            with open(filename, 'rb') as file:
                return pickle.load(file)
        else:
            print("配置文件不存在，创建默认配置")
            return cls()

# 使用示例
config = AppConfig()
config.recent_files = ['file1.py', 'file2.txt', 'file3.jpg']
config.save_config()

# 程序重启后加载配置
new_config = AppConfig.load_config()
print(f"主题: {new_config.theme}")
print(f"最近文件: {new_config.recent_files}")

import pickle
import time
import hashlib
import os

class DataCache:
    def __init__(self, cache_dir='cache'):
        self.cache_dir = cache_dir
        if not os.path.exists(cache_dir):
            os.makedirs(cache_dir)
    
    def _get_cache_key(self, data):
        """生成缓存键"""
        data_str = str(data).encode('utf-8')
        return hashlib.md5(data_str).hexdigest()
    
    def set(self, key, data, expire_seconds=3600):
        """设置缓存"""
        cache_key = self._get_cache_key(key)
        cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl")
        
        cache_data = {
            'data': data,
            'expire_time': time.time() + expire_seconds,
            'created_time': time.time()
        }
        
        with open(cache_file, 'wb') as file:
            pickle.dump(cache_data, file)
    
    def get(self, key):
        """获取缓存"""
        cache_key = self._get_cache_key(key)
        cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl")
        
        if not os.path.exists(cache_file):
            return None
        
        try:
            with open(cache_file, 'rb') as file:
                cache_data = pickle.load(file)
            
            # 检查是否过期
            if time.time() > cache_data['expire_time']:
                os.remove(cache_file)  # 删除过期缓存
                return None
            
            return cache_data['data']
        
        except Exception as e:
            print(f"读取缓存失败: {e}")
            return None

# 使用缓存系统
cache = DataCache()

# 模拟耗时计算
def expensive_calculation(n):
    print(f"执行计算: {n}")
    time.sleep(1)  # 模拟耗时操作
    return n * n

# 使用缓存
number = 5
result = cache.get(number)

if result is None:
    result = expensive_calculation(number)
    cache.set(number, result, expire_seconds=30)  # 缓存30秒
    print("计算结果已缓存")
else:
    print("从缓存获取结果")

print(f"结果: {result}")

import pickle

def safe_pickle_load(filename):
    """安全地加载pickle文件"""
    try:
        with open(filename, 'rb') as file:
            return pickle.load(file)
    except FileNotFoundError:
        print(f"文件不存在: {filename}")
        return None
    except pickle.UnpicklingError as e:
        print(f"数据损坏: {e}")
        return None
    except Exception as e:
        print(f"读取错误: {e}")
        return None

# 安全保存数据
def safe_pickle_save(data, filename):
    """安全地保存数据"""
    try:
        with open(filename, 'wb') as file:
            pickle.dump(data, file)
        return True
    except Exception as e:
        print(f"保存失败: {e}")
        return False

# 使用安全函数
data = {"测试": "数据"}
if safe_pickle_save(data, "test.pkl"):
    loaded_data = safe_pickle_load("test.pkl")
    print(f"加载的数据: {loaded_data}")

import pickle

class ValidatedData:
    def __init__(self, data_type, max_size=1024*1024):  # 默认1MB限制
        self.data_type = data_type
        self.max_size = max_size
    
    def validate_and_save(self, data, filename):
        """验证并保存数据"""
        # 检查数据类型
        if not isinstance(data, self.data_type):
            raise ValueError(f"数据类型错误，期望 {self.data_type}")
        
        # 检查数据大小
        data_size = len(pickle.dumps(data))
        if data_size > self.max_size:
            raise ValueError(f"数据过大: {data_size} 字节 (限制: {self.max_size} 字节)")
        
        # 保存数据
        with open(filename, 'wb') as file:
            pickle.dump(data, file)
        
        print(f"数据验证通过并保存: {data_size} 字节")

# 使用验证器
validator = ValidatedData(dict, max_size=1024*500)  # 限制500KB

try:
    test_data = {"key": "value" * 1000}  # 创建较大数据
    validator.validate_and_save(test_data, "validated.pkl")
except ValueError as e:
    print(f"验证失败: {e}")

import pickle
import time

def benchmark_pickle():
    """测试pickle性能"""
    # 创建测试数据
    large_list = [str(i) * 100 for i in range(10000)]  # 10000个字符串
    
    # 测试序列化性能
    start_time = time.time()
    serialized_data = pickle.dumps(large_list, protocol=pickle.HIGHEST_PROTOCOL)
    serialize_time = time.time() - start_time
    
    # 测试反序列化性能
    start_time = time.time()
    deserialized_data = pickle.loads(serialized_data)
    deserialize_time = time.time() - start_time
    
    print(f"数据大小: {len(serialized_data)} 字节")
    print(f"序列化时间: {serialize_time:.3f} 秒")
    print(f"反序列化时间: {deserialize_time:.3f} 秒")
    print(f"数据验证: {len(deserialized_data) == len(large_list)}")

benchmark_pickle()