如何在 Flask-SQLAlchemy 应用程序中执行原始 SQL

在本文中，我们将了解如何使用Python在 Flask-SQLAlchemy 中执行原始 SQL。

安装要求

使用 pip 安装 Flask 和 Flask-SQLAlchemy 库

pip install Flask
pip install flask_sqlalchemy

句法

要运行原始 SQL 查询，我们首先创建一个 flask-SQLAlchemy 引擎对象，使用它我们可以连接到数据库并执行 SQL 查询。语法是——

flask_sqlalchemy.SQLAlchemy.engine.execute(statement)

Executes a SQL expression construct or string statement within the current transaction.

Parameters:

• statement: SQL expression

Returns:

• sqlalchemy.engine.result.ResultProxy

示例 1

Python

# 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 
    ('[e-mail]', 'John', 'Doe', 'john@123');
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('[e-mail]', 'John', 'Doe', 'johndoe@777');
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('[e-mail]', 'Emma', 'Noah', 'emaaa!00');
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('[e-mail]', 'Emma', 'Noah', 'whrfc2bfh904');
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('[e-mail]', 'Emma', 'Noah', 'emaaa!00');
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('[e-mail]', 'Liam', 'Olivia', 'lolivia#900');
    INSERT INTO users(email, first_name, last_name, passwd) VALUES 
    ('[e-mail]', '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()

输出：

在这个例子中，我们创建了一个简单的烧瓶应用程序，它没有任何路由，而是运行原始 SQL 查询。我们创建了 SQLAlchemy 连接，然后执行了 3 个不同的原始 SQL 查询。第一个查询创建用户表。第二个查询在表中插入一些示例记录。第三个查询获取所有记录并将它们显示在终端中。

在所有三种情况下，我们都使用了db.engine.execute()方法。 db.engine 提供了一个 SQLAlchemy 引擎连接，并且 execute 方法接受一个 SQL 查询来执行请求。

示例 2

在此示例中，我们创建了 2 条不同的路线供使用。这些路由将充当 API，我们可以在其中发送带有查询键的 POST 请求。此查询键的值将是我们需要执行的原始 SQL 查询。 get_results API 将用于获取我们从 SELECT 查询中获得的记录。 execute_query API 用于执行原始 SQL 查询，如果查询成功与否，将返回响应消息。

Python

# 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()

输出：

我们将通过 POSTMAN 测试路由。以下是使用 POSTMAN 测试的 3 个案例。

1. 运行 SELECT 查询以通过 get_results API 获取所有记录

2. 接下来，我们将测试 execute_query API 是否有有效的 INSERT 查询

3. 最后，我们将随机查询并查看是否收到任何错误消息