当前位置：首页 > news >正文

Flask实现异步调用sqlalchemy的模型类

news 2025/7/27 21:43:44

事情是这样的，我这边需要在一次请求里面，搞一个异步不阻碍的任务，来执行耗时的操作。
一开始，我准备写的代码是这样的：

from flask import Flask
import time
from concurrent.futures import ThreadPoolExecutorexecutor = ThreadPoolExecutor(1)app = Flask(__name__)@app.route('/synchronize')
def update_redis():executor.submit(do_update)return 'ok'def do_update():time.sleep(3)print('start update')if __name__ == '__main__':app.run()

但是，异步函数里面没有app的上下文，没办法调用sqlalchemy模型类、config内容、logger等。
所以，我封装成了一个接口，异步只负责发起耗时请求，这样就可以在耗时请求TtsCopyVoiceRaw里面正常使用app上下文了：

# 复制音色 异步
class TtsCopyVoiceRaw(Resource):def post(self):user_id = request.json.get('user_id')name = request.json.get('name')speed = request.json.get('speed', 0.75)audio_url = request.json.get('audio_url')current_app.logger.info('copy_voice 1 --------------')current_app.logger.info(user_id, name, audio_url, speed)try:voice_id = TTSApi().copy_voice(user_id, name, audio_url, speed)audition_url = TTSApi().tts(user_id, '欢迎使用直播助手，我是{}'.format(name), voice_id, speed)current_app.logger.info('copy_voice voice_id --------------')current_app.logger.info(voice_id)voice_obj = db.session.query(UserTimbre).filter(UserTimbre.user_id == user_id,UserTimbre.name == name,).first()voice_obj.voice_id = voice_idvoice_obj.audition_url = audition_urlvoice_obj.status = 3db.session.commit()except Exception as e:current_app.logger.info('copy_voice failed --------------')current_app.logger.info(traceback.format_exc())voice_obj = db.session.query(UserTimbre).filter(UserTimbre.user_id == user_id,UserTimbre.name == name,).first()voice_obj.status = 4db.session.commit()def copy_voice(full_path, user_id, name, audio_url, speed):print('copy_voice -----------')print(full_path, user_id, name, audio_url, speed)requests.post(full_path + '/raw', json=dict(user_id=user_id,name=name,audio_url=audio_url,speed=speed,))# 复制音色
class TtsCopyVoice(ResourceApiPassed):def post(self):user_id = self.user_info.get('id')name = request.json.get('name')speed = request.json.get('speed', 0.75)audio_url = request.json.get('audio_url')total = db.session.query(UserTimbre).filter(UserTimbre.user_id == user_id,UserTimbre.name == name,).count()if total:raise APIException('名称已存在，请改名重新创建')current_app.logger.warn('TtsCopyVoice ******* 1')current_app.logger.warn(user_id)current_app.logger.warn(name)current_app.logger.warn(speed)current_app.logger.warn(audio_url)try:current_app.logger.warn('obj ------------- 0')# 创建 用户音色obj = UserTimbre(user_id = user_id,name = name,status = 1,)db.session.add(obj)db.session.commit()except Exception as e:current_app.logger.warn('e ------------- 0')current_app.logger.warn(e)current_app.logger.warn('e ------------- 1')current_app.logger.warn(traceback.format_exc())raise APIException('创建失败')current_app.executor.submit(copy_voice, request.url, user_id, name, audio_url, speed)return success({'id': obj.id})