Блог

Я создаю приложение, используя Python и принципы чистой архитектуры, с TDD.

Некоторые модульные тесты требуют выполнения некоторых необработанных SQL-запросов к базе данных в памяти.

Я пытаюсь переключиться с sqlite на postgresql inmemory data, используя pytest-postgres.

Проблема

• При использовании базы данных sqlite inmemory я могу как вставлять, так и выбирать данные.
• При использовании базы данных Postgresql inmemory я могу только ВЫБРАТЬ (ошибка необработанной ВСТАВКИ).

Вставка работы в 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 я понимаю, что это связано с pyscopg2.
Это предотвращает внедрение необработанного динамического 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
  

Вопрос

Есть ли способ выполнить необработанные динамические запросы insert в Postgres с использованием SQL Alchemy?

Я вынужден предупредить вас о внедрении 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}")
  

Как указывали другие, в большинстве случаев следует избегать внедрения SQL подобным образом.

Здесь SQL записан в самом модульном тестировании. При внедрении SQL не происходит утечки внешних входных данных, что снижает риск для безопасности.

Решение Майка Органека не полностью сработало для меня, но оно указало мне правильное направление: мне просто нужно было также удалить скобки из ls_vals.

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

Это позволило выполнить тест insert как при использовании движка sqlite, так и при использовании движка postgres.