[Django]-How can I check database connection to mysql in django

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

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

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!'))

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)

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

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!

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"```