mysql檢視查詢語句效能explain
阿新 • • 發佈:2021-09-24
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