Cómo ejecutar SQL sin procesar en la aplicación Flask-SQLAlchemy

¿Cómo se ejecuta SQL sin procesar en SQLAlchemy?

Tengo una aplicación web de Python que se ejecuta en un matraz e interfaces con la base de datos a través de SQLAlchemy.

Necesito una forma de ejecutar el SQL sin formato. La consulta involucra múltiples combinaciones de tablas junto con vistas en línea.

He intentado:

connection = db.session.connection()
connection.execute( <sql here> )

Pero sigo recibiendo errores de puerta de enlace.

Has probado:

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

o:

from sqlalchemy import text

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

Los objetos de sesión de SQL Alchemy tienen su propio executemétodo:

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

Todas las consultas de su aplicación deben pasar por un objeto de sesión, ya sean SQL sin formato o no. Esto asegura que las consultas sean gestionadas adecuadamente por una transacción , lo que permite que múltiples consultas en la misma solicitud se confirmen o reviertan como una sola unidad. Salir de la transacción utilizando el motor o la conexión lo pone en un riesgo mucho mayor de errores sutiles, posiblemente difíciles de detectar, que pueden dejarlo con datos corruptos. Cada solicitud debe asociarse con una sola transacción, y el uso db.sessiongarantizará que este sea el caso de su aplicación.

También tenga en cuenta que executeestá diseñado para consultas parametrizadas . Use parámetros, como :valen el ejemplo, para cualquier entrada a la consulta para protegerse de los ataques de inyección SQL. Puede proporcionar el valor de estos parámetros pasando un dictsegundo argumento, donde cada clave es el nombre del parámetro tal como aparece en la consulta. La sintaxis exacta del parámetro en sí puede ser diferente según su base de datos, pero todas las principales bases de datos relacionales los admiten de alguna forma.

Asumiendo que es una SELECTconsulta, esto devolverá un iterable de RowProxyobjetos.

Puede acceder a columnas individuales con una variedad de técnicas:

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

Personalmente, prefiero convertir los resultados en namedtuples:

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)

Si no está usando la extensión Flask-SQLAlchemy, aún puede usar fácilmente una sesión:

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: Tutorial del lenguaje de expresiones SQL - Usar texto

ejemplo:

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

Puede obtener los resultados de las consultas SELECT SQL utilizando from_statement()y text()como se muestra aquí . No tienes que lidiar con las tuplas de esta manera. Como ejemplo para una clase que Usertiene el nombre de la tabla usersque puede probar,

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

ejecuta el <sql here>pero no lo confirma a menos que esté en autocommitmodo. Por lo tanto, las inserciones y actualizaciones no se reflejarían en la base de datos.

Para comprometerse después de los cambios, haga

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

Esta es una respuesta simplificada sobre cómo ejecutar una consulta SQL desde Flask Shell

Primero, asigne su módulo (si su módulo / aplicación es manage.py en la carpeta principal y usted está en un sistema operativo UNIX), ejecute:

export FLASK_APP=manage

Ejecutar frasco shell

flask shell

Importar lo que necesitamos ::

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

Ejecute su consulta:

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

Esto utiliza la conexión de base de datos actual que tiene la aplicación.

¿Has intentado usar connection.execute(text( <sql here> ), <bind params here> )y vincular parámetros como se describe en los documentos ? Esto puede ayudar a resolver muchos problemas de formato y rendimiento de parámetros. Tal vez el error de puerta de enlace es un tiempo de espera? Los parámetros de enlace tienden a hacer que las consultas complejas se ejecuten sustancialmente más rápido.

Si se quiere evitar tuplas, otra forma es llamando a los first, oneo allmétodos:

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

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