在Flask应用中使用Tortoise ORM初始化数据库及调用异步代码的方法

看起来你在整合Flask和Tortoise ORM时踩了两个典型的坑：数据库初始化时机错误，以及在Flask的同步请求线程中调用异步函数时的事件循环问题。我来一步步帮你解决：

一、数据库初始化的正确姿势 #

绝对不要在每个请求的异步函数里调用await db.init()——这会重复创建数据库连接，既浪费资源又影响性能。正确的做法是在Flask应用启动时一次性完成初始化：

首先定义你的Tortoise配置和初始化函数：

from flask import Flask
from tortoise import Tortoise

app = Flask(__name__)

# 替换成你的数据库配置
TORTOISE_ORM = {
    "connections": {
        "default": "sqlite:///your_database.db"  # 可替换为PostgreSQL、MySQL等数据库链接
    },
    "apps": {
        "your_app_name": {
            "models": ["your_app.models"],  # 指向你的模型文件所在路径
            "default_connection": "default",
        }
    }
}

async def init_db():
    # 初始化Tortoise连接
    await Tortoise.init(config=TORTOISE_ORM)
    # 自动生成表结构（生产环境建议改用迁移工具，关闭此选项）
    await Tortoise.generate_schemas()

然后在应用启动时执行初始化：

if __name__ == "__main__":
    # 用Tortoise的run_async在主线程运行初始化逻辑
    from tortoise import run_async
    run_async(init_db())
    app.run(debug=True)

如果是用WSGI服务器（比如Gunicorn）部署，没法直接在__main__里执行，可以用标记变量配合before_request确保只初始化一次：

db_initialized = False

@app.before_request
def setup_db():
    global db_initialized
    if not db_initialized:
        from tortoise import run_async
        run_async(init_db())
        db_initialized = True

二、在Flask请求中调用异步数据库操作 #

你遇到的RuntimeError: There is no current event loop in thread 'Thread-7'错误，是因为Flask默认会为每个请求分配独立的子线程，而asyncio不会在子线程中自动创建事件循环。这里有两种靠谱的解决方式：

方式1：手动创建并管理事件循环 #

直接在请求处理函数里创建事件循环，运行异步任务后关闭：

import asyncio
from flask import request, jsonify, render_template

@doc_income_app.route("new", methods=['GET', 'POST'])
def new():
    if request.method == 'GET':
        return render_template('doc_income/new.html')
    else:
        # 创建新的事件循环并设置为当前线程的默认循环
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        # 运行异步任务直到完成
        loop.run_until_complete(create_doc(request.values['account_id'], request.values['sum']))
        loop.close()
        return jsonify(ok=True)

async def create_doc(account_id, sum):
    # 无需再初始化db，启动时已完成全局初始化
    doc = db.DocIncome(
        account_id=account_id,
        sum=sum
    )
    await doc.save()

方式2：用asgiref.sync简化异步转同步 #

这是更简洁的方案，asgiref的async_to_sync工具会自动帮你处理事件循环的创建和管理：

先安装依赖：

pip install asgiref

然后修改请求处理函数：

from asgiref.sync import async_to_sync
from flask import request, jsonify, render_template

@doc_income_app.route("new", methods=['GET', 'POST'])
def new():
    if request.method == 'GET':
        return render_template('doc_income/new.html')
    else:
        # 直接将异步函数转为同步调用
        async_to_sync(create_doc)(request.values['account_id'], request.values['sum'])
        return jsonify(ok=True)

async def create_doc(account_id, sum):
    doc = db.DocIncome(
        account_id=account_id,
        sum=sum
    )
    await doc.save()

额外提醒 #

• 尽量避免在异步函数里做阻塞操作（比如同步IO），否则会抵消异步的性能优势。
• 如果你的应用有大量异步操作需求，更推荐使用支持ASGI的框架（比如FastAPI）配合Tortoise ORM，这样异步代码的编写会更自然流畅。

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