多数据库路由

官方文档：https://tortoise.github.io/databases.html
适用版本：tortoise-orm 0.17+（2026-05-08 核实）
最后更新：2026-04-11

一、配置多个数据库连接

在 Tortoise.init() 的 config 中，connections 字典支持定义任意数量的命名连接。

1.1 connections 配置参数

每个连接支持两种写法：URL 字符串或完整 credentials 字典。

URL 字符串写法：

TORTOISE_ORM = {
    "connections": {
        "default": "mysql://user:pass@localhost:3306/main_db",
        "replica": "postgres://user:pass@replica-host:5432/read_db",
    },
    ...
}

完整 credentials 写法（可配置连接池）：

TORTOISE_ORM = {
    "connections": {
        "default": {
            "engine": "tortoise.backends.mysql",
            "credentials": {
                "host": "localhost",
                "port": 3306,
                "user": "myuser",
                "password": "mypass",
                "database": "main_db",
                "minsize": 1,
                "maxsize": 20,
                "connect_timeout": 5,
            },
        },
        "replica": {
            "engine": "tortoise.backends.asyncpg",
            "credentials": {
                "host": "replica-host",
                "port": 5432,
                "user": "readonly_user",
                "password": "readpass",
                "database": "read_db",
                "minsize": 2,
                "maxsize": 50,
            },
        },
    },
    ...
}

credentials 通用参数：

二、apps 中指定 default_connection

每个 app 可以绑定到不同的数据库连接，通过 default_connection 指定：

TORTOISE_ORM = {
    "connections": {
        "primary": "mysql://user:pass@primary-host:3306/main_db",
        "analytics": "postgres://user:pass@analytics-host:5432/analytics_db",
    },
    "apps": {
        "models": {
            "models": ["myapp.models", "aerich.models"],
            "default_connection": "primary",   # 此 app 下的模型默认使用 primary
        },
        "analytics": {
            "models": ["myapp.analytics.models"],
            "default_connection": "analytics", # analytics app 使用独立数据库
        },
    },
}

同一 app 下的所有模型默认使用该 app 的 default_connection，除非模型自身通过 Meta 覆盖。

三、Model.Meta.using 绑定模型到指定数据库

在模型定义中通过 Meta.using 硬编码模型与数据库连接的绑定关系：

from tortoise.models import Model
from tortoise import fields


class AnalyticsEvent(Model):
    id = fields.IntField(pk=True)
    event_type = fields.CharField(max_length=100)
    user_id = fields.IntField()
    occurred_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "analytics_event"
        using = "analytics"   # 所有对此模型的操作默认使用 analytics 连接


class User(Model):
    id = fields.IntField(pk=True)
    username = fields.CharField(max_length=50)

    class Meta:
        table = "user"
        # 不指定 using，使用所在 app 的 default_connection

四、查询时用 using_db 切换数据库

即使模型已绑定默认连接，也可以在单次查询时通过 using_db 临时切换：

from tortoise import Tortoise


# 获取指定连接
replica_conn = Tortoise.get_connection("replica")

# 查询时指定连接
users = await User.all().using_db(replica_conn)
user = await User.filter(id=1).using_db(replica_conn).first()

# 单次写操作切换到主库
primary_conn = Tortoise.get_connection("primary")
await User.create(username="alice", using_db=primary_conn)

using_db 参数说明：

五、读写分离配置模式

5.1 配置结构

TORTOISE_ORM = {
    "connections": {
        "master": {
            "engine": "tortoise.backends.mysql",
            "credentials": {
                "host": "primary.db.internal",
                "port": 3306,
                "user": "app_user",
                "password": "strongpass",
                "database": "mydb",
                "maxsize": 20,
            },
        },
        "slave": {
            "engine": "tortoise.backends.mysql",
            "credentials": {
                "host": "replica.db.internal",
                "port": 3306,
                "user": "readonly_user",
                "password": "readpass",
                "database": "mydb",
                "maxsize": 50,   # 读库可以配更大连接池
            },
        },
    },
    "apps": {
        "models": {
            "models": ["myapp.models", "aerich.models"],
            "default_connection": "master",  # 默认写主库
        }
    },
}

5.2 封装读写路由工具

# db.py
from tortoise import Tortoise


def get_master():
    """获取主库连接（写操作）"""
    return Tortoise.get_connection("master")


def get_slave():
    """获取从库连接（读操作）"""
    return Tortoise.get_connection("slave")

