时间戳

首页 > 解决方案 > SQLAlchemy 'Connection' 对象没有属性 '_Inspector__engine'

python - SQLAlchemy 'Connection' 对象没有属性 '_Inspector__engine'

问题描述

我想使用 SQLAlchemy 读取未映射到对象的数据库（需要访问开发时未知的数据库）。其中一项功能是读取不同表的列名。因此我写了这个连接器：

我的连接器.py

import sqlalchemy as db

class MyConnector:
    __engine = None

    def __init__(self, db_connection):
        self.__engine = db.create_engine(db_connection)

    def __get_table(self, tbl_name):
        metadata = db.MetaData()
        table = db.Table(tbl_name, metadata)
        inspector = db.inspect(self.__engine)
        inspector.reflect_table(table, None)
        return table

    def get_table_columns(self, tbl_name):
        table = self.__get_table(tbl_name)
        return table.columns.keys()

通过以下 UnitTest 进行测试：

连接器测试.py

import unittest
from MyConnector import MyConnector

class MyTestCase(unittest.TestCase):
    db_string = "sqlite:///myTest.db"
    expected_columns_user = ["id", "f_name", "l_name", "note"]
    expected_columns_address = ["id", "street", "number", "postal_code", "additional_information", "fk_user"]

    def test_get_table_columns_user(self):
        connector = MyConnector(self.db_string)
        answer = connector.get_table_columns("user")
        self.assertListEqual(self.expected_columns_user, answer)

    def test_get_table_columns_address(self):
        connector = MyConnector(self.db_string)
        answer = connector.get_table_columns("address")
        self.assertListEqual(self.expected_columns_address, answer)


if __name__ == '__main__':
    unittest.main()

SQLite DB 只包含这两个空表：

CREATE TABLE user (
    id INTEGER NOT NULL DEFAULT AUTO_INCREMENT,
    f_name VARCHAR,
    l_name VARCHAR,
    note VARCHAR,
    PRIMARY KEY (
        id
    )
);

CREATE TABLE address (
    id INTEGER NOT NULL DEFAULT AUTO_INCREMENT,
    street VARCHAR NOT NULL,
    number INTEGER,
    postal_code INTEGER NOT NULL,
    additional_information VARCHAR,
    fk_user INTEGER NOT NULL,
    PRIMARY KEY (
        id
    ),
    FOREIGN KEY (
        fk_user
    )
    REFERENCES user(id) ON DELETE CASCADE
);

测试test_get_table_columns_user按预期工作。测试test_get_table_columns_address引发错误：

Error
Traceback (most recent call last):
  File "ConnectorTest.py", line 17, in test_get_table_columns_address
    answer = connector.get_table_columns("address")
  File "MyConnector.py", line 17, in get_table_columns
    table = self.__get_table(tbl_name)
  File "MyConnector.py", line 13, in __get_table
    inspector.reflect_table(table, None)
  File "venv\lib\site-packages\sqlalchemy\engine\reflection.py", line 802, in reflect_table
    reflection_options,
  File "venv\lib\site-packages\sqlalchemy\engine\reflection.py", line 988, in _reflect_fk
    **reflection_options
  File "<string>", line 2, in __new__
  File "venv\lib\site-packages\sqlalchemy\util\deprecations.py", line 298, in warned
    return fn(*args, **kwargs)
  File "venv\lib\site-packages\sqlalchemy\sql\schema.py", line 601, in __new__
    metadata._remove_table(name, schema)
  File "venv\lib\site-packages\sqlalchemy\util\langhelpers.py", line 72, in __exit__
    with_traceback=exc_tb,
  File "venv\lib\site-packages\sqlalchemy\util\compat.py", line 207, in raise_
    raise exception
  File "venv\lib\site-packages\sqlalchemy\sql\schema.py", line 596, in __new__
    table._init(name, metadata, *args, **kw)
  File "venv\lib\site-packages\sqlalchemy\sql\schema.py", line 676, in _init
    resolve_fks=resolve_fks,
  File "venv\lib\site-packages\sqlalchemy\sql\schema.py", line 705, in _autoload
    with insp._inspection_context() as conn_insp:
  File "AppData\Local\Programs\Python\Python37\lib\contextlib.py", line 112, in __enter__
    return next(self.gen)
  File "venv\lib\site-packages\sqlalchemy\engine\reflection.py", line 217, in _inspection_context
    sub_insp = self._construct(self.__class__._init_connection, conn)
  File "venv\lib\site-packages\sqlalchemy\engine\reflection.py", line 117, in _construct
    init(self, bind)
  File "venv\lib\site-packages\sqlalchemy\engine\reflection.py", line 135, in _init_connection
    self.engine = connection.__engine
AttributeError: 'Connection' object has no attribute '_Inspector__engine'

由于两个测试都运行完全相同的代码，除了表名的参数，我对结果很困惑。

使用谷歌我发现了这个 PR：https ://github.com/laughingman7743/PyAthena/pull/65/files 似乎与这个问题无关（引擎，不是连接）。

许多关于 Sessions 的帖子（如Python SQLAlchemy Query: AttributeError: 'Connection' object has no attribute 'contextual_connect'或https://github.com/sqlalchemy/sqlalchemy/issues/4046）。这无关紧要，因为在这种情况下使用 Inspector 似乎甚至不需要调用engine.connect().

为了排除挥之不去的锁或类似问题，我也将参数更改test_get_table_columns_address为"user"。这适用于断言，显然不匹配。所以在我看来，表（-name？）是个问题——这很奇怪。

对 SQLAlchemy 有更多经验和洞察力的人可以指出，我的代码的问题在哪里/是什么？提前致谢！

Python 3.7.4

SQLAlchemy 1.4.20

标签： pythonpython-3.xsqlitesqlalchemy

解决方案

太糟糕了，我不能给任何人一个更好的答案，我怀疑依赖关系被搞砸了......我想尝试不同版本的 SQLAlchemy 来编写具有上述行为的错误报告。因此，我通过以下命令更改了我的 venv 几次：

pip freeze > uninstall.txt
pip uninstall -r uninstall.txt -y
pip install -r requirements.txt

同时将 requirements.txt 更改为几个不同的 SQLAlchemy 版本。请注意，requirements.txt 始终只包含一个需求（不同版本的 SQLAlchemy）。

看到每个测试版本的行为都符合预期并在问题的单元测试中成功，我改回 1.4.20。现在即使使用 1.4.20 版，单元测试也会成功。

为了避免可能受到其他文件的影响，该项目始终只包含两个 py 文件和 db. 所以我认为我们可以排除项目本身的变化作为原因。正如@snakecharmerb 指出的那样，_我还尝试了“私人”机智。__测试通过双下划线和单下划线成功。所以我唯一的猜测是，首先是需求出了问题。可悲的是，我无法返回并检查第一个 uninstall.txt，因为它被覆盖了很多次。

推荐阅读