Как выполнить сырой SQL в приложении SQLAlchemy-flask
Как вы выполняете сырой SQL в SQLAlchemy?
У меня есть веб-приложение Python, которое работает на колбе и взаимодействует с базой данных через SQLAlchemy.
Мне нужен способ запустить сырой SQL. Запрос включает в себя несколько объединений таблиц наряду с встроенными представлениями.
Я пробовал:
connection = db.session.connection()
connection.execute( <sql here> )
Но я продолжаю получать ошибки шлюза.
10 ответов
Ты пытался:
result = db.engine.execute("<sql here>")
или же:
from sqlalchemy import text
sql = text('select name from penguins')
result = db.engine.execute(sql)
names = []
for row in result:
names.append(row[0])
print names
Если вы хотите использовать сеанс (как подсказывает ваш вопрос), используйте его execute метод напрямую:
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})
Следующее может быть специфичным для моего драйвера базы данных (psycopg2); Я не уверен. Независимо от того, как я вытаскиваю свои ценности.
from collections import namedtuple
Record = namedtuple('Record', result.keys())
records = [Record(*r) for r in result.fetchall()]
for r in records:
print(r)
Ключевым моментом является fetchall() вызов. namedtuple Я считаю, что часть - это то, что облегчает мою жизнь, предоставляя доступ по имени.
Кроме того, это транзакция без ручного управления. Сказать make_session это функция, которая создает сессию:
>>> s1 = make_session()
>>> s1.execute('CREATE TABLE blah (id INTEGER)')
<sqlalchemy.engine.result.ResultProxy object at 0x02CD86F0>
>>> s1.commit()
>>>
>>> s1.execute('INSERT INTO blah VALUES (1)')
<sqlalchemy.engine.result.ResultProxy object at 0x02CD8870>
>>> s1.execute('SELECT * FROM blah').fetchall()
[(1,)]
>>>
>>> s2 = make_session()
>>> s2.execute('SELECT * FROM blah').fetchall()
[]
>>> s2.close()
>>>
>>> s1.commit()
>>>
>>> s2 = make_session()
>>> s2.execute('SELECT * FROM blah').fetchall()
[(1,)]
>>> s2.close()
>>> s1.close()
Вы можете получить результаты SELECT SQL-запросов, используя 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
Документы: Учебник по языку выражений 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')
Начиная с SQLAlchemy 1.4, выполнение без установления соединения или неявное выполнение устарело, т. Е.
db.engine.execute(...) # DEPRECATED
а также простые строки в качестве запросов.
Новый API требует явного подключения, например
from sqlalchemy import text
with db.engine.connect() as connection:
result = connection.execute(text("SELECT * FROM ..."))
for row in result:
# ...
Точно так же рекомендуется использовать существующий сеанс, если он доступен:
result = session.execute(sqlalchemy.text("SELECT * FROM ..."))
или используя параметры:
session.execute(sqlalchemy.text("SELECT * FROM a_table WHERE a_column = :val"),
{'val': 5})
Дополнительные сведения см. В документации « Выполнение без установления соединения, неявное выполнение ».
result = db.engine.execute(text("<sql here>"))
выполняет <sql here> но не совершает это, если вы не на autocommit Режим. Таким образом, вставки и обновления не отражаются в базе данных.
Чтобы зафиксировать после изменений, сделайте
result = db.engine.execute(text("<sql here>").execution_options(autocommit=True))
Это упрощенный ответ о том, как запустить 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))
Это использует текущее соединение с базой данных, которое имеет приложение.
Flask-SQLAlchemy v: 3.0.x / SQLAlchemy v: 1.4
users = db.session.execute(db.select(User).order_by(User.title.desc()).limit(150)).scalars()
Таким образом, в основном для последней стабильной версии flask -sqlalchemy, в частности , документация предлагает использоватьsession.execute()метод в сочетании сdb.select(Object).
Вы пытались использовать connection.execute(text( <sql here> ), <bind params here> ) и параметры привязки, как описано в документах? Это может помочь решить многие проблемы форматирования параметров и производительности. Может быть, ошибка шлюза - это тайм-аут? Параметры связывания обычно ускоряют выполнение сложных запросов.
Если вы хотите избежать кортежей, другой способ - вызвать first, one или all методы:
query = db.engine.execute("SELECT * FROM blogs "
"WHERE id = 1 ")
assert query.first().name == "Welcome to my blog"