sqlalchemy.exc.NoSuchColumnError如何转换工作SQL查询

Question

我正在使用SQLAlchemy和pytest\u pgsql（它为setup/teardown提供了一个postgresql\u db fixture）重新创建一个postgres数据库，它看起来创建得很好，但是我不能插入一个SQL查询（在实时数据库上工作）

调用all（）工作并将表orgs_ocds_test中的数据打印到控制台：

result = postgresql_db.session.query(orgs_ocds_test).all()
for row in result:
    print(row)

.('GB-COH', 'IP22353R', '', 'WEST COUNTRY HOUSING ASSOCIATION (AT) LIMITED', 'orgs_data_cache', datetime.datetime(2018, 9, 2, 6, 18, 21, 975143, tzinfo=psycopg2.tz.FixedOffsetTimezone(offset=0, name=None)), 1)

这也适用于：

result = postgresql_db.session.query(orgs_ocds_test)
      .from_statement(text("SELECT id, scheme, uri, legalname, source,
               created_at, orgs_ocds_scheme_id_idx from orgs_ocds_test"))

完整代码：

conftest.py

 @pytest.fixture()
    def test_create_reg_table(postgresql_db):

        class orgs_ocds_test(declarative_base()):
            __tablename__ = 'orgs_ocds_test'

            scheme = Column(String)
            id = Column(String)
            uri = Column(String)
            legalname = Column(String)
            source = Column(String)
            created_at = Column(DateTime(timezone=True), server_default=func.now())
            orgs_ocds_scheme_id_idx = Column(Integer, primary_key=True)


        postgresql_db.create_table(orgs_ocds_test)

        postgresql_db.load_csv('test_data/orgs_ocds_test_data.csv', orgs_ocds_test)

        orgs_ocds_test = postgresql_db.get_table('orgs_ocds_test')

        return postgresql_db, orgs_ocds_test

test_db_calls.py

 def test_fetchregData(settings, test_create_reg_table):

        # Pull through postgresql_db object and table from fixture        
        postgresql_db, orgs_ocds_test = test_create_reg_table

        # Overwrite production table with fixture/test table
        settings.reg_data_source = orgs_ocds_test

        query = db_calls.FetchData.createRegistryDataSQLQuery(settings)
        result = postgresql_db.session.query(orgs_ocds_test).from_statement(text(query))       
        for row in result:
            print(row)

我尝试运行的实际查询来自：

  def createRegistryDataSQLQuery(self):
        query = \
            """
            SELECT
           legalname as reg_name,
           id as reg_id,
           '' as reg_address,
           scheme as reg_scheme,
           source as reg_source,
           created_at as reg_created_at
            from {}

            """.format(self.reg_data_source)
        return query

所以当我试着运行这个测试时，我得到了一个错误：

sqlalchemy.exc.NoSuchColumnError: "Could not locate column in row for column 'orgs_ocds_test.scheme'"

如果不重新编写在createRegistryDataSQLQuery中定义的SQL，我怎么能让它工作呢？有没有更好的方法来进行这个测试