如何在 Django 中检查与 mysql 的数据库连接

Question

我该怎么做？

我想，我可以从数据库中读取一些东西，但是看起来太多了，有没有类似的东西？：

settings.DATABASES['default'].check_connection()

Answer 1

您需要做的就是启动一个应用程序，如果它没有连接，它将失败。您可以尝试的其他方法是 shell 尝试以下 -

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

Answer 2

我使用以下名为 wait_for_db 的 Django 管理命令：

import time

from django.db import connection
from django.db.utils import OperationalError
from django.core.management.base import BaseCommand


class Command(BaseCommand):
    """Django command that waits for database to be available"""

    def handle(self, *args, **options):
        """Handle the command"""
        self.stdout.write('Waiting for database...')
        db_conn = None
        while not db_conn:
            try:
                connection.ensure_connection()
                db_conn = True
            except OperationalError:
                self.stdout.write('Database unavailable, waiting 1 second...')
                time.sleep(1)

        self.stdout.write(self.style.SUCCESS('Database available!'))

Answer 3

我有一个更复杂的案例，我在 djongo 模块和 RDS mysql 后面使用 mongodb。所以不仅是多数据库，djongo反而抛出SQLDecode错误。我还必须执行并获取才能使其正常工作：

from django.conf import settings

if settings.DEBUG:
    # Quick database check here
    from django.db import connections
    from django.db.utils import OperationalError
    dbs = settings.DATABASES.keys()
    for db in dbs:
        db_conn = connections[db]  # i.e. default
        try:
            c = db_conn.cursor()
            c.execute("""SELECT "non_existent_table"."id" FROM "non_existent_table" LIMIT 1""")
            c.fetchone()
            print("Database '{}' connection ok.".format(db))  # This case is for djongo decoding sql ok
        except OperationalError as e:
            if 'no such table' in str(e):
                print("Database '{}' connection ok.".format(db))  # This is ok, db is present
            else:
                raise  # Another type of op error
        except Exception:  # djongo sql decode error
            print("ERROR: Database {} looks to be down.".format(db))
            raise

我将其加载到我的应用程序 __init__.py 中，因为我希望它运行在启动时只加载一次，并且仅在启用 DEBUG 时加载。希望对您有所帮助！

Answer 4

假设你因为 docker 需要这个，但不限于 docker，记住这是在一天结束时 Bash，因此无处不在 *NIX.

您首先需要使用 django-environ，因为它会让这一切变得更容易。

DATABASE_URL 环境变量将在您的 Django 应用程序内部和此处使用。您的设置如下所示：

import environ

env = environ.Env()

...

DATABASES = {
    'default': env.db('DATABASE_URL'),
    'other': env.db('DATABASE_OTHER_URL')  # for illustration purposes
}

...

您的环境变量应如下所示：（更多信息 here）

# This works with ALL the databases django supports ie (mysql/mssql/sqlite/...)
DATABASE_URL=postgres://user:pass@name_of_box:5432/database_name
DATABASE_OTHER_URL=oracle://user:pass@/(description=(address=(host=name_of_box)(protocol=tcp)(port=1521))(connect_data=(SERVICE_NAME=EX)))

在你的 entrypoint.sh 中做这样的事情：

function database_ready() {
  # You need to pass a single argument called "evironment_dsn"
  python << EOF
import sys
import environ
from django.db.utils import ConnectionHandler, OperationalError

env = environ.Env()
try:
   ConnectionHandler(databases={'default': env.db('')})['default'].ensure_connection()
except (OperationalError, DatabaseError):
   sys.exit(-1)
sys.exit(0)
EOF
}

然后，假设您想等待您的主数据库 [在本例中为 postgres]，您将其添加到相同的 entrypoint.sh 中，在 database_ready 函数下。

until database_ready DATABASE_URL; do
  >&2 echo "Main DB is unavailable - sleeping"
  sleep 1
done

这只会继续，如果 postgres 已经启动并且运行。甲骨文呢？同样的，在上面的代码下，我们添加：

until database_ready DATABASE_OTHER_URL; do
  >&2 echo "Secondary DB is unavailable - sleeping"
  sleep 1
done

这样做会给你带来一些好处：

您无需担心其他依赖项，例如二进制文件等。
您可以切换数据库而不必担心这种破坏。（代码与数据库 100% 无关）

Answer 5

Javier 的回答似乎不再有效。他是我放在一起执行检查 Docker 入口点中数据库可用性任务的人，假设您有可用的 psycopg2 库（例如，您是运行 Django 应用程序）：

function database_ready() {
    python << EOF
import psycopg2
try:
    db = psycopg2.connect(host="", port="", dbname="", user="", password="")
except:
    exit(1)

exit(0)
EOF
}

until database_ready $DATABASE_HOST $DATABASE_PORT $DATABASE_NAME $DATABASE_USER $DATABASE_PASSWORD; do
  >&2 echo "Database is unavailable at $DATABASE_HOST:$DATABASE_PORT/$DATABASE_NAME - sleeping..."
  sleep 1
done
echo "Database is ready - $DATABASE_HOST:$DATABASE_PORT/$DATABASE_NAME"```

Answer 6

运行 shell

python manage.py shell

执行这个脚本

import django
print(django.db.connection.ensure_connection())

如果它打印 None 意味着一切正常，否则如果你的数据库连接出现问题它会抛出错误

Answer 7

创建文件your_app_name/management/commands/waitdb.py并粘贴下面的内容。

import time
from django.core.management.base import BaseCommand
from django.db import connection
from django.db.utils import OperationalError
from django.utils.translation import ngettext


class Command(BaseCommand):
    help = 'Checks database connection'

    def add_arguments(self, parser):
        parser.add_argument(
            '--seconds',
            nargs='?',
            type=int,
            help='Number of seconds to wait before retrying',
            default=1,
        )
        parser.add_argument(
            '--retries',
            nargs='?',
            type=int,
            help='Number of retries before exiting',
            default=3,
        )

    def handle(self, *args, **options):
        wait, retries = options['seconds'], options['retries']
        current_retries = 0
        while current_retries < retries:
            current_retries += 1
            try:
                connection.ensure_connection()
                break
            except OperationalError:
                plural_time = ngettext('second', 'seconds', wait)
                self.stdout.write(
                    self.style.WARNING(
                        f'Database unavailable, retrying after {wait} {plural_time}!'
                    )
                )
                time.sleep(wait)

python manage.py waitdb --seconds 5 --retries 2
python manage.py waitdb # defaults to 1 seconds & 3 retries

Answer 8

这是一个老问题，但需要更新答案

python manage.py check --database default

如果您没有使用默认值，或者如果您想测试设置中列出的其他数据库，只需命名即可。

从 3.1 版开始可用 +

勾选documentation

如何在 Django 中检查与 mysql 的数据库连接

how can I check database connection to mysql in django

python

django

django-settings