Скомпилируйте запрос из необработанной строки (без использования .text (...)), используя соединение Sqlalchemy и Postgres - PullRequest
Новая
       41

Скомпилируйте запрос из необработанной строки (без использования .text (...)), используя соединение Sqlalchemy и Postgres

0 голосов
/ 23 мая 2019

Я использую Sqlalchemy 1.3 для подключения к базе данных PostgreSQL 9.6 (через Psycopg).

У меня есть очень, очень необработанная строка Sql, отформатированная с использованием синтаксиса Psycopg2 , которую я не могу изменить из-за некоторых унаследованных проблем:

statement_str = SELECT * FROM users WHERE user_id=%(user_id)s

Обратите внимание на %(user_id)s

Я могу с радостью выполнить это, используя соединение sqlalchemy, просто выполнив: 

connection = sqlalch_engine.connect()
rows = conn.execute(statement_str, user_id=self.user_id)

И это прекрасно работает. Я получил мой пользователь, и все хорошо и хорошо.

Теперь для целей отладки я бы хотел получить фактический запрос с аргументом %(user_id)s, расширенным до фактического значения. Например: если user_id = "foo", тогда получите SELECT * FROM users WHERE user_id = 'foo'

Я видел множество примеров использования sqlalchemy.text(...) для создания оператора, а затем получения скомпилированной версии. У меня есть это благодаря другим ответам, таким как этот или этот смог выдать приличный str, когда у меня есть запрос SqlAlchemy.

Однако в данном конкретном случае, поскольку я использую более специфичный для курсора синтаксис %(user_id), я не могу этого сделать. Если я попытаюсь:

text(statement_str).bindparams(user_id="foo")

Я получаю: This text() construct doesn't define a bound parameter named 'user_id'

Так что я думаю, что то, что я ищу, будет примерно таким: 

conn.compile(statement_str, user_id=self.user_id)

Но я не смог этого получить.

Ответы [ 2 ]

1 голос
/ 23 мая 2019

Не уверен, что это то, что вы хотите, но здесь идет.

Предположим, что Statement_str на самом деле является строкой: 

import sqlalchemy as sa

statement_str = "SELECT * FROM users WHERE user_id=%(user_id)s"
params = {'user_id': 'foo'}

query_text = sa.text(statement_str % params)

# str(query_text) should print "select * from users where user_id=foo"
0 голосов
/ 23 мая 2019

Хорошо, я думаю Я понял.

Комбинация SqlAlchemy's raw_connection + Psycopg's mogrify кажется, что ответ. 

conn = sqlalch_engine.raw_connection()
try:
    cursor = conn.cursor()
    s_str = cursor.mogrify(statement_str, {'user_id': self.user_id})
    s_str = s_str.decode("utf-8")  # mogrify returns bytes

    # Some cleanup for niceness:
    s_str = s_str.replace('\n', ' ')
    s_str = re.sub(r'\s{2,}', ' ', s_str)
finally:
    conn.close()

Я надеюсь, что кто-то еще найдет это полезным

...