So führen Sie Raw SQL in der Flask-SQLAlchemy-App aus

Wie führt man Raw SQL in SQLAlchemy aus?

Ich habe eine Python-Web-App, die auf einer Flasche ausgeführt wird und über SQLAlchemy mit der Datenbank verbunden ist.

Ich brauche eine Möglichkeit, um das rohe SQL auszuführen. Die Abfrage umfasst mehrere Tabellenverknüpfungen zusammen mit 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

SQL Alchemy-Sitzungsobjekte haben ihre eigene executeMethode:

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

Alle Ihre Anwendungsabfragen sollten ein Sitzungsobjekt durchlaufen, unabhängig davon, ob es sich um unformatiertes SQL handelt oder nicht. Dadurch wird sichergestellt, dass die Abfragen ordnungsgemäß von einer Transaktion verwaltet werden , sodass mehrere Abfragen in derselben Anforderung als eine Einheit festgeschrieben oder zurückgesetzt werden können. Wenn Sie die Transaktion über die Engine oder die Verbindung verlassen , besteht ein viel höheres Risiko für subtile, möglicherweise schwer zu erkennende Fehler, die zu beschädigten Daten führen können. Jede Anforderung sollte nur einer Transaktion zugeordnet sein. Durch die Verwendung db.sessionwird sichergestellt, dass dies für Ihre Anwendung der Fall ist.

Beachten Sie auch, dass dies executefür parametrisierte Abfragen ausgelegt ist . Verwenden Sie Parameter wie :valim Beispiel für alle Eingaben in die Abfrage, um sich vor SQL-Injection-Angriffen zu schützen. Sie können den Wert für diese Parameter angeben, indem Sie a dictals zweites Argument übergeben, wobei jeder Schlüssel der Name des Parameters ist, wie er in der Abfrage angezeigt wird. Die genaue Syntax des Parameters selbst kann je nach Datenbank unterschiedlich sein, aber alle wichtigen relationalen Datenbanken unterstützen sie in irgendeiner Form.

Angenommen, es handelt sich um eine SELECTAbfrage, wird eine iterierbare Anzahl von RowProxyObjekten zurückgegeben.

Sie können mit verschiedenen Techniken auf einzelne Spalten zugreifen:

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

Persönlich ziehe ich es vor, die Ergebnisse in namedtuples umzuwandeln :

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)

Wenn Sie die Flask-SQLAlchemy-Erweiterung nicht verwenden, können Sie dennoch problemlos eine Sitzung verwenden:

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})

docs: SQL Expression Language Tutorial - Verwenden von Text

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, 'Tim', 'Gurra', 'Staff', '991-509-9284')
(1, 'Jim', 'Carey', 'Staff', '832-252-1910')
(2, 'Lee', 'Asher', 'Staff', '897-747-1564')
(3, 'Ben', 'Hayes', 'Staff', '584-255-2631')

Sie können die Ergebnisse von SELECT SQL-Abfragen mit from_statement()und text()wie hier gezeigt abrufen . Auf diese Weise müssen Sie nicht mit Tupeln umgehen. Als Beispiel für eine Klasse Usermit dem Tabellennamen userskö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 das aus <sql here>, schreibt es jedoch nur fest, wenn Sie sich im autocommitModus befinden. Einfügungen und Aktualisierungen werden also nicht in der Datenbank angezeigt.

Um sich nach den Änderungen zu verpflichten, tun Sie dies

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

Dies ist eine vereinfachte Antwort zum Ausführen einer SQL-Abfrage über Flask Shell

Ordnen Sie zunächst Ihr Modul zu (wenn sich Ihr Modul / Ihre App im Hauptordner "manage.py" befindet und Sie sich in einem UNIX-Betriebssystem befinden), führen Sie Folgendes aus:

export FLASK_APP=manage

Führen Sie die Kolbenschale aus

flask shell

Importieren Sie, 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 aktuelle Datenbankverbindung, die die Anwendung hat.

Haben Sie versucht, connection.execute(text( <sql here> ), <bind params here> )Parameter wie in den Dokumenten beschrieben zu verwenden und zu binden ? Dies kann helfen, viele Parameterformatierungs- und Leistungsprobleme zu lösen. Vielleicht ist der Gateway-Fehler eine Zeitüberschreitung? Bindungsparameter führen dazu, dass komplexe Abfragen wesentlich schneller ausgeführt werden.

Wenn Sie Tupeln vermeiden wollen, ist eine andere Art und Weise durch den Aufruf der first, oneoder allMethoden:

query = db.engine.execute("SELECT * FROM blogs "
                           "WHERE id = 1 ")

assert query.first().name == "Welcome to my blog"