5.3 在 Service 层应用读写分离

# services/user_service.py
from myapp.db import get_master, get_slave
from myapp.models import User


async def get_user(user_id: int) -> User | None:
    """读操作走从库"""
    return await User.get_or_none(id=user_id).using_db(get_slave())


async def list_users() -> list[User]:
    """读操作走从库"""
    return await User.all().using_db(get_slave())


async def create_user(username: str, email: str) -> User:
    """写操作走主库"""
    return await User.create(username=username, email=email, using_db=get_master())


async def update_user(user_id: int, **kwargs) -> int:
    """写操作走主库"""
    return await User.filter(id=user_id).using_db(get_master()).update(**kwargs)

六、完整配置示例（MySQL 主库 + PostgreSQL 只读副本）

这种架构适合数据写入用 MySQL，分析查询用 PostgreSQL（通过 ETL 同步数据）的场景。

# config.py
import os

TORTOISE_ORM = {
    "connections": {
        "default": {
            "engine": "tortoise.backends.mysql",
            "credentials": {
                "host": os.getenv("MYSQL_HOST", "localhost"),
                "port": int(os.getenv("MYSQL_PORT", "3306")),
                "user": os.getenv("MYSQL_USER", "app_user"),
                "password": os.getenv("MYSQL_PASSWORD", ""),
                "database": os.getenv("MYSQL_DATABASE", "mydb"),
                "minsize": 2,
                "maxsize": 20,
                "connect_timeout": 10,
                "ssl": os.getenv("MYSQL_SSL", "false").lower() == "true",
            },
        },
        "pg_readonly": {
            "engine": "tortoise.backends.asyncpg",
            "credentials": {
                "host": os.getenv("PG_HOST", "localhost"),
                "port": int(os.getenv("PG_PORT", "5432")),
                "user": os.getenv("PG_USER", "readonly_user"),
                "password": os.getenv("PG_PASSWORD", ""),
                "database": os.getenv("PG_DATABASE", "analytics_db"),
                "minsize": 1,
                "maxsize": 30,
            },
        },
    },
    "apps": {
        "models": {
            "models": ["myapp.models", "aerich.models"],
            "default_connection": "default",   # 主要业务模型走 MySQL
        },
        "reports": {
            "models": ["myapp.reports.models"],
            "default_connection": "pg_readonly",  # 报表模型走 PostgreSQL
        },
    },
    "use_tz": True,
    "timezone": "Asia/Shanghai",
}

# models.py
from tortoise.models import Model
from tortoise import fields


class User(Model):
    """主业务模型，存储于 MySQL"""
    id = fields.IntField(pk=True)
    username = fields.CharField(max_length=50, unique=True)
    email = fields.CharField(max_length=255)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "user"
        # default_connection 由 app 的 default_connection 决定（即 "default"/MySQL）


# reports/models.py
class UserReport(Model):
    """报表模型，读取 PostgreSQL 中的只读视图"""
    user_id = fields.IntField()
    total_orders = fields.IntField()
    total_spend = fields.DecimalField(max_digits=12, decimal_places=2)
    report_date = fields.DateField()

    class Meta:
        table = "v_user_report"   # PostgreSQL 视图
        using = "pg_readonly"     # 明确绑定到 PostgreSQL 连接

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from tortoise import Tortoise
from myapp.config import TORTOISE_ORM


@asynccontextmanager
async def lifespan(app: FastAPI):
    await Tortoise.init(config=TORTOISE_ORM)
    yield
    await Tortoise.close_connections()


app = FastAPI(lifespan=lifespan)

七、事务跨数据库注意事项

7.1 事务只在单个连接内有效

Tortoise ORM 的事务绑定到具体连接，不支持跨数据库的分布式事务。

from tortoise.transactions import in_transaction


# 正确：事务限定在单个连接内
async with in_transaction("default") as conn:
    user = await User.create(username="alice", using_db=conn)
    await Profile.create(user_id=user.id, using_db=conn)
    # User 和 Profile 在同一个 "default" 数据库，事务有效


# 错误：跨库操作不在同一事务中
async with in_transaction("default") as conn:
    user = await User.create(username="alice", using_db=conn)
    # AnalyticsEvent 在 "analytics" 数据库，此操作不在上面的事务内
    await AnalyticsEvent.create(user_id=user.id)  # 独立执行，不受事务保护

