63 lines
2.5 KiB
Python
63 lines
2.5 KiB
Python
import re
|
|
|
|
from django.core.management.base import BaseCommand, CommandError
|
|
from django.db import connection
|
|
|
|
MINIMUM_VERSION = (8, 0, 35)
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "检查生产 MySQL 的版本、字符集、引擎、严格模式和事务隔离级别"
|
|
|
|
def handle(self, *args, **options):
|
|
if connection.vendor != "mysql":
|
|
raise CommandError(f"数据库必须是 MySQL，当前为 {connection.vendor}")
|
|
|
|
with connection.cursor() as cursor:
|
|
cursor.execute(
|
|
"""
|
|
SELECT VERSION(),
|
|
@@character_set_server,
|
|
@@collation_server,
|
|
@@sql_mode,
|
|
@@transaction_isolation,
|
|
@@default_storage_engine
|
|
"""
|
|
)
|
|
(
|
|
version_text,
|
|
charset,
|
|
collation,
|
|
sql_mode,
|
|
isolation,
|
|
storage_engine,
|
|
) = cursor.fetchone()
|
|
|
|
match = re.match(r"^(\d+)\.(\d+)\.(\d+)", version_text)
|
|
if match is None:
|
|
raise CommandError(f"无法解析 MySQL 版本: {version_text}")
|
|
version = tuple(int(part) for part in match.groups())
|
|
if version < MINIMUM_VERSION:
|
|
minimum = ".".join(str(part) for part in MINIMUM_VERSION)
|
|
raise CommandError(f"MySQL 版本过低: {version_text}，最低要求 {minimum}")
|
|
if charset.lower() != "utf8mb4":
|
|
raise CommandError(f"character_set_server 必须是 utf8mb4，当前为 {charset}")
|
|
if not collation.lower().startswith("utf8mb4_"):
|
|
raise CommandError(f"collation_server 必须属于 utf8mb4，当前为 {collation}")
|
|
|
|
modes = {mode.strip().upper() for mode in sql_mode.split(",")}
|
|
if not {"STRICT_TRANS_TABLES", "STRICT_ALL_TABLES"} & modes:
|
|
raise CommandError(f"MySQL 未启用严格模式: {sql_mode}")
|
|
if isolation.upper().replace("_", "-") != "READ-COMMITTED":
|
|
raise CommandError(f"事务隔离级别必须是 READ-COMMITTED，当前为 {isolation}")
|
|
if storage_engine.lower() != "innodb":
|
|
raise CommandError(f"默认存储引擎必须是 InnoDB，当前为 {storage_engine}")
|
|
|
|
self.stdout.write(
|
|
self.style.SUCCESS(
|
|
"MySQL 检查通过: "
|
|
f"version={version_text}, charset={charset}, "
|
|
f"collation={collation}, isolation={isolation}, engine={storage_engine}"
|
|
)
|
|
)
|