Как выполнить сырой SQL в приложении Flask-SQLAlchemy

Ответ 1

Ты пытался:

result = db.engine.execute("<sql here>")

или же:

from sqlalchemy import text

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

Ответ 2

Объекты сеанса SQL Alchemy имеют свой собственный метод execute:

result = db.session.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})

Все запросы вашего приложения должны проходить через объект сеанса, независимо от того, являются ли они необработанным SQL или нет. Это гарантирует, что запросы правильно управляются транзакцией, что позволяет фиксировать или откатывать несколько запросов в одном запросе как один блок. Выход за пределы транзакции с использованием движка или соединения подвергает вас гораздо большему риску скрытых ошибок, возможно, трудно обнаружить ошибки, которые могут привести к повреждению данных. Каждый запрос должен быть связан только с одной транзакцией, и использование db.session гарантирует, что это применимо к вашему приложению.

Также обратите внимание, что execute предназначен для параметризованных запросов. Используйте параметры, такие как :val в примере, для любых входных данных запроса, чтобы защитить себя от атак SQL-инъекций. Вы можете предоставить значение для этих параметров, передав в качестве второго аргумента dict, где каждый ключ - это имя параметра, которое отображается в запросе. Точный синтаксис самого параметра может отличаться в зависимости от вашей базы данных, но все основные реляционные базы данных поддерживают их в той или иной форме.

Предполагая, что это запрос SELECT, он вернет итерируемый из RowProxy объектов.

Вы можете получить доступ к отдельным столбцам различными способами:

for r in result:
    print(r[0]) # Access by positional index
    print(r['my_column']) # Access by column name as a string
    r_dict = dict(r.items()) # convert to dict keyed by column names

Лично я предпочитаю конвертировать результаты в namedtuple s:

from collections import namedtuple

Record = namedtuple('Record', result.keys())
records = [Record(*r) for r in result.fetchall()]
for r in records:
    print(r.my_column)
    print(r)

Если вы не используете расширение Flask-SQLAlchemy, вы все равно можете легко использовать сеанс:

import sqlalchemy
from sqlalchemy.orm import sessionmaker, scoped_session

engine = sqlalchemy.create_engine('my connection string')
Session = scoped_session(sessionmaker(bind=engine))

s = Session()
result = s.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})

Ответ 3

документы: Учебник по языку выражений SQL - Использование текста

пример:

from sqlalchemy.sql import text

connection = engine.connect()

# recommended
cmd = 'select * from Employees where EmployeeGroup == :group'
employeeGroup = 'Staff'
employees = connection.execute(text(cmd), group = employeeGroup)

# or - wee more difficult to interpret the command
employeeGroup = 'Staff'
employees = connection.execute(
                  text('select * from Employees where EmployeeGroup == :group'), 
                  group = employeeGroup)

# or - notice the requirement to quote "Staff"
employees = connection.execute(
                  text('select * from Employees where EmployeeGroup == "Staff"'))


for employee in employees: logger.debug(employee)
# output
(0, u'Tim', u'Gurra', u'Staff', u'991-509-9284')
(1, u'Jim', u'Carey', u'Staff', u'832-252-1910')
(2, u'Lee', u'Asher', u'Staff', u'897-747-1564')
(3, u'Ben', u'Hayes', u'Staff', u'584-255-2631')

Ответ 4

Вы можете получить результаты SQL-запросов SELECT, используя from_statement() и text(), как показано здесь. Вы не должны иметь дело с кортежами таким образом. В качестве примера для класса User с именем таблицы users вы можете попробовать

from sqlalchemy.sql import text
.
.
.
user = session.query(User).from_statement(
    text("SELECT * FROM users where name=:name")).\
    params(name='ed').all()

return user

Ответ 5

result = db.engine.execute(text("<sql here>"))

выполняет <sql here>, но не фиксирует его, если вы не находитесь в режиме autocommit. Таким образом, вставки и обновления не отражаются в базе данных.

Чтобы зафиксировать после изменений, сделайте

result = db.engine.execute(text("<sql here>").execution_options(autocommit=True))

Ответ 6

Это упрощенный ответ о том, как запустить SQL-запрос из Flask Shell

Во-первых, сопоставьте ваш модуль (если ваш модуль/приложение - manage.py в основной папке, а вы в операционной системе UNIX), запустите:

export FLASK_APP=manage

Запустить колбу

flask shell

Импортируйте то, что нам нужно:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy(app)
from sqlalchemy import text

Запустите ваш запрос:

result = db.engine.execute(text("<sql here>").execution_options(autocommit=True))

Это использует текущее соединение с базой данных, которое имеет приложение.

Ответ 7

Вы пытались использовать connection.execute(text( <sql here> ), <bind params here> ) и связывать параметры, как описано в документации? Это может помочь решить многие проблемы форматирования параметров и производительности. Может быть, ошибка шлюза - это тайм-аут? Параметры связывания обычно ускоряют выполнение сложных запросов.