Django副数据库连接失败的静默降级与默认库Fallback方案

针对你的需求——ovi_db需VPN连接，失败时自动降级到默认库，同时解决启动崩溃问题，可通过以下步骤实现：

1. 自定义数据库后端，避免启动时强制连接崩溃 #

Django默认启动时会尝试连接所有配置的数据库，我们通过包装PostgreSQL后端，捕获连接异常并返回非致命状态：

在项目根目录创建custom_db_backends/psycopg2.py：

from django.db.backends.postgresql.base import DatabaseWrapper as PostgresWrapper
from psycopg2 import OperationalError
import logging

logger = logging.getLogger(__name__)

class DatabaseWrapper(PostgresWrapper):
    def get_new_connection(self, conn_params):
        try:
            return super().get_new_connection(conn_params)
        except OperationalError as e:
            logger.warning(f"ovi_db连接失败: {str(e)}")
            # 连接失败返回None，由后续路由处理降级
            return None

修改settings.py中ovi_db的配置，使用自定义后端并缩短连接超时：

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        # 你的默认数据库配置
        'NAME': 'default_db',
        'USER': 'user',
        'PASSWORD': 'pass',
        'HOST': 'localhost',
        'PORT': '5432',
    },
    'ovi_db': {
        'ENGINE': 'custom_db_backends.psycopg2',  # 替换为自定义后端
        'NAME': 'ovi_db',
        'USER': 'ovi_user',
        'PASSWORD': 'ovi_pass',
        'HOST': 'ovi_vpn_host',
        'PORT': '5432',
        'OPTIONS': {
            'connect_timeout': 5,  # 缩短连接超时，避免启动阻塞
        },
        'CONN_MAX_AGE': 0,  # 不保留持久连接，每次请求重新尝试连接
    }
}

2. 优化数据库路由，动态判断数据库可用性 #

编写工具函数检查数据库是否可用，再更新路由逻辑，确保只有当ovi_db可用时才路由到它：

from django.db import connections
from psycopg2 import OperationalError

def is_db_available(db_name):
    """检查指定数据库是否可连接"""
    try:
        with connections[db_name].cursor():
            return True
    except (OperationalError, AttributeError):
        # AttributeError捕获连接为None的情况
        return False

更新路由类：

from django.db.models import Model
from typing import Any, Optional

class OVIRouter:
    db_name = 'ovi_db'

    def db_for_read(self, model: Model, **hints: Any) -> Optional[str]:
        if model._meta.model_name == "ovidevice":
            return self.db_name if is_db_available(self.db_name) else 'default'
        return None

    def db_for_write(self, model: Model, **hints: Any) -> Optional[str]:
        if model._meta.model_name == "ovidevice":
            return self.db_name if is_db_available(self.db_name) else 'default'
        return None

    def allow_relation(self, obj1: Model, obj2: Model, **hints: Any) -> Optional[bool]:
        if obj1._meta.model_name == "ovidevice" or obj2._meta.model_name == "ovidevice":
            return True
        return None

    def allow_migrate(
        self,
        db: str,
        app_label: str,
        model_name: Optional[str] = None,
        **hints: Any
    ) -> Optional[bool]:
        if model_name == "ovidevice":
            return db == self.db_name if is_db_available(self.db_name) else db == 'default'
        return None

3. 额外韧性优化 #

• 动态重试：可在is_db_available中添加重试逻辑，比如尝试2次后再判定不可用
• 日志监控：通过自定义后端的日志输出，实时掌握ovi_db的连接状态
• 避免迁移冲突：allow_migrate逻辑确保OVIDevice模型只会在可用的数据库上执行迁移

这样配置后，应用启动时不会强制连接ovi_db，只有当访问OVIDevice模型时才会尝试连接；连接失败时自动降级到默认库，后续请求会自动重试连接，一旦VPN开启将自动切换回ovi_db。

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