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

Question

我正在使用 django ORM 和 postgres 数据库。一小组用户使用导入和导出脚本与其进行交互。该数据库仅在我们的内网上可用。如果有人在 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 应用程序，因此是一个中间件解决方案 如何我要在 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.create_connection(("server", port)) 并查看它引发的异常...

Answer 2

，

使用多数据库路由 技巧是检查数据库在 django 初始化时是否可访问，然后将所有 ORM 查询路由到后备数据库。

检查数据库是否存在的一种方法是在 try .. except 块中运行 ORM 查询，并设置一个可在路由器中访问的变量.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

Answer 3

from django.db import connections
db_conn = connections['default']
try:
    c = db_conn.cursor()
except Exception as e:
    connected = False
else:
    connected = True

这可用于检查与数据库的连接。