在SQLAlchemy中操作多个Snowflake表时遇Schema错误

使用SQLAlchemy 1.4.47版本，尝试在SQLAlchemy中引用多个Snowflake数据库创建表对象以用于后续关联时出现错误。示例代码如下：

from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL
import sqlalchemy

connection_parameters = {
    "account": 'myaccount',
    "user": 'brian',
    "password": 'xyzzy',
    "role": "myrole",
    "warehouse": 'myware',
    "schema": 'qed',
    'database': 'DATABASE_1'
}

engine = create_engine(URL(**connection_parameters))

connection = engine.connect()
meta = sqlalchemy.MetaData(engine)
# This table, in DATABASE_1, works fine.
prim_tbl = sqlalchemy.Table('prim_tbl'.lower(), meta, schema='qed', autoload_with=engine)
connection.execute('USE DATABASE DATABASE_2;').fetchone()
sec_tbl = sqlalchemy.Table('sec_tbl'.lower(), meta, schema='deq', autoload_with=engine)

创建DATABASE_1中的prim_tbl表对象正常，但执行切换数据库语句后，创建sec_tbl时抛出错误："Schema 'DATABASE_1.DEQ' does not exist or not authorized"，请问如何让SQLAlchemy基于DATABASE_2创建sec_tbl？

SQLAlchemy的MetaData和Table对象会依赖初始连接时指定的数据库参数，直接执行USE DATABASE语句不会更新SQLAlchemy内部的数据库上下文，所以仍会默认使用DATABASE_1。可通过以下方式解决：

方法1：指定完整的数据库+Schema路径 #

定义sec_tbl时，直接将数据库名包含在schema参数中，格式为数据库名.模式名：

sec_tbl = sqlalchemy.Table('sec_tbl'.lower(), meta, schema='DATABASE_2.deq', autoload_with=engine)

无需切换数据库，直接明确指定表所在的数据库和模式，是最直接的方案。

方法2：创建新的连接/引擎 #

为第二个数据库单独创建引擎和连接，避免上下文冲突：

# 复制连接参数并修改数据库与schema
connection_params_db2 = connection_parameters.copy()
connection_params_db2['database'] = 'DATABASE_2'
connection_params_db2['schema'] = 'deq'

engine_db2 = create_engine(URL(**connection_params_db2))
connection_db2 = engine_db2.connect()
meta_db2 = sqlalchemy.MetaData(engine_db2)

sec_tbl = sqlalchemy.Table('sec_tbl'.lower(), meta_db2, schema='deq', autoload_with=engine_db2)

适合需要频繁操作多个数据库的场景，每个数据库使用独立的引擎和元数据对象，彻底避免上下文干扰。

方法3：更新连接内部的数据库上下文（不推荐） #

若坚持使用同一个连接，执行USE DATABASE后手动修改连接的内部属性（依赖Snowflake驱动实现）：

connection.execute('USE DATABASE DATABASE_2;').fetchone()
# 更新连接内部的数据库标识
connection.connection._database = 'DATABASE_2'
# 重新创建表对象
sec_tbl = sqlalchemy.Table('sec_tbl'.lower(), meta, schema='deq', autoload_with=engine)

注意：该方法依赖驱动内部细节，版本更新可能导致失效，优先选择前两种方法。

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