Выполнение необработанного SQL в Flask-SQLAlchemy

При работе с Flask-SQLAlchemy вы можете столкнуться со сценарием, в котором вам необходимо выполнить необработанный SQL запросы. Эти запросы могут включать сложные соединения таблиц или встроенные представления. Первоначальный подход, продемонстрированный в вопросе с использованием Connection.execute(), может вызвать ошибки шлюза. Чтобы решить эту проблему, давайте рассмотрим рекомендуемые подходы в SQLAlchemy версий 1.x и 2.0.

SQLAlchemy 2.0

Чтобы выполнить необработанный SQL в SQLAlchemy 2.0, вы можете использовать метод engine.connect():

with engine.connect() as connection:
    result = connection.execute(text('SELECT * FROM your_table'))
    # Perform operations on the result

Используя контекстный менеджер engine.connect(), вы гарантируете, что соединение будет установлено и автоматически закрыто при выходе из контекста.

SQLAlchemy 1. x

В SQLAlchemy 1.x вы можете выполнять необработанный SQL, используя метод db.engine.execute() в сочетании с функцией text():

from sqlalchemy import text

sql = text('select name from penguins')
result = db.engine.execute(sql)
names = [row[0] for row in result]
print(names)

Обратите внимание, что метод db.engine.execute() выполняет запросы «без установления соединения», то есть не устанавливает сеанс базы данных. Этот метод устарел в SQLAlchemy 2.0, и его следует избегать из соображений единообразия и лучших практик.