python - SQLAlchemy : how can I execute a raw INSERT sql query in a Postgres database?

Question

我正在使用 Python 和干净的架构原则以及 TDD 构建一个应用程序。
一些单元测试需要对内存数据库执行一些原始 SQL 查询。
我正在尝试使用 pytest-postgres 从 sqlite 切换到 postgresql 内存数据。
问题

使用 sqlite inmemory 数据库时，我可以插入和选择数据。

使用 Postgresql 内存数据库时，我只能选择(原始插入失败)。

在 sqlite 中插入工作...

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

…但在 Postgres 中失败:

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

从这个 psycopg 文档 page ，我知道这是由于 pyscopg2。
它可以防止注入(inject)原始动态 SQL，看来我应该添加这个:

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

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

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

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

Answer 1

我不得不警告您有关 SQL 注入(inject)的信息，但因为这是针对您的测试而不必担心的。
需要进行两项更改:

ls_vals 中的值需要用单引号括起来，而不是双引号

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