跨数据库操作需要在应用层自行处理一致性，例如使用补偿事务或消息队列。

7.2 in_transaction 参数

from tortoise.transactions import in_transaction

async with in_transaction(connection_name="default") as conn:
    ...

八、踩坑与注意事项

8.1 事务不能跨数据库

如上文所述，in_transaction 只对指定连接生效。如果在 in_transaction("default") 上下文中对绑定到其他连接的模型执行写操作，该操作不在事务保护范围内，发生错误时不会自动回滚。

解决方案：将需要保持原子性的数据放在同一个数据库中。

8.2 外键约束只在同库有效

不同数据库之间的表不能建立数据库层面的外键约束，即使在 Tortoise 模型层定义了 ForeignKeyField，底层数据库也无法跨库执行约束检查。

class Order(Model):
    # 错误：user 在 MySQL，Order 在 PostgreSQL，数据库层的外键约束不生效
    user: fields.ForeignKeyRelation["User"] = fields.ForeignKeyField(
        "models.User", related_name="orders"
    )

    class Meta:
        using = "pg_readonly"

跨数据库的关联完整性需要在应用层代码中手动维护（查询前校验、删除前清理）。

8.3 aerich 迁移仅作用于绑定的连接

Aerich 迁移针对的是 app 的 default_connection。若有多个 app 对应不同数据库，需要分别运行迁移：

aerich upgrade --app models      # 迁移 MySQL 主库
aerich upgrade --app analytics   # 迁移 PostgreSQL 分析库

详见 Aerich迁移指南 中的多应用配置章节。

8.4 连接名称拼写错误导致静默失败

Tortoise.get_connection("typo_name") 在连接名不存在时会抛出 ConfigurationError，但如果连接名在 Meta.using 中写错，会在第一次查询时才报错，而非启动时。

建议将连接名定义为常量：

# constants.py
DB_PRIMARY = "primary"
DB_REPLICA = "replica"
DB_ANALYTICS = "analytics"

class AnalyticsEvent(Model):
    class Meta:
        using = DB_ANALYTICS  # 引用常量，避免拼写错误

最佳实践

用常量代替魔法字符串命名连接：DB_DEFAULT = "default" 定义常量，在 Meta.using 和 using_db() 中引用常量，拼写错误在开发期即可发现（IDE 提示）。

跨库事务必须用同一数据库连接：Tortoise ORM 的事务绑定单个数据库连接，跨不同数据库的操作无法放在同一原子事务中，需在应用层设计补偿机制（Saga 模式）。

只读副本用 using_db("readonly") 路由：将耗时报表查询路由到只读副本，减轻主库压力；读写分离时注意主从延迟，避免写后立即从副本读取。

Tortoise.get_connection() 获取连接做原生 SQL：需要多数据库的复杂 SQL 时，用 conn = Tortoise.get_connection("analytics") 后 await conn.execute_query(sql) 精确指定连接。

aerich 迁移按 app 独立管理：多数据库场景下，每个 app 的迁移文件目录独立，aerich upgrade --app <name> 分别执行，避免迁移历史混淆。

常见陷阱

陷阱：未配置 Meta.using 的模型路由到意外的数据库

现象： 新模型未设置 Meta.using，查询时报找不到表，或写入到了错误的数据库。
原因： 没有 Meta.using 的模型默认路由到 "default" 连接，若该模型的表在其他数据库则找不到。
解决： 严格为非 default 数据库的模型设置 Meta.using，并在 Code Review 中检查新 Model 是否遗漏。

陷阱：register_tortoise 中 apps 配置遗漏模型

现象： 多数据库配置时某个 app 的模型不生效，查询报表不存在。
原因： TORTOISE_ORM["apps"] 中 models 列表未包含该模型的模块路径。
解决： 检查 apps 配置，确保所有模型模块路径都在正确 app 的 models 列表中；generate_schemas=True 调试时可检查哪些表被创建。

陷阱：在事务中混用两个不同数据库的操作

现象： @atomic() 装饰的函数中同时操作 default 和 analytics 数据库，其中一个出错后另一个没有回滚。
原因： @atomic() 的事务只能绑定单一连接，跨连接操作无法原子化。
解决： 设计时避免跨库原子操作；如果必须保证一致性，用消息队列 + 幂等操作 + 补偿事务实现最终一致性。

参见

初始化与配置
Aerich迁移指南
查询操作完全指南