https://bbs.huaweicloud.com/blogs/177202

https://www.cnblogs.com/yycc/p/7338894.html

先用explain查看性能，如果性能达标可以执行，如果不达标需要添加索引查询

from sqlalchemy import create_engine

class ConnectMysql:
    """操作Mysql"""

    instance = None

    def __new__(cls, *args, **kwargs):
        if not cls.instance:
            cls.instance = object.__new__(cls)
            return cls.instance
        else:
            return cls.instance

    def __init__(self, db_url, db_name):
        self.engine = create_engine(db_url + db_name)
        self.conn = self.engine.connect()

    def execute(self, sql):
        """执行sql语句，仅限创建/删除索引"""
        try:
            assert ('alter' in sql or 'ALTER' in sql or 'drop' in sql or 'DROP' in sql), '非创建/删除索引语句，查询请用fetch_one'
            self.conn.execute(sql)
        except Exception as e:
            log.error("非创建/删除索引语句,sql:{}".format(sql))
            log.exception(e)

    def explain_sql(self, sql):
        """查看sql性能"""
        type_list = ['system', 'const', 'eq_ref', 'ref', 'fulltext', 'ref_or_null', 'index_merge', 'unique_subquery',
                     'index_subquery', 'range']
        try:
            execute = self.conn.execute("explain " + sql)
            value = execute.fetchone()
            if value[3] in type_list:
                return True
            return False
        except Exception as e:
            log.error("未查询到数据库,sql:{}".format(sql))
            log.exception(e)
            return False

    def fetch_one(self, sql):
        """查询sql语句返回的第一条数据"""
        try:
            assert self.explain_sql(sql), 'sql语句性能未达标'
            execute = self.conn.execute(sql)
            value = execute.fetchone()
            return value[0]
        except Exception as e:
            log.error("未查询到数据库,sql:{}".format(sql))
            log.exception(e)

    def fetch_one_with_index(self, add_index_sql, sql, drop_index_sql):
        """创建索引，查询sql语句返回的第一条数据"""
        self.execute(add_index_sql)
        value = self.fetch_one(sql)
        self.execute(drop_index_sql)
        return value