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

En este artículo, veremos cómo ejecutar SQL sin procesar en Flask-SQLAlchemy usando Python.

Requisitos de instalación

Instale las bibliotecas Flask y Flask-SQLAlchemy usando pip

pip install Flask
pip install flask_sqlalchemy

Sintaxis

Para ejecutar consultas SQL sin procesar, primero creamos un objeto de motor de frasco-SQLAlchemy mediante el cual podemos conectarnos a la base de datos y ejecutar las consultas SQL. La sintaxis es –

matraz_sqlalchemy.SQLAlchemy.engine.execute(declaración)

Ejecuta una construcción de expresión SQL o una declaración de string dentro de la transacción actual.

Parámetros:

• instrucción: expresión SQL

Devoluciones:

• sqlalchemy.engine.result.ResultProxy

Ejemplo 1

# IMPORT REQUIRED LIBRARIES
from flask import Flask, request
from flask_sqlalchemy import SQLAlchemy
  
# CREATE THE FLASK APP
app = Flask(__name__)
  
# ADD THE DATABASE CONNECTION TO THE FLASK APP
db  = SQLAlchemy(app)
db_cred = {
    'user': 'root',         # DATABASE USER
    'pass': 'password',     # DATABASE PASSWORD
    'host': '127.0.0.1',    # DATABASE HOSTNAME
    'name': 'Geeks4Geeks'   # DATABASE NAME
}
app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://\
{db_cred['user']}:{db_cred['pass']}@{db_cred['host']}/\
{db_cred['name']}"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
  
# CREATE A users TABLE USING RAW SQL QUERY
db.engine.execute(
    '''
    CREATE TABLE users (
        email VARCHAR(50),
        first_name VARCHAR(50),
        last_name VARCHAR(50),
        passwd VARCHAR(50)
    );
    '''
)
  
# INSERT TEMP VALUES IN THE users TABLE USING RAW SQL QUERY
db.engine.execute(
    '''
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('john.doe@zmail.com', 'John', 'Doe', 'john@123');
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('john.doe@zmail.com', 'John', 'Doe', 'johndoe@777');
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('noah.emma@wmail.com', 'Emma', 'Noah', 'emaaa!00');
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('emma@tmail.com', 'Emma', 'Noah', 'whrfc2bfh904');
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('noah.emma@wmail.com', 'Emma', 'Noah', 'emaaa!00');
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('liam.olivia@wmail.com', 'Liam', 'Olivia', 'lolivia#900');
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('liam.olivia@wmail.com', 'Liam', 'Olivia', 'lolivia$345');
    '''
)
  
# VIEW THE RECORDS INSERTED
for record in db.engine.execute('SELECT * FROM users;'):
    print(record)
  
  
# RUN THE APP
if __name__ == '__main__':
    app.run()

Producción:

En este ejemplo, creamos una aplicación de matraz simple que no tiene ninguna ruta, sino que ejecuta consultas SQL sin formato. Creamos la conexión SQLAlchemy y luego ejecutamos 3 consultas SQL sin procesar diferentes. La primera consulta crea la tabla del usuario. La segunda consulta inserta algunos registros de muestra en la tabla. La tercera consulta obtiene todos los registros y los muestra en la terminal.

En los tres casos, hemos utilizado el método db.engine.execute() . db.engine proporciona una conexión de motor SQLAlchemy y el método de ejecución toma una consulta SQL para ejecutar la solicitud.

Ejemplo 2

En este ejemplo, hemos creado 2 rutas diferentes para trabajar. Estas rutas actuarán como una API donde podemos enviar una solicitud POST con una clave de consulta en el cuerpo. El valor de esta clave de consulta será la consulta SQL sin formato que debemos ejecutar. La API get_results se usará para obtener los registros que obtenemos de la consulta SELECT. La API de execute_query se usa para ejecutar consultas de SQL sin procesar y devolverá el mensaje de respuesta si la consulta se ejecuta con éxito o no.

# IMPORT REQUIRED LIBRARIES
from flask import Flask, request
from flask_sqlalchemy import SQLAlchemy
  
# CREATE THE FLASK APP
app = Flask(__name__)
  
# ADD THE DATABASE CONNECTION TO THE FLASK APP
db  = SQLAlchemy(app)
db_cred = {
    'user': 'root',         # DATABASE USER
    'pass': 'password',     # DATABASE PASSWORD
    'host': '127.0.0.1',    # DATABASE HOSTNAME
    'name': 'Geeks4Geeks'   # DATABASE NAME
}
app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://\
{db_cred['user']}:{db_cred['pass']}@{db_cred['host']}/\
{db_cred['name']}"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
  
# APP ROUTE TO GET RESULTS FOR SELECT QUERY
@app.route('/get_results', methods=['POST'])
def get_results():
      
    # GET THE SQLALCHEMY RESULTPROXY OBJECT
    result = db.engine.execute(request.get_json()['query'])
    response = {}
    i = 1
  
    # ITERATE OVER EACH RECORD IN RESULT AND ADD IT 
    # IN A PYTHON DICT OBJECT
    for each in result:
        response.update({f'Record {i}': list(each)})
        i+= 1
  
    return response
  
# APP ROUTE TO RUN RAW SQL QUERIES
@app.route('/execute_query', methods=['POST'])
def execute_query():
  
    try:
        db.engine.execute(request.get_json()['query'])
    except:
        return {"message": "Request could not be completed."}
  
    return {"message": "Query executed successfully."}
  
  
# RUN THE APP
if __name__ == '__main__':
    app.run()

Producción:

Probaremos las rutas a través de POSTMAN. A continuación se muestran los 3 casos que se prueban con POSTMAN.

1. Ejecutar una consulta SELECT para obtener todos los registros a través de la API get_results
 

2. A continuación, probaremos la API execute_query para una consulta INSERT válida

3. Por último, pondremos cualquier consulta aleatoria y veremos si recibimos algún mensaje de error.