Combined MySQL backend server info queries.

This commit is contained in:
Adam Johnson 2020-06-20 18:17:22 +01:00 committed by Mariusz Felisiak
parent e62d55a4fe
commit e37f809618
3 changed files with 42 additions and 33 deletions

View File

@ -355,10 +355,32 @@ class DatabaseWrapper(BaseDatabaseWrapper):
return {}
@cached_property
def mysql_server_info(self):
def mysql_server_data(self):
with self.temporary_connection() as cursor:
cursor.execute('SELECT VERSION()')
return cursor.fetchone()[0]
# Select some server variables and test if the time zone
# definitions are installed. CONVERT_TZ returns NULL if 'UTC'
# timezone isn't loaded into the mysql.time_zone table.
cursor.execute("""
SELECT VERSION(),
@@sql_mode,
@@default_storage_engine,
@@sql_auto_is_null,
@@lower_case_table_names,
CONVERT_TZ('2001-01-01 01:00:00', 'UTC', 'UTC') IS NOT NULL
""")
row = cursor.fetchone()
return {
'version': row[0],
'sql_mode': row[1],
'default_storage_engine': row[2],
'sql_auto_is_null': bool(row[3]),
'lower_case_table_names': bool(row[4]),
'has_zoneinfo_database': bool(row[5]),
}
@cached_property
def mysql_server_info(self):
return self.mysql_server_data['version']
@cached_property
def mysql_version(self):
@ -373,7 +395,5 @@ class DatabaseWrapper(BaseDatabaseWrapper):
@cached_property
def sql_mode(self):
with self.cursor() as cursor:
cursor.execute('SELECT @@sql_mode')
sql_mode = cursor.fetchone()
return set(sql_mode[0].split(',') if sql_mode else ())
sql_mode = self.mysql_server_data['sql_mode']
return set(sql_mode.split(',') if sql_mode else ())

View File

@ -50,10 +50,7 @@ class DatabaseFeatures(BaseDatabaseFeatures):
@cached_property
def _mysql_storage_engine(self):
"Internal method used in Django tests. Don't rely on this from your code"
with self.connection.cursor() as cursor:
cursor.execute("SELECT ENGINE FROM INFORMATION_SCHEMA.ENGINES WHERE SUPPORT = 'DEFAULT'")
result = cursor.fetchone()
return result[0]
return self.connection.mysql_server_data['default_storage_engine']
@cached_property
def update_can_self_select(self):
@ -82,18 +79,11 @@ class DatabaseFeatures(BaseDatabaseFeatures):
@cached_property
def has_zoneinfo_database(self):
# Test if the time zone definitions are installed. CONVERT_TZ returns
# NULL if 'UTC' timezone isn't loaded into the mysql.time_zone.
with self.connection.cursor() as cursor:
cursor.execute("SELECT CONVERT_TZ('2001-01-01 01:00:00', 'UTC', 'UTC')")
return cursor.fetchone()[0] is not None
return self.connection.mysql_server_data['has_zoneinfo_database']
@cached_property
def is_sql_auto_is_null_enabled(self):
with self.connection.cursor() as cursor:
cursor.execute('SELECT @@SQL_AUTO_IS_NULL')
result = cursor.fetchone()
return result and result[0] == 1
return self.connection.mysql_server_data['sql_auto_is_null']
@cached_property
def supports_over_clause(self):
@ -150,10 +140,7 @@ class DatabaseFeatures(BaseDatabaseFeatures):
@cached_property
def ignores_table_name_case(self):
with self.connection.cursor() as cursor:
cursor.execute('SELECT @@LOWER_CASE_TABLE_NAMES')
result = cursor.fetchone()
return result and result[0] != 0
return self.connection.mysql_server_data['lower_case_table_names']
@cached_property
def supports_default_in_lead_lag(self):

View File

@ -29,20 +29,22 @@ class DatabaseCheckTests(TestCase):
'STRICT_TRANS_TABLES',
'STRICT_ALL_TABLES',
]
for response in good_sql_modes:
with mock.patch(
'django.db.backends.utils.CursorWrapper.fetchone', create=True,
return_value=(response,)
for sql_mode in good_sql_modes:
with mock.patch.object(
connection, 'mysql_server_data', {'sql_mode': sql_mode},
):
self.assertEqual(check_database_backends(databases=self.databases), [])
_clean_sql_mode()
bad_sql_modes = ['', 'WHATEVER']
for response in bad_sql_modes:
with mock.patch(
'django.db.backends.utils.CursorWrapper.fetchone', create=True,
return_value=(response,)
):
for sql_mode in bad_sql_modes:
mocker_default = mock.patch.object(
connection, 'mysql_server_data', {'sql_mode': sql_mode},
)
mocker_other = mock.patch.object(
connections['other'], 'mysql_server_data', {'sql_mode': sql_mode},
)
with mocker_default, mocker_other:
# One warning for each database alias
result = check_database_backends(databases=self.databases)
self.assertEqual(len(result), 2)