从Django测试PostgreSQL数据库的可访问性

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应用程序，因此像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)

这似乎可以解决问题。

- Michael Dunn

---

你尝试过找出为什么数据库连接会断开吗？ - Ignacio Vazquez-Abrams

它并没有被删除：数据库只在我们的内部网络上可用。如果您想在外部使用它，您必须首先通过虚拟私人网络连接。我希望警告用户注意这一点，而不是让他们遇到冻结的终端。 - Michael Dunn

Answer 1

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

- benjaoming

---

我已经编写了一个函数来完成这个任务，在将数据库设置为在线/离线模式后调用该函数--请参见问题的编辑。 - Michael Dunn

看起来不错！你可能想为 socket.error 异常添加一个 except 块。 - benjaoming

缺少VPN会导致socket.error吗？因为如果这不是用户需要处理的问题，那么我希望正常的异常可以被触发。 - Michael Dunn

是的，我认为如果网络接口关闭或远程端不接受连接（但存在），则会引发socket.error。 - benjaoming

Answer 2

使用多数据库路由，

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

我检查数据库是否存在的一种方法是在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

- srj

---

from django.db.utils import OperationalError - Akshay

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

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

- Shah Vipul