SQLAlchemy：如何在 Postgres 数据库中执行原始 INSERT sql 查询？

Question

我\xe2\x80\x99m 使用 Python 和干净的架构原则以及 TDD 构建一个应用程序。

\n

某些单元测试需要对内存数据库执行一些原始 SQL 查询。

\n

我正在尝试使用 pytest-postgres 从 sqlite 切换到 postgresql 内存数据。

\n

问题

\n

\n
• 当使用sqlite内存数据库时，我可以插入和选择数据。
\n
• 当使用Postgresql内存数据库时，我只能选择（原始插入失败）。
\n

\n

将工作插入 sqlite\xe2\x80\xa6

\n

    s_tb_name = "tb_customer"\n    ls_cols = ["first_name", "last_name", "email"]\n    ls_vals = [\'("John", "Doe", "john.doe@mail.net")\',\n               \'("Jane", "Doe", "jane.doe@mail.net")\',\n               \'("Eric", "Dal", "eric.d@home.com")\']\n    s_cols = \', \'.join(ls_cols)\n    s_vals = \', \'.join(ls_vals)\n    session.execute(f"INSERT INTO {s_tb_name} ({s_cols}) VALUES ({s_vals})")\n

\n

\xe2\x80\xa6但在 Postgres 中失败：

\n

E       sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedColumn) column "John" does not exist\nE       LINE 1: ..., email) VALUES (("John"....\n

\n

从这个 psycopg 文档页面，我了解到这是由于 pyscopg2 造成的。
\n它可以防止注入原始动态 SQL，看来我应该添加以下内容：

\n

tb_sql_id = sql.Identifier(s_tb_name)\ncols_sql_id = sql.SQL(\' ,\').join(map(sql.Identifier, ls_cols))\nvals_sql_id = sql.SQL(\' ,\').join(map(sql.Literal, ls_vals))\npsycopg2_query = sql.SQL(f"INSERT INTO {tb_sql_id} ({cols_sql_id}) VALUES ({vals_sql_id})")\n

\n

但从逻辑上讲，sqlalchemy 拒绝执行psycopg2_query：

\n

sqlalchemy.exc.ArgumentError: SQL expression object expected, got object of type <class \'psycopg2.sql.SQL\'> instead\n

\n

问题

\n

有没有办法使用 SQL Alchemy 在 Postgres 中执行原始动态插入查询？

\n

Answer 1

我不得不警告您有关 SQL 注入的问题，但由于这是为了您的测试，所以不必担心。

需要进行两处更改：

1. 需要将值ls_vals括在单引号中，而不是双引号中
2. 后面多余的括号VALUES需要去掉

    s_tb_name = "tb_customer"
    ls_cols = ["first_name", "last_name", "email"]
    ls_vals = ["('John', 'Doe', 'john.doe@mail.net')",
               "('Jane', 'Doe', 'jane.doe@mail.net')",
               "('Eric', 'Dal', 'eric.d@home.com')"]
    s_cols = ', '.join(ls_cols)
    s_vals = ', '.join(ls_vals)
    session.execute(f"INSERT INTO {s_tb_name} ({s_cols}) VALUES {s_vals}")

Answer 2

正如其他人指出的那样，在大多数情况下应该避免像这样注入 SQL。

\n

这里，SQL 是在单元测试本身中编写的。不存在外部输入泄露到SQL注入的情况，从而降低了安全风险。

\n

Mike Organek\xe2\x80\x99s 解决方案并不完全适合我，但它为我指明了正确的方向：我只需要从 ls_vals 中删除括号即可。

\n

    s_tb_name = "tb_customer"\n    ls_cols = ["first_name", "last_name", "email"]\n    ls_vals = ["'John', 'Doe', 'john.doe@mail.net'",\n               "'Jane', 'Doe', 'jane.doe@mail.net'",\n               "'Eric', 'Dal', 'eric.d@home.com'"]\n    s_cols = ', '.join(ls_cols)\n    s_vals = '(' + '), ('.join(ls_vals) + ')'\n    session.execute(f"INSERT INTO {s_tb_name} ({s_cols}) VALUES {s_vals}")\n

\n

这使得插入测试通过，无论是使用 sqlite 引擎还是 postgres 引擎。

\n