Есть ли какой-либо синтаксис интерполяции списка sqlalchemy, который работает как в sqlite (для теста), так и в mysql (для продукта)?

Question

Вот самый крошечный пример кода, который мне нужен для работы (мы используем mysql в prod, нам нравятся необработанные sql запросы для сложных запросов, потому что аналитики данных могут их читать / изменять / проверять, мы используем sqlite для модульное тестирование)

    from typing import List, Any

    def test_sqlite_params_demo(self, session):
      session.add(build_foo())
      session.commit()
      query = "select * from foo where id not in :id_list"
      result = session.execute(query, {"id_list": [9, 99, 999]})
      result_dict: List[Any] = [dict(r) for r in result]
      assert len(result_dict) is 1

Вот ошибка, вызванная вышеуказанным:

        def do_execute(self, cursor, statement, parameters, context=None):
>       cursor.execute(statement, parameters)
E       sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) near "?": syntax error
E       [SQL: select * from patient where id not in ?]
E       [parameters: ([9, 99, 999],)]
E       (Background on this error at: http://sqlalche.me/e/e3q8)

Если я использую кортеж вместо result = session.execute(query, {"id_list": (9, 99, 999)}), то ошибка будет выглядеть следующим образом:

>       cursor.execute(statement, parameters)
E       sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) near "?": syntax error
E       [SQL: select * from foo where id not in ?]
E       [parameters: ((9, 99, 999),)]
E       (Background on this error at: http://sqlalche.me/e/e3q8)

Примечание: эта проблема возникает только при интерполяции списка , а не только одного значения

Я использую SQLAlchemy==1.3.11 в моем файле require.txt

Прямо сейчас Я столкнулся с неприятным выбором между переписыванием сложного кода (в менее желательный формат) перед добавлением теста или добавлением теста, который пропускает / высмеивает ту часть кода, которую я больше всего хочу протестировать. Какой вариант лучше?

Этот вопрос уже задавался раньше, но я не смог его найти. Пожалуйста, укажите мне, если вы знаете, где это:)

Answer 1

Я создал ответ, который работает для меня

def is_test():
  """
  Tests which want to run code that executes raw sql can override this to be True
  Example:
  @patch('my_app.helpers.raw_sql_runner.is_test', side_effect=lambda: True)
  def test_run_raw_sql_string(self, is_test_mock, session):
  """
  return False

def run_raw_sql(session, sql, params):
  if is_test():
    for key, val in params.items():
        if isinstance(val, (list, tuple)):
            interpolated_sql = sql.replace(f':{key}', f"({','.join(map(str, val))})")
        elif isinstance(val, str):
            interpolated_sql = sql.replace(f':{key}', f"'{str(val)}'")
        else:
            interpolated_sql = sql.replace(f':{key}', str(val))
    return session.execute(interpolated_sql, params)
  else:
    return session.execute(sql, params)

Мне также нужно было создать некоторые методы, которые mysql имеет, чего нет у sqlite.

# conftest.py
def fake_sqlite_concat(a, b):
  return f'{a}{b}'

def create_sqlite_functions(con):
  con.create_function("concat", 2, fake_sqlite_concat)
  # more functions go here as needed

con = engine.connect().connection
create_sqlite_functions(con)

Answer 2

Если вы приведете свой запрос к объекту sqlalchemy.text, это должно сработать.

from sqlalchemy import text
result = session.execute(text(query), {"id_list": [9, 99, 999]})