SQLAlchemy MetaData 对象的序列化：提升大型数据库应用性能

import pickle
import sqlalchemy as sa
from sqlalchemy import Table, Column, Integer

# 1. 创建一个内存SQLite引擎和连接
# 实际应用中，这里会是你的数据库连接字符串
engine = sa.create_engine('sqlite://')

# 2. 在数据库中创建一个示例表，用于反射
with engine.connect() as conn:
    conn.execute(sa.text("""
        CREATE TABLE my_table (
            id INTEGER PRIMARY KEY,
            name VARCHAR(50)
        )
    """))
    conn.execute(sa.text("""
        CREATE TABLE another_table (
            value TEXT
        )
    """))
    conn.commit() # 提交更改

# 3. 创建 MetaData 对象并进行反射
print("开始反射数据库元数据...")
metadata = sa.MetaData()
metadata.reflect(engine)
print("元数据反射完成。")

# 4. 序列化 MetaData 对象为字节流
print("序列化 MetaData 对象...")
pickled_metadata_bytes = pickle.dumps(metadata)
print(f"序列化后的字节流大小: {len(pickled_metadata_bytes)} 字节")

# 5. 反序列化字节流，恢复 MetaData 对象
print("反序列化 MetaData 对象...")
loaded_metadata = pickle.loads(pickled_metadata_bytes)
print("MetaData 对象反序列化完成。")

# 6. 验证反序列化后的 MetaData 对象
print("\n验证反序列化后的 MetaData 对象:")
print(f"原始 metadata.tables: {metadata.tables}")
print(f"加载的 loaded_metadata.tables: {loaded_metadata.tables}")

# 检查反射的表是否存在且结构正确
assert 'my_table' in loaded_metadata.tables
assert 'another_table' in loaded_metadata.tables

my_table_from_loaded = loaded_metadata.tables['my_table']
print(f"加载的 'my_table' 列: {[c.name for c in my_table_from_loaded.columns]}")
assert 'id' in [c.name for c in my_table_from_loaded.columns]
assert 'name' in [c.name for c in my_table_from_loaded.columns]

print("\nMetaData 对象已成功序列化、反序列化并验证。")

# 实际应用中，你可以将 pickled_metadata_bytes 写入文件或缓存
# with open('metadata.pkl', 'wb') as f:
#     f.write(pickled_metadata_bytes)
#
# # 之后从文件加载
# with open('metadata.pkl', 'rb') as f:
#     loaded_bytes = f.read()
#     reloaded_metadata = pickle.loads(loaded_bytes)
#     print(f"从文件加载的 reloaded_metadata.tables: {reloaded_metadata.tables}")

开始反射数据库元数据...
元数据反射完成。
序列化 MetaData 对象...
序列化后的字节流大小: XXXX 字节 (具体数值取决于表结构复杂性)
反序列化 MetaData 对象...
MetaData 对象反序列化完成。

验证反序列化后的 MetaData 对象:
原始 metadata.tables: FacadeDict({'my_table': Table('my_table', MetaData(), Column('id', INTEGER(), table=<my_table>, primary_key=True, nullable=False), Column('name', VARCHAR(length=50), table=<my_table>), schema=None), 'another_table': Table('another_table', MetaData(), Column('value', TEXT(), table=<another_table>), schema=None)})
加载的 loaded_metadata.tables: FacadeDict({'my_table': Table('my_table', MetaData(), Column('id', INTEGER(), table=<my_table>, primary_key=True, nullable=False), Column('name', VARCHAR(length=50), table=<my_table>), schema=None), 'another_table': Table('another_table', MetaData(), Column('value', TEXT(), table=<another_table>), schema=None)})
加载的 'my_table' 列: ['id', 'name']

MetaData 对象已成功序列化、反序列化并验证。