Flask应用中如何实现index()执行完成后再运行read_and_exec()

我正在开发一款Flask应用，功能是接收用户输入并存入数据库，之后读取数据库最后一行数据，将其中的字符串在谷歌搜索并截取对应图片。目前已实现两个核心函数：

• index()：处理用户输入并将数据存入数据库
• read_and_exec()：读取数据库最后一行数据，执行谷歌搜索并保存截图

但现在无法让两个函数按顺序执行，使用回调会导致一个任务未完成就启动另一个，请问该如何放置read_and_exec()，保证index()完全执行后再运行它？

相关代码 #

1. index()及数据库定义

app.config['SQLALCHEMY_DATABASE_URI'] = "sqlite:///automationDB.db"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)


class Automate(db.Model):
    sno = db.Column(db.Integer, primary_key=True)
    title_name = db.Column(db.String(20), nullable=False)
    second_name = db.Column(db.String(20))
    third_name = db.Column(db.String(20))
    fourth_name = db.Column(db.String(20))

def index():
    if request.method=='POST':
        category = request.form['category']
        alternative_1 = request.form['searchtexts1']
        alternative_2 = request.form['searchtexts2']
        alternative_3 = request.form['searchtexts3']
        
        example = Automate(title_name=category, second_name = alternative_1, third_name = alternative_2, fourth_name = alternative_3 )
        db.session.add(example)
        db.session.commit()
        
    allTodo = Automate.query.all() 
    return render_template('index.html', allTodo=allTodo)

contents = sqlite3.connect('automationDB.db')
content = contents.cursor()

2. read_and_exec()函数

def read_and_exec():
    last_row = content.execute('select * from automate').fetchall()[-1]

    os.mkdir("C:\\Users\\susha\\OneDrive\\Desktop\\Automation Project\\Content Images\\"+last_row[1]+" Images\\")
   
    os.environ['PATH'] += r"C:/Users/susha/OneDrive/Desktop/Automation Project"

    options = webdriver.ChromeOptions()
    options.add_experimental_option('excludeSwitches', ['enable-logging'])
    driver = webdriver.Chrome(options=options)

    driver.get('https://www.google.ca/imghp?hl=en&tab=ri&authuser=0&ogbl')
    
    my_element = driver.find_element(By.XPATH, '/html/body/div[1]/div[3]/form/div[1]/div[1]/div[1]/div/div[2]/input')
    for x in range(1,5):  
        my_element.send_keys(last_row[x])
        my_element.send_keys(Keys.ENTER)
       
        for i in range(1,10):
          sc_content = "C:\\Users\\susha\\OneDrive\\Desktop\\Automation Project\\Content Images\\"+last_row[1]+" Images\\"+last_row[x]+" ("+ str(i)+ ").png"
          driver.find_element(By.XPATH, '//*[@id="islrg"]/div[1]/div['+str(i)+']/a[1]/div[1]/img').screenshot(sc_content)

if __name__ == "__main__":
  app.run()

解决方案 #

方案1：同步调用（简单直接，适合测试场景）

在index()的POST请求处理完成、数据提交到数据库后，直接调用read_and_exec()，确保数据已持久化后再执行后续操作：

def index():
    if request.method=='POST':
        category = request.form['category']
        alternative_1 = request.form['searchtexts1']
        alternative_2 = request.form['searchtexts2']
        alternative_3 = request.form['searchtexts3']
        
        example = Automate(title_name=category, second_name=alternative_1, third_name=alternative_2, fourth_name=alternative_3)
        db.session.add(example)
        db.session.commit()
        
        # 数据已存入数据库，立即执行截图任务
        read_and_exec()
        
    allTodo = Automate.query.all() 
    return render_template('index.html', allTodo=allTodo)

注意：该方式会阻塞用户请求，直到截图任务完成，可能导致用户等待时间过长甚至请求超时，仅适合测试或低并发场景。

方案2：异步调用（推荐，不影响用户体验）

使用线程在后台执行read_and_exec()，让用户提交数据后立即得到响应：

import threading

def index():
    if request.method=='POST':
        category = request.form['category']
        alternative_1 = request.form['searchtexts1']
        alternative_2 = request.form['searchtexts2']
        alternative_3 = request.form['searchtexts3']
        
        example = Automate(title_name=category, second_name=alternative_1, third_name=alternative_2, fourth_name=alternative_3)
        db.session.add(example)
        db.session.commit()
        
        # 启动后台线程执行截图任务，不阻塞请求响应
        threading.Thread(target=read_and_exec).start()
        
    allTodo = Automate.query.all() 
    return render_template('index.html', allTodo=allTodo)

方案3：统一数据库操作（修复潜在数据不一致问题）

你的代码同时使用了SQLAlchemy和原生sqlite3连接，可能导致数据读取不一致。建议统一用SQLAlchemy读取数据，修改read_and_exec()：

import os
from selenium.webdriver.common.keys import Keys

def read_and_exec():
    # 使用SQLAlchemy读取最后一行数据
    last_row = Automate.query.order_by(Automate.sno.desc()).first()
    if not last_row:
        return
    
    # 先判断目录是否存在，避免重复创建报错
    dir_path = f"C:\\Users\\susha\\OneDrive\\Desktop\\Automation Project\\Content Images\\{last_row.title_name} Images\\"
    if not os.path.exists(dir_path):
        os.mkdir(dir_path)
   
    os.environ['PATH'] += r"C:/Users/susha/OneDrive/Desktop/Automation Project"

    options = webdriver.ChromeOptions()
    options.add_experimental_option('excludeSwitches', ['enable-logging'])
    driver = webdriver.Chrome(options=options)

    try:
        driver.get('https://www.google.ca/imghp?hl=en&tab=ri&authuser=0&ogbl')
        my_element = driver.find_element(By.XPATH, '/html/body/div[1]/div[3]/form/div[1]/div[1]/div[1]/div/div[2]/input')
        
        # 遍历字段，处理空值情况
        for field in [last_row.title_name, last_row.second_name, last_row.third_name, last_row.fourth_name]:
            if not field:
                continue
            my_element.clear()
            my_element.send_keys(field)
            my_element.send_keys(Keys.ENTER)
            
            for i in range(1,10):
                sc_content = f"{dir_path}{field} ({i}).png"
                try:
                    driver.find_element(By.XPATH, f'//*[@id="islrg"]/div[1]/div[{i}]/a[1]/div[1]/img').screenshot(sc_content)
                except Exception as e:
                    print(f"截图失败：{e}")
                    continue
    finally:
        driver.quit()  # 确保浏览器关闭

内容的提问来源于stack exchange，提问作者BigPy