zoukankan      html  css  js  c++  java
  • mysql查看查询语句性能explain

    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
  • 相关阅读:
    在内部局域网内搭建HTTPs
    如何创建自己的Nuget包
    Scratch入门
    灰度发布 & 蓝绿部署
    .net异步方法初探
    [转]批处理静默自动安装证书
    住院病案首页数据填写质量规范(暂行)
    DRGs
    【规范】电子病历系统功能规范(试行)
    Red Hat Enterprise Linux 8配置YUM源的两种方式
  • 原文地址:https://www.cnblogs.com/erchun/p/15329113.html
Copyright © 2011-2022 走看看