正在进入ing...

FastAPI 配置多数据库踩坑记录

发布时间:2025-12-22 浏览量: 143 文章分类: python

FastAPI 配置多数据库踩坑记录

先说一下背景， 我的FastAPI的库相对简单，主要就是通过鉴权接收日志，然后通过定时任务把redis的日志批量一起写入pgsql里面即可。 但是因为涉及从我们的老后台读取配置、身份鉴权列表等信息，所以FastAPI需要同时连接Mysql 、 Pgsql 、 Redis 3 个数据库。

我使用的数据库框架是Tortoise-orm,需要通过一个ORM来进行2 个数据库连接和绑定注册。 定时任务使用 apscheduler 框架整合在项目的crontab 目录下

项目结构

我还是喜欢遵循 Django的目录结构， 觉得这样非常清晰，所以基本也会在FastAPI创建遵循Django的模式，不过会把数据库注册 这些独立出来到database 目录下单独配置

├── config.py
├── core
│   ├── auth.py
│   ├── base_models.py
│   ├── event.py
│   ├── exception.py
│   ├── response.py
│   └── scheduler.py
├── crontab
│   ├── auto_save_logs.py
│   ├── auto_update_auth.py
│   └── crontab_main.py
├── database
│   ├── db.py
│   └── redis.py
├── docker-compose.yml
├── Dockerfile
├── game_report
│   ├── models
│   │   ├── report_logs_models.py
│   │   └── request_auth_models.py
│   ├── urls.py
│   ├── validaror
│   │   └── report_game_logs_pydantic.py
│   └── views
│       ├── find_logs_views.py
│       ├── get_report_status_views.py
│       └── report_game_logs_views.py
├── logs
│   └── webserver.log
├── main.py
├── pyproject.toml
├── README.md
├── requirements.txt
├── router
│   └── router.py
├── start.sh
├── utils
│   ├── logs.py
│   └── pageintor.py
└── uv.lock

Redis注册

这个其实没什么坑点，因为我并没有把Redis和Tortoise绑定，而是使用redis这个库自己实现了一个连接池绑定到app上，在 event.py上面注册register_redis方法即可。

# 创建异步 Redis 连接池
async def create_redis_pool():
    # 使用 redis-py 的异步模块创建连接池
    is_server = settings.IS_SERVER
    if not is_server:
        # 开发
        redis_host = settings.DEV_REDIS_HOST
        redis_db = settings.DEV_REDIS_DB
        redis_port = settings.DEV_REDIS_PORT
        redis_password = settings.DEV_REDIS_PASSWORD
    else:
        # 线上生产环境
        redis_host = settings.PROD_REDIS_HOST
        redis_db = settings.PROD_REDIS_DB
        redis_port = settings.PROD_REDIS_PORT
        redis_password = settings.PROD_REDIS_PASSWORD

    redis_url = f"redis://:{redis_password}@{redis_host}:{redis_port}/{redis_db}"
    logger.debug(redis_url)
    pool = aioredis.ConnectionPool.from_url(
        url=redis_url,
        max_connections=500,
        encoding="utf-8",
        socket_keepalive=True,
        decode_responses=True,
        socket_connect_timeout=5,
        socket_timeout=10,
        retry_on_timeout=True,
        retry_on_error=[ConnectionError, TimeoutError],
        retry=Retry(ExponentialBackoff(cap=10, base=1), 3),
        health_check_interval=30,
    )
    return aioredis.Redis(connection_pool=pool)


async def register_redis(app: FastAPI):
    # 注册 Redis 连接池
    try:
        app.state.redis = await create_redis_pool()
        logger.debug("Redis 连接池注册完成")
    except Exception as e:
        logger.error(f"Redis 连接失败: {e}")
        raise


async def shutdown_redis(app: FastAPI):
    # 关闭 Redis 连接池
    try:
        await app.state.redis.close()
        await app.state.redis.wait_closed()
        logger.debug("Redis 连接池已关闭")
    except Exception as e:
        logger.error(f"关闭 Redis 连接池失败: {e}")

MySQL & PgSQL 注册的噩梦

开始我其实没有多想，是 mysql 一个注册方式、pgsql一个注册方式，然后继续注册就可以了。 实际证明这个方法是没问题的，但是如果你通过定时任务使用，就开始出现识别不到的问题了。 每次定时任务都会出现tortoise.exceptions.ConfigurationError: default_connection for the model <class 'game_report.models.report_logs_models.ReportGameLogsModels'> cannot be None类似这样的提示，开始我其实没有多想，因为觉得这个问题是先后注册顺序，或者没找到的问题造成的。

  • 调整注册顺序，增加返回，确保注册成功后在执行下一步 ， 无法解决
  • 调整数据库配置，在models文件指定app_label = 'xxx' ， 无法解决
  • 尝试仅注册 mysql \ pgsql ，无异常，可以正常执行，但是 2 个库就错了
  • 查资料，问 AI，给的解决方法更是各种离谱，各种瞎折腾配置，p 用没有

