how can I check database connection to mysql in django

Question

how can I do it?

I thought, I can read something from database, but it looks too much, is there something like?:

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

Answer 1

All you need to do is start a application and if its not connected it will fail. Other way you can try is on shell try following -

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

Run the shell

python manage.py shell

Execute this script

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

If it print None means everything is okay, otherwise it will throw an error if something wrong happens on your db connection

Answer 3

It's an old question but it needs an updated answer

python manage.py check --database default

If you're not using default or if you want to test other databases listed in your settings just name it.

It is available since version 3.1 +

Check the documentation

Answer 4

I use the following Django management command called wait_for_db:

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 5

Assuming you needed this because of docker, BUT is not limitted to docker, remember this is at the end of the day Bash, and thus works everywhere *NIX.

You will first need to be using django-environ, since it will make this a whole lot easier.

The DATABASE_URL environment variable will be used inside your Django app, and here. Your settings would look like this:

import environ

env = environ.Env()

...

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

...

Your environment variables should look something like this: (more info 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)))

Inside your entrypoint.sh do something like this:

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('$1')})['default'].ensure_connection()
except (OperationalError, DatabaseError):
   sys.exit(-1)
sys.exit(0)
EOF
}

Then, lets say you want to wait for your main db [the postgres in this case], you add this inside the same entrypoint.sh, under the database_ready function.

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

This will only continue, IF postgres is up and running. What about oracle? Same thing, under the code above, we add:

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

Doing it this way will give you a couple of advantages:

1. you don't need to worry about other dependencies such as binaries and the likes.

2. you can switch databases and not have to worry about this breaking. (code is 100% database agnostic)

Answer 6

Create a file your_app_name/management/commands/waitdb.py and paste the bellow.

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 7

I had a more complicated case where I am using mongodb behind djongo module, and RDS mysql. So not only is it multiple databases, but djongo throws an SQLDecode error instead. I also had to execute and fetch to get this working:

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

I load this in my app __init__.py, as I want it to run on startup only once and only if DEBUG is enabled. Hope it helps!

Answer 8

It seems Javier's answer is no longer working. He's one I put together to perform the task of checking database availability in a Docker entrypoint, assuming you have the psycopg2 library available (you're running a Django application, for instance):

function database_ready() {
    python << EOF
import psycopg2
try:
    db = psycopg2.connect(host="$1", port="$2", dbname="$3", user="$4", password="$5")
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"```