sqlalchemy.exc.NoSuchColumnError - как конвертировать рабочий SQL-запрос

Question

Я воссоздаю базу данных postgres, используя SQLAlchemy и pytest_pgsql (которые предоставляют фикстуру postgresql_db для установки / разрыва), которая выглядит хорошо созданной, однако я не могу вставить 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: "Не удалось найти столбец в строке для столбца" orgs_ocds_test.scheme ""

Как я могу заставить это работать без переписывания SQL, определенного в createRegistryDataSQLQuery? Есть ли лучший способ приблизиться к этому тесту?