是否有任何适用于 sqlite（用于测试）和 mysql（用于生产）的 sqlalchemy 列表插值语法？

Question

这是我需要工作的最小代码示例（我们在 prod 中使用 mysql，我们喜欢原始 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)

注意：这个问题只发生在插入 list 而不仅仅是单个值时

我在 requirements.txt 文件中使用SQLAlchemy==1.3.11

现在我面临着在添加测试之前重写复杂代码（以不太理想的格式）或添加跳过/模拟我最想测试的代码部分的测试之间的不愉快选择。有什么更好的选择？

这个问题之前一定有人问过，但我一直没能找到。如果你知道它在哪里，请指点我:)

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]})