CODEWITHSUNDEEP
pythonpostgresqlsqlalchemy
Lionel Hamayon
Lionel Hamayon

Reputation: 1330

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

I’m building an app using Python and the clean architecture principles, with TDD.

Some unit tests require executing some raw SQL queries against an in-memory database.

I am trying to switch from sqlite to postgresql inmemory data, using pytest-postgres.

Problem

Insert work in sqlite…

    s_tb_name = "tb_customer"
    ls_cols = ["first_name", "last_name", "email"]
    ls_vals = ['("John", "Doe", "[email protected]")',
               '("Jane", "Doe", "[email protected]")',
               '("Eric", "Dal", "[email protected]")']
    s_cols = ', '.join(ls_cols)
    s_vals = ', '.join(ls_vals)
    session.execute(f"INSERT INTO {s_tb_name} ({s_cols}) VALUES ({s_vals})")

…but fail in Postgres:

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

From this psycopg documentation page, I understand this is due to pyscopg2.
It prevents injecting raw dynamic SQL, and it seems I should add this :

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

but logically, sqlalchemy refuses to execute the psycopg2_query :

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

Question

Is there a way to execute raw dynamic insert queries in Postgres using SQL Alchemy?

Upvotes: 9

Views: 11915

Answers (2)

Lionel Hamayon
Lionel Hamayon

Reputation: 1330

As pointed by others, injecting SQL like this is to be avoided in most cases.

Here, the SQL is written in the unit test itself. There is no external input leaking to the SQL injection, which alleviates the security risk.

Mike Organek’s solution did not fully work for me, but it pointed me to the right direction : I just had to also remove the parens from ls_vals.

    s_tb_name = "tb_customer"
    ls_cols = ["first_name", "last_name", "email"]
    ls_vals = ["'John', 'Doe', '[email protected]'",
               "'Jane', 'Doe', '[email protected]'",
               "'Eric', 'Dal', '[email protected]'"]
    s_cols = ', '.join(ls_cols)
    s_vals = '(' + '), ('.join(ls_vals) + ')'
    session.execute(f"INSERT INTO {s_tb_name} ({s_cols}) VALUES {s_vals}")

This made the insert test pass, both when using the sqlite engine and the postgres engine.

Upvotes: 5

Mike Organek
Mike Organek

Reputation: 12484

I am compelled to warn you about SQL injection, but since this is for your tests that should not be a concern.

Two changes are needed:

  1. The values in ls_vals need to be enclosed in single, rather than double quotes
  2. The extra parens after VALUES need to be removed
    s_tb_name = "tb_customer"
    ls_cols = ["first_name", "last_name", "email"]
    ls_vals = ["('John', 'Doe', '[email protected]')",
               "('Jane', 'Doe', '[email protected]')",
               "('Eric', 'Dal', '[email protected]')"]
    s_cols = ', '.join(ls_cols)
    s_vals = ', '.join(ls_vals)
    session.execute(f"INSERT INTO {s_tb_name} ({s_cols}) VALUES {s_vals}")

Upvotes: 5

Related Questions