如何在FastAPI+Tortoise ORM中设置总超时（连接+查询执行）限制

核心说明 #

Tortoise-ORM 针对 MySQL 底层依赖 asyncmy 异步驱动，超时参数需匹配驱动规则；同时Tortoise本身无内置全局查询超时配置，需结合asyncio工具实现。

1. 正确配置数据库连接超时 #

连接超时控制建立数据库连接的最大等待时间，两种有效配置方式：

URI 配置方式

在连接字符串末尾追加 connect_timeout 参数（单位：秒）：

import os
from dotenv import load_dotenv

load_dotenv()
DATABASE_CONFIG = {
    'connections': {
        'myconnection': f"mysql://{os.getenv('DB_USERNAME', 'root')}:{os.getenv('DB_PASSWORD', '')}@your-host.com:3306/mydb?connect_timeout=5"
    },
    'apps': {
        'your_app': {
            'models': ['your_project.models'],
            'default_connection': 'myconnection',
        }
    }
}

字典配置方式

直接在 credentials 字典中添加 connect_timeout 参数，Tortoise会自动传递给asyncmy驱动：

import os
from dotenv import load_dotenv

load_dotenv()
DATABASE_CONFIG = {
    'connections': {
        'myconnection': {
            'engine': 'tortoise.backends.mysql',
            'credentials': {
                'host': os.getenv('DB_HOST', 'mysql'),
                'port': int(os.getenv('DB_PORT', 3306)),
                'user': os.getenv('DB_USERNAME', 'root'),
                'password': os.getenv('DB_PASSWORD', ''),
                'database': os.getenv('DB_NAME', 'mydb'),
                'connect_timeout': 5  # 连接超时，单位秒
            }
        },
    },
    'apps': {
        'your_app': {
            'models': ['your_project.models'],
            'default_connection': 'myconnection',
        }
    }
}

2. 设置查询执行超时 #

Tortoise无全局默认查询超时，可通过以下两种方式实现：

单个查询单独设置

用 asyncio.wait_for 包裹查询操作，指定超时时间：

from asyncio import wait_for, TimeoutError
from fastapi import HTTPException
from your_project.models import YourModel

async def get_data():
    try:
        # 限制该查询最多执行3秒
        result = await wait_for(YourModel.all().values(), timeout=3)
        return result
    except TimeoutError:
        raise HTTPException(status_code=500, detail="查询超时，请稍后重试")

全局统一设置查询超时

通过抽象模型类包装所有查询方法，实现全局超时控制：

from tortoise.models import Model
from asyncio import wait_for, TimeoutError
from fastapi import HTTPException

class TimeoutBaseModel(Model):
    class Meta:
        abstract = True

    @classmethod
    async def all(cls, *args, **kwargs):
        try:
            return await wait_for(super().all(*args, **kwargs), timeout=5)
        except TimeoutError:
            raise HTTPException(status_code=500, detail="全局查询超时")

    # 可继续包装get、filter等其他查询方法
    @classmethod
    async def get(cls, *args, **kwargs):
        try:
            return await wait_for(super().get(*args, **kwargs), timeout=5)
        except TimeoutError:
            raise HTTPException(status_code=500, detail="全局查询超时")

# 业务模型继承该抽象类
class YourModel(TimeoutBaseModel):
    name = fields.CharField(max_length=50)
    # 其他字段...

3. 连接+查询总超时控制 #

若需限制建立连接+执行查询的总时间，直接在接口层面用wait_for包裹整个数据库操作流程：

from fastapi import FastAPI, HTTPException
from asyncio import wait_for, TimeoutError
from your_project.services import get_data

app = FastAPI()

@app.get("/api/data")
async def fetch_data():
    try:
        # 总超时8秒（包含连接建立和查询执行的时间）
        result = await wait_for(get_data(), timeout=8)
        return {"data": result}
    except TimeoutError:
        raise HTTPException(status_code=500, detail="请求超时，请稍后重试")

你之前配置的问题说明 #

你之前在options字典中添加的timeout、socketTimeout等参数，Tortoise的MySQL引擎不会传递给asyncmy驱动，因此无法生效。连接相关参数必须放在credentials中，查询超时则需要通过asyncio.wait_for手动实现。

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