SQLAlchemy Automap映射现有MySQL表：从调试到成功实践

import os
from sqlalchemy import create_engine, text
from sqlalchemy.ext.automap import automap_base

# 数据库连接配置 (请根据您的实际情况修改)
# 建议使用环境变量来管理敏感信息
DB_USER = os.getenv('DB_USER', 'root')
DB_PASS = os.getenv('DB_PASS', 'your_password') # 替换为您的数据库密码
DB_IP = os.getenv('DB_IP', '127.0.0.1')
DB_PORT = os.getenv('DB_PORT', '3306')
DB_NAME = os.getenv('DB_NAME', 'test_db') # 替换为您的数据库名

# 创建数据库引擎
# 设置 echo=True 可打印SQL语句，对于调试连接和反射过程非常有帮助
engine = create_engine(
    f'mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_IP}:{DB_PORT}/{DB_NAME}',
    echo=True
)

# --- 1. 验证数据库连接及表是否存在 (可选但强烈推荐) ---
# 此步骤旨在确认数据库连接是有效的，并且目标表是可见的。
# 如果您的表已存在，可以跳过创建表的SQL，但保留SHOW TABLES以验证。
try:
    with engine.connect() as conn, conn.begin():
        # 如果需要，创建一张测试表。如果表已存在，此语句不会报错。
        conn.execute(text('''
            CREATE TABLE IF NOT EXISTS tbl_Xyz (
                ID INT(11) NOT NULL AUTO_INCREMENT PRIMARY KEY,
                Name VARCHAR(45) NOT NULL
            )
        '''))
        # 验证表是否已创建或存在于当前数据库中
        existing_tables = [row[0] for row in conn.execute(text('SHOW TABLES'))]
        print(f"\n--- 数据库 '{DB_NAME}' 中存在的表: {existing_tables} ---")
        if 'tbl_Xyz' not in existing_tables:
            print("警告: 'tbl_Xyz' 表未在数据库中找到，请检查创建语句或数据库名。")

except Exception as e:
    print(f"\n错误: 数据库连接或表操作失败: {e}")
    print("请检查数据库连接字符串、凭据以及数据库是否已启动。")
    exit() # 如果连接失败，后续Automap操作也无法进行

# --- 2. 初始化 Automap Base ---
Base = automap_base()

# --- 3. 准备（反射）数据库表结构 ---
# autoload_with 参数是关键，它指示 Automap 使用哪个引擎来加载元数据
print("\n--- 正在执行 Base.prepare() 进行表反射 ---")
Base.prepare(autoload_with=engine)
print("--- 表反射完成 ---")

# --- 4. 验证 Automap 是否成功映射 ---
print('\n--- Automap 映射结果验证 ---')
print(f"Base.classes 中的项目数量: {len(Base.classes.items())}")
print(f"Base.metadata.tables 中的表数量: {len(Base.metadata.tables.keys())}")

# 检查特定表是否被映射
if 'tbl_Xyz' in Base.classes:
    TblXyz = Base.classes.tbl_Xyz
    print(f"成功映射表: {TblXyz.__tablename__}")
    print(f"表 '{TblXyz.__tablename__}' 的列: {[c.name for c in TblXyz.__table__.columns]}")

    # 进一步断言以确保映射正确
    assert len(Base.classes.items()) > 0, "Base.classes 为空，映射失败。"
    assert TblXyz is not None, "tbl_Xyz 类未被映射。"
    assert TblXyz.__table__ is not None, "tbl_Xyz 类的 __table__ 属性为空。"
    assert TblXyz.__table__.c.ID is not None, "tbl_Xyz 类的 ID 列未被映射。"
    print("Automap 映射成功并通过断言验证。")
else:
    print("错误: 'tbl_Xyz' 表未被 Automap 成功映射。请检查上述步骤和数据库配置。")

# 至此，您可以通过 TblXyz 类像使用常规ORM模型一样操作 tbl_Xyz 表了
# 例如，创建一个Session并进行查询：
# from sqlalchemy.orm import Session
# with Session(engine) as session:
#     all_xyz_records = session.query(TblXyz).all()
#     print(f"查询到的 tbl_Xyz 记录数量: {len(all_xyz_records)}")