时间戳

首页 > 解决方案 > SQLAlchemy 数据库 URI 中的架构名称未反映到元数据

python - SQLAlchemy 数据库 URI 中的架构名称未反映到元数据

问题描述

我在config.py文件中构造一个 SQLAlchemy 数据库 URI，如下所示：

    db_user = getenv("MYSQL_USER")
    db_password = getenv("MYSQL_ROOT_PASSWORD")
    db_host = getenv("MYSQL_HOST")
    db_port = getenv("MYSQL_PORT")
    db_name = getenv("MYSQL_DATABASE")
    SQLALCHEMY_DATABASE_URI = f"mysql+pymysql://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}?charset=utf8"

然后我实例化一个元数据对象，backend.database.db只要我想通过元数据 API 与数据库交互，应用程序的其余部分就会导入：

engine = create_engine(Config.SQLALCHEMY_DATABASE_URI)
db_metadata = MetaData(bind=engine, reflect=True)

我的引擎和生成的元数据对象都对正确的 DB URI 进行编码，其中包括 DB 名称：

In [16]: db_metadata                                                                                                                                                                                                                    
Out[16]: MetaData(bind=Engine(mysql+pymysql://root:***@db:3306/main))

In [17]: engine                                                                                                                                                                                                                         
Out[17]: Engine(mysql+pymysql://root:***@db:3306/main)

然而，db_metadata.schema是None。Table这是一个问题，因为针对类对象的下游操作因错误而失败sqlalchemy.exc.InternalError: (pymysql.err.InternalError) (1046, 'No database selected')。

我可以通过传递给我的调用来解决这个问题：schema="main"MetaData

db_metadata = MetaData(bind=engine, reflect=True, schema="main")

这让我超越了数据库未找到错误，但现在我的问题是，现在当我从元数据数据中查找Table对象时，我必须在它们前面加上main，这必然会使代码混乱。

似乎是一个快速的变化，但我一直在阅读 SQLAlchemy 文档，但没有什么能引起我的注意。文档说：

如上所述，MetaData.schema 参数仅指将应用于传入 Table 对象的 Table.schema 参数的默认值。它不涉及如何在 MetaData 中对表进行编目，这与未定义此参数的 MetaData 集合保持一致。MetaData 中的表仍将根据其模式限定名称进行键控

所以我有两个问题：

  1. 为什么我对 MetaData 的调用没有自动获取main模式？
  2. 我是否必须使用schema.table语法引用索引表，或者是否有仅使用表名的最佳实践方法？

环境注意事项：

标签： pythonmysqlsqlalchemy

解决方案

我整理了这个。我写的所有内容都得到了妥善处理，问题是在我尝试使用元数据对象进行表更新的部分之前，我有一个函数，reset_db看起来像这样：

def reset_db():
    with db_session.connection() as conn:
        conn.execute("DROP DATABASE main;")
        conn.execute("CREATE DATABASE main;")
        db_session.commit()
    os.system("flask db upgrade")

我不只是清除旧数据，而是破坏数据库并切断它与 ORM 的链接。重构这个函数如下，现在它工作得很好：

def reset_db():
    db_metadata.drop_all()
    os.system("flask db upgrade")

诀窍是直接使用我认为是问题根源的 MetaData API。BaseClass由于我不会为每次测试运行删除和重新创建整个架构，因此对我的测试运行程序的此更新大大加快了我的测试实例化速度。

推荐阅读