解决SQLAlchemy Automap无法映射现有MySQL表的常见问题

pip install sqlalchemy pymysql

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

# 数据库连接配置 (请根据你的实际情况修改)
# 建议使用环境变量或配置文件管理敏感信息
DB_USER = os.getenv('DB_USER', 'root')
DB_PASS = os.getenv('DB_PASS', 'your_password') # 替换为你的MySQL密码
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') # 确保这个数据库存在或将被创建

# 1. 构建数据库引擎
# 使用 echo=True 可以打印所有执行的SQL语句，有助于调试
try:
    engine = create_engine(
        f'mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_IP}:{DB_PORT}/{DB_NAME}',
        echo=True
    )
    print(f"尝试连接到数据库: mysql+pymysql://{DB_USER}:*****@{DB_IP}:{DB_PORT}/{DB_NAME}")
    # 尝试连接一次，验证连接字符串是否有效
    with engine.connect() as conn:
        conn.execute(text("SELECT 1"))
    print("数据库连接成功！")

except Exception as e:
    print(f"数据库连接失败，请检查连接字符串和凭据：{e}")
    exit()

# 2. 确保目标数据库存在并创建测试表
# 在实际生产环境中，你可能不会在这里创建表，而是确保现有表可见
with engine.connect() as conn, conn.begin():
    # 创建数据库（如果不存在）
    # 注意：创建数据库通常需要更高的权限，如果你的用户没有此权限，
    # 确保 DB_NAME 已经存在
    try:
        conn.execute(text(f"CREATE DATABASE IF NOT EXISTS {DB_NAME}"))
        conn.execute(text(f"USE {DB_NAME}")) # 确保在正确的数据库上下文中
    except Exception as e:
        print(f"无法创建或使用数据库 {DB_NAME}，请确保权限和名称正确：{e}")
        # 如果无法创建数据库，但希望连接到已存在的数据库，可以忽略此错误
        # 或者在 engine = create_engine(...) 中直接指定 DB_NAME

    # 创建一个测试表
    conn.execute(text('''
        CREATE TABLE IF NOT EXISTS tbl_Xyz (
            ID INT(11) NOT NULL AUTO_INCREMENT PRIMARY KEY,
            Name VARCHAR(45) NOT NULL
        )
    '''))
    print("测试表 'tbl_Xyz' 已创建或已存在。")

    # 验证表是否存在于当前连接的数据库中
    result = conn.execute(text("SHOW TABLES"))
    existing_tables = [row[0] for row in result]
    print(f"当前数据库中的表: {existing_tables}")
    if 'tbl_Xyz' not in existing_tables:
        print("警告: 'tbl_Xyz' 未在 'SHOW TABLES' 结果中找到，Automap可能无法发现它。")
    else:
        print("'tbl_Xyz' 已在数据库中被发现。")

# 3. 使用Automap进行反射
Base = automap_base()
Base.prepare(autoload_with=engine) # 使用 autoload_with 参数将 engine 传递给 prepare

print("\n--- Automap 反射结果 ---")

# 4. 验证 Automap 是否成功生成了类
if len(Base.classes.items()) > 0:
    print(f"Automap 成功映射了 {len(Base.classes.items())} 个表。")
    print("映射的类名:", [cls.__name__ for cls in Base.classes])
else:
    print("Automap 未能映射任何表，Base.classes 为空。")

# 进一步验证特定表的映射
if hasattr(Base.classes, 'tbl_Xyz'):
    print("\n成功映射 'tbl_Xyz' 表。")
    TblXyz = Base.classes.tbl_Xyz
    print(f"映射类名: {TblXyz.__name__}")
    print(f"对应的表名: {TblXyz.__table__.name}")
    print(f"ID 列是否存在: {TblXyz.__table__.c.ID is not None}")
    print(f"Name 列是否存在: {TblXyz.__table__.c.Name is not None}")

    # 示例: 使用映射类进行查询
    from sqlalchemy.orm import Session
    with Session(engine) as session:
        # 插入数据
        new_entry = TblXyz(Name="Test Name 1")
        session.add(new_entry)
        session.commit()
        print(f"插入了一条新的数据: ID={new_entry.ID}, Name={new_entry.Name}")

        # 查询数据
        first_entry = session.query(TblXyz).first()
        if first_entry:
            print(f"查询到的第一条数据: ID={first_entry.ID}, Name={first_entry.Name}")
        else:
            print("未查询到数据。")

else:
    print("未能映射 'tbl_Xyz' 表。")

print("\n--- 教程结束 ---")