Wie führen Sie Raw SQL in SQLAlchemy aus?
Ich habe eine Python-Web-App, die auf einer Flasche läuft und über SQLAlchemy mit der Datenbank verbunden ist.
Ich brauche eine Möglichkeit, die rohe SQL auszuführen. Die Abfrage umfasst mehrere Tabellenverknüpfungen sowie Inline-Ansichten.
Ich habe es versucht:
connection = db.session.connection()
connection.execute( <sql here> )
Aber ich bekomme immer wieder Gateway-Fehler.
Hast du es versucht:
result = db.engine.execute("<sql here>")
oder:
from sqlalchemy import text
sql = text('select name from penguins')
result = db.engine.execute(sql)
names = [row[0] for row in result]
print names
Wenn Sie eine Sitzung verwenden möchten (wie in Ihrer Frage vorgeschlagen), verwenden Sie die execute
-Methode direkt:
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})
Das Folgende ist möglicherweise spezifisch für meinen Datenbanktreiber (psycopg2). Ich bin mir nicht sicher. Egal, ich ziehe meine Werte heraus.
from collections import namedtuple
Record = namedtuple('Record', result.keys())
records = [Record(*r) for r in result.fetchall()]
for r in records:
print(r)
Der Schlüsselpunkt ist der Aufruf fetchall()
. Der namedtuple
-Teil ist einfach etwas, das ich finde, das mein Leben einfacher macht, indem er einen namenbasierten Zugriff gibt.
Darüber hinaus ist dies transaktional ohne manuell zu verwalten. Angenommen, make_session
ist eine Funktion, die eine Sitzung erstellt:
>>> 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()
docs: Lernprogramm für SQL-Ausdruckssprache - Text verwenden
beispiel:
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')
Sie können die Ergebnisse von SELECT SQL-Abfragen mithilfe von from_statement()
und text()
wie hier hier abrufen. Sie müssen nicht mit Tupulen auf diese Weise umgehen. Als Beispiel für eine Klasse Benutzer mit dem Tabellennamen 'Benutzer' können Sie Folgendes versuchen:
from sqlalchemy.sql import text
.
.
.
user = session.query(User).from_statement(
text("SELECT * FROM users where name=:name")).\
params(name='ed').all()
return user
result = db.engine.execute(text("<sql here>"))
führt den <sql here>
aus, bindet ihn jedoch nur, wenn Sie sich im autocommit
-Modus befinden Inserts und Updates würden sich also nicht in der Datenbank widerspiegeln.
Um die Änderungen zu übernehmen, machen Sie
result = db.engine.execute(text("<sql here>").execution_options(autocommit=True))
Dies ist eine vereinfachte Antwort auf die Ausführung von SQL-Abfragen von Flask Shell
Ordnen Sie zuerst Ihr Modul zu (wenn sich Ihr Modul/Ihre Anwendung im Hauptordner manage.py befindet und Sie sich in einem UNIX-Betriebssystem befinden), führen Sie Folgendes aus:
export FLASK_APP=manage
Flask Shell ausführen
flask Shell
Importiere, was wir brauchen:
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy(app)
from sqlalchemy import text
Führen Sie Ihre Abfrage aus:
result = db.engine.execute(text("<sql here>").execution_options(autocommit=True))
Dies verwendet die aktuell Datenbankverbindung, die die Anwendung hat.
Haben Sie es mit connection.execute(text( <sql here> ), <bind params here> )
und Bindungsparametern wie in in den Dokumenten beschrieben versucht? Dadurch können viele Parameterformatierungs- und Leistungsprobleme gelöst werden. Vielleicht ist der Gateway-Fehler ein Timeout? Bindungsparameter führen dazu, dass komplexe Abfragen wesentlich schneller ausgeführt werden.