SQLAlchemy Automap 映射现有MySQL表失败的排查与实践指南

import os
from sqlalchemy.ext.automap import automap_base
from sqlalchemy import create_engine, text, MetaData, Column, Integer, String, PrimaryKeyConstraint
from sqlalchemy.orm import Session

# --- 1. 数据库连接参数配置 ---
# 建议使用环境变量管理敏感信息
DB_USER = os.getenv("DB_USER", "root")
DB_PASS = os.getenv("DB_PASS", "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") # 确保这是包含目标表的数据库名

# --- 2. 配置数据库引擎，开启echo=True方便调试 ---
# 使用pymysql作为MySQL驱动，并开启echo以打印所有SQL语句
print("--- 正在配置数据库引擎 ---")
engine = create_engine(
    f'mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_IP}:{DB_PORT}/{DB_NAME}',
    echo=True,  # 开启日志，观察SQLAlchemy的内部操作
    pool_recycle=3600 # 可选：设置连接池回收时间，防止MySQL wait_timeout导致连接失效
)

# --- 3. 验证数据库连接和现有表 ---
print("\n--- 验证数据库连接和现有表是否存在 ---")
try:
    with engine.connect() as conn:
        # 尝试创建一个示例表（如果不存在），模拟一个已存在的表
        # 这里使用用户原始的表结构，不带AUTO_INCREMENT，Automap同样可以处理
        conn.execute(text('''
            CREATE TABLE IF NOT EXISTS tbl_Xyz (
                ID INT(11) NOT NULL,
                Name VARCHAR(45) NOT NULL,
                PRIMARY KEY (ID)
            )
        '''))
        conn.commit() # 提交创建表的事务

        # 查询当前数据库中的所有表，确认目标表是否存在
        result = conn.execute(text("SHOW TABLES"))
        existing_tables = [row[0] for row in result]
        print(f"当前数据库 '{DB_NAME}' 中检测到的表: {existing_tables}")

        if 'tbl_Xyz' in existing_tables:
            print("成功找到 'tbl_Xyz' 表，可以进行Automap映射。")
        else:
            print(f"警告: 'tbl_Xyz' 表未在数据库 '{DB_NAME}' 中找到。Automap可能无法映射。")

except Exception as e:
    print(f"错误: 数据库连接或表操作失败: {e}")
    print("请检查您的数据库连接字符串、凭据、网络连接以及数据库是否可访问。")
    exit() # 如果连接失败，后续Automap操作无意义

# --- 4. 使用Automap进行反射 ---
print("\n--- 尝试使用Automap反射数据库元数据 ---")
Base = automap_base()
try:
    # 关键步骤：使用autoload_with参数，将引擎直接传递给prepare方法
    # 这会指示Automap使用该引擎连接数据库并加载元数据
    Base.prepare(autoload_with=engine)
    print("Automap prepare 完成，元数据已加载。")
except Exception as e:
    print(f"错误: Automap prepare 失败: {e}")
    print("请检查数据库元数据是否完整（如主键是否存在，表结构是否规范）。")
    exit()

# --- 5. 验证映射结果 ---
print("\n--- 验证Automap映射结果 ---")
if len(Base.classes.items()) > 0:
    print(f"成功映射到 {len(Base.classes.items())} 个类。")
    print("映射的类名列表:", [name for name, _ in Base.classes.items()])
else:
    print("警告: Automap 未能映射到任何类。请再次检查数据库连接和表结构。")

# 尝试访问映射后的类和其属性
try:
    # Automap 会将表名转换为类名，通常是去除前缀并首字母大写，或直接使用表名
    # 这里我们假设表名为 'tbl_Xyz'，映射后的类名为 'tbl_Xyz'
    TblXyz = Base.classes.tbl_Xyz
    print(f"\n成功访问映射后的类: {TblXyz}")
    print(f"类 '{TblXyz.__name__}' 对应的数据库表名: {TblXyz.__table__.name}")
    print(f"类 '{TblXyz.__name__}' 的列: {[c.name for c in TblXyz.__table__.columns]}")

    # 示例：使用Session进行数据操作 (可选)
    # with Session(engine) as session:
    #     # 插入一些示例数据 (如果表为空)
    #     if session.query(TblXyz).count() == 0:
    #         session.add(TblXyz(ID=1, Name="Alice"))
    #         session.add(TblXyz(ID=2, Name="Bob"))
    #         session.commit()
    #         print("插入示例数据。")
    #
    #     # 查询所有数据
    #     print("\n查询 'tbl_Xyz' 表中的数据:")
    #     for user in session.query(TblXyz).all():
    #         print(f"  ID={user.ID}, Name={user.Name}")

except AttributeError:
    print("错误: 'tbl_Xyz' 类未被成功映射到 Base.classes 中。")
except Exception as e:
    print(f"访问映射类或进行数据操作时发生错误: {e}")