如何在SQLAlchemy+MySQL中避免或释放元数据锁

使用SQLAlchemy 2.0操作MySQL数据库时，通过Metadata.create_all()创建表后直接调用drop_all()可以正常执行；但如果在创建表后、删除前用pd.read_sql_query执行查询操作，后续drop_all()会出现冻结，通过show processlist查看进程处于「Waiting for table metadata lock」状态。

示例代码 #

from sqlalchemy import Table, Column, Integer, String, MetaData, create_engine, inspect, select
import pandas as pd

engine = create_engine(con_str)

meta = MetaData(schema='test')
table1 = Table('table1',meta, Column('id1', Integer, primary_key = True), Column('col1', String(50)))
table2 = Table('table2',meta, Column('id2', Integer, primary_key = True), Column('col2', String(50)))

meta.create_all(bind=engine)
print(f"Tables after creation: {inspect(engine.connect()).get_table_names()}")

# pd.read_sql_query(sql=select(table1), con=engine.connect())

meta.drop_all(bind=engine)
print(f"Tables after drop_all: {inspect(engine.connect()).get_table_names()}")

取消注释pd.read_sql_query行后，查询能正常返回空DataFrame，但drop_all()会冻结。

原因分析 #

每次调用engine.connect()都会创建一个新的数据库连接，pd.read_sql_query使用该连接完成查询后，这个连接没有被主动关闭，导致它持有表的元数据锁。当后续drop_all()尝试获取元数据锁来删除表时，就会被阻塞，出现「Waiting for table metadata lock」状态。

解决方法 #

方法1：用上下文管理器自动管理连接 #

使用with语句包裹连接操作，上下文管理器会在代码块结束后自动关闭连接，释放锁：

with engine.connect() as conn:
    pd.read_sql_query(sql=select(table1), con=conn)

方法2：手动关闭连接 #

如果不使用上下文管理器，需要显式调用close()方法关闭连接：

conn = engine.connect()
pd.read_sql_query(sql=select(table1), con=conn)
conn.close()  # 显式关闭连接释放锁

方法3：直接传入engine作为连接参数（推荐） #

pd.read_sql_query支持直接接收SQLAlchemy Engine对象作为con参数，pandas会自动处理连接的创建和关闭，无需手动管理：

pd.read_sql_query(sql=select(table1), con=engine)

修改后的完整代码示例 #

from sqlalchemy import Table, Column, Integer, String, MetaData, create_engine, inspect, select
import pandas as pd

engine = create_engine(con_str)

meta = MetaData(schema='test')
table1 = Table('table1',meta, Column('id1', Integer, primary_key = True), Column('col1', String(50)))
table2 = Table('table2',meta, Column('id2', Integer, primary_key = True), Column('col2', String(50)))

meta.create_all(bind=engine)
print(f"Tables after creation: {inspect(engine.connect()).get_table_names()}")

# 使用推荐方法：直接传入engine
pd.read_sql_query(sql=select(table1), con=engine)

meta.drop_all(bind=engine)
print(f"Tables after drop_all: {inspect(engine.connect()).get_table_names()}")

内容的提问来源于stack exchange，提问作者langtang