从 django 测试 postgres 数据库的可访问性

Question

我正在使用带有 postgres 数据库的 django ORM。一小部分用户使用导入和导出脚本与之交互。该数据库仅在我们的 Intranet 上可用。如果有人在 postgres 不可用时尝试使用数据库，则脚本会挂起。我想让脚本在尝试处理任何数据之前测试数据库是否可用。

我可以使用 shell 连接到数据库，导入模型并尝试进行查询：

from myapp.models import mymodel
mymodel.objects.count()

这会导致很长的延迟，但随后 django 会引发 OperationalError 并带有信息性消息（“无法连接到服务器：网络无法访问...”）。

我想通过对数据库进行最小查询来测试数据库访问，例如：

from django.db import connection
cursor = connection.cursor()
cursor.execute("select 1")

但这永远不会超出cursor = connection.cursor() 行。没有错误信息。

• 为什么其中一个查询会引发错误，而另一个不会？
• 从脚本测试数据库是否可用的最佳方法是什么？
• 如果连接在合理（可能是用户指定的）时间内没有成功，我如何确保会引发错误？

这不是一个 Web 应用程序，因此不可能使用 How do I test a database connection in Django? 的中间件解决方案。

编辑

按照@benjaoming 的建议，我制作了一个函数来测试连接：

import socket

def test_connection():
    """Test whether the postgres database is available. Usage:

        if "--offline" in sys.argv:
            os.environ['DJANGO_SETTINGS_MODULE'] = 'myapp.settings.offline'
        else:
            os.environ['DJANGO_SETTINGS_MODULE'] = 'myapp.settings.standard'
            from myapp.functions.connection import test_connection
            test_connection()
    """
    try:
        s = socket.create_connection(("example.net", 5432), 5)
        s.close()
    except socket.timeout:
        msg = """Can't detect the postgres server. If you're outside the
        intranet, you might need to turn the VPN on."""
        raise socket.timeout(msg)

这似乎可以解决问题。

Answer 1

您可以使用另一种方法来确定 postgres 服务器是否可访问：例如 socket 模块 -- 只需执行 socket.create_connection(("server", port)) 并查看它引发的异常...

Answer 2

使用multiple database routing，

诀窍是在 django 初始化时检查数据库是否可访问，然后 route 所有 ORM 查询到您的备用数据库。

我检查数据库是否存在的一种方法是在 try..except 块内运行 ORM 查询并设置一个可在 routers.py 中访问的变量

from django.db import connections
conn = connections['default']
try:
    c = conn.cursor() #this will take some time if error
except OperationalError:
    reachable = False
else:
    reachable = True

您可以将此代码放入urls.py 或routers.py 本身。

自定义路由器将检查变量是否已设置并路由到您的备用数据库

class AppRouter(object):
    def db_for_read(self, model, **hints):
       if reachable :
           return 'actual_db'
       else:
           return 'fallback_db'

    # define db_for_write likewise