def check_database_connected(app_configs, **kwargs):
"""
A Django check to see if connecting to the configured default
database backend succeeds.
"""
errors = []
try:
connection.ensure_connection()
except OperationalError as e:
msg = 'Could not connect to database: {!s}'.format(e)
errors.append(checks.Error(msg, id=ERROR_CANNOT_CONNECT_DATABASE))
except ImproperlyConfigured as e:
msg = 'Datbase misconfigured: "{!s}"'.format(e)
errors.append(checks.Error(msg, id=ERROR_MISCONFIGURED_DATABASE))
else:
if not connection.is_usable():
errors.append(checks.Error('Database connection is not usable',
id=ERROR_UNUSABLE_DATABASE))
return errors
python类is_usable()的实例源码
def _should_check_constraints(self, connection):
return (
connection.features.can_defer_constraint_checks and
not connection.needs_rollback and connection.is_usable()
)
def make_sure_mysql_usable():
# mysql is lazily connected to in django.
# connection.connection is None means
# you have not connected to mysql before
if connection.connection and not connection.is_usable():
# destroy the default mysql connection
# after this line, when you use ORM methods
# django will reconnect to the default mysql
del connections._connections.default
#
# def get_item(data_dict,item):
# try:
# item_value = data_dict[item]
# return item_value
# except:
# return '-1'
#
# def get_config(group,config_name):
# config = ConfigParser.ConfigParser()
# config.readfp(open('./myapp/etc/config.ini','r'))
# config_value=config.get(group,config_name).strip(' ').strip('\'').strip('\"')
# return config_value
#
# def filters(data):
# return data.strip(' ').strip('\n').strip('\br')
#
# select_limit = int(get_config('settings','select_limit'))
# export_limit = int(get_config('settings','export_limit'))
# host = get_config('settings','host')
# port = get_config('settings','port')
# user = get_config('settings','user')
# passwd = get_config('settings','passwd')
# dbname = get_config('settings','dbname')
# wrong_msg = get_config('settings','wrong_msg')
# incp_host = get_config('settings','incp_host')
# incp_port = int(get_config('settings','incp_port'))
# incp_user = get_config('settings','incp_user')
# incp_passwd = get_config('settings','incp_passwd')
# public_user = get_config('settings','public_user')
def _should_check_constraints(self, connection):
return (
connection.features.can_defer_constraint_checks and
not connection.needs_rollback and connection.is_usable()
)
def check_db_connection():
from django.db import connection
if connection.connection:
#NOTE: (zacky, 2016.MAR.21st) IF CONNECTION IS CLOSED BY BACKEND, CLOSE IT AT DJANGO, WHICH WILL BE SETUP AFTERWARDS.
if not connection.is_usable():
connection.close()