最终解决方法，先写了一个check_db.py ，但是把注册功能迁移出来，独立检查一下到底注册的情况。

import asyncio
from config import settings
from tortoise import Tortoise

async def check_config():
    config = {
        "connections": {
            "mysql": {
                "engine": "tortoise.backends.mysql",
                "credentials": {
                    "host": settings.MYSQL_DB_HOST,
                    "user": settings.MYSQL_DB_USER,
                    "password": settings.MYSQL_DB_PASSWORD,
                    "port": settings.MYSQL_DB_PORT,
                    "database": settings.MYSQL_DB_DATABASE,
                },
            },
            "pgsql": {
                "engine": "tortoise.backends.asyncpg",
                "credentials": {
                    "host": settings.PGSQL_DB_HOST,
                    "user": settings.PGSQL_DB_USER,
                    "password": settings.PGSQL_DB_PASSWORD,
                    "port": settings.PGSQL_DB_PORT,
                    "database": settings.PGSQL_DB_DATABASE,
                },
            }
        },
        "apps": {
            "mysql_app": {
                "models": ["game_report.models.request_auth_models"],
                "default_connection": "mysql",
            },
            "pg_app": {
                "models": ["game_report.models.report_logs_models"],
                "default_connection": "pgsql",
            },
        },
        "use_tz": False,
        "timezone": "Asia/Shanghai",
    }

    await Tortoise.init(config)

    for app_name in Tortoise.apps:
        print(f"  {app_name}:")
        for model_name, model_class in Tortoise.apps[app_name].items():
            print(f"    - {model_name}")
            print(f"      connection: {model_class._meta.db}")

    await Tortoise.close_connections()

asyncio.run(check_config())

具体折腾的过程就不说了， 这里是我已经测试过了，运行后会有如下输出才对

已注册的app:
  mysql_app:
    - RequestAuthModels
      connection: <tortoise.backends.mysql.client.MySQLClient object at 0x10b8e1f10>
  pg_app:
    - ReportGameLogsModels
      connection: <tortoise.backends.asyncpg.client.AsyncpgDBClient object at 0x10ba1f310>

如果没有输出 pg_app 或者 mysql_app ， 那都是配置出现问题，需要逐个检测。

通过这个配置文件， 相信已经看出问题所在了，就是Tortoise-rom 的 init 方法会覆盖，所以后面会覆盖前面的，导致 2 个库总是不能同时注册，所以最好的解决办法，是一次在DB_CONFIG中直接配置好 2 个数据库，然后一次注册即可，这样以后有多个数据库， 也只要在配置文件中修改即可。

同时附上完整代码。

# db.py
from fastapi import FastAPI
from tortoise.contrib.fastapi import RegisterTortoise
from utils.logs import logger
from config import settings


DB_ORM_CONFIG = {
    "connections": {
        "mysql": {
            "engine": "tortoise.backends.mysql",
            "credentials": {
                "host": settings.MYSQL_DB_HOST,
                "user": settings.MYSQL_DB_USER,
                "password": settings.MYSQL_DB_PASSWORD,
                "port": settings.MYSQL_DB_PORT,
                "database": settings.MYSQL_DB_DATABASE,
                "minsize": 1,
                "maxsize": 5,  # 减小连接池大小
                "connect_timeout": 30,  # 添加连接超时
                "echo": False,  # 开启SQL日志，方便调试
            },
        },
        "pgsql": {
            "engine": "tortoise.backends.asyncpg",
            "credentials": {
                "host": settings.PGSQL_DB_HOST,
                "user": settings.PGSQL_DB_USER,
                "password": settings.PGSQL_DB_PASSWORD,
                "port": settings.PGSQL_DB_PORT,
                "database": settings.PGSQL_DB_DATABASE,
                "minsize": 1,  # 最小连接数
                "maxsize": 10,  # 最大连接数
                "command_timeout": 30,  # 添加超时
                "server_settings": {
                    "application_name": "report_api"
                }
            },
        },
    },
    "apps": {
        "mysql_app": {
            "models": ["game_report.models.request_auth_models"],
            "default_connection": "mysql",
        },
        "pg_app": {
            "models": ["game_report.models.report_logs_models"],
            "default_connection": "pgsql",
        },
    },
    "use_tz": False,
    "timezone": "Asia/Shanghai"
}


async def register_db(app: FastAPI):
    register_tortoise = RegisterTortoise(
        app, config=DB_ORM_CONFIG, generate_schemas=True, add_exception_handlers=True
    )

    await register_tortoise.init_orm()

    logger.debug("db注册成功")