python 获取mysql数据库列表以及用户权限

发布时间：2020-02-23 12:30:57编辑：admin阅读（2545）

一、需求分析

需要统计出当前数据库的所有数据库名，以及每个用户的授权信息。

 

获取所有数据库

在mysql里面，使用命令：

show databases

 

就可以获取所有数据库了

 

获取所有用户

执行命令：

select User from mysql.user

 

注意：需要排除到默认的用户，比如：

"root", "mysql.sys", "mysql.session"

 

获取用户权限

语法：

show grants for 用户名;

 

比如：

show grants for test;

执行输出：

GRANT USAGE ON *.* TO 'test'@'%'
GRANT SELECT ON `DB01`.* TO 'test'@'%'
GRANT ALL PRIVILEGES ON `DB02`.* TO 'test'@'%' WITH GRANT OPTION




注意：这段信息表示，test用户，对DB01数据库下的所有表，是只读权限。

对对DB02数据库下的所有表，拥有读写权限。

 

二、python实现

由于时间关系，这里不一一解释了，代码里面都有注释。

完整代码如下：

#!/usr/bin/env python
# coding: utf-8

import json
import pymysql


class Mysql(object):
    # mysql 端口号,注意：必须是int类型
    def __init__(self, host, user, passwd, port, db_name):
        self.host = host
        self.user = user
        self.passwd = passwd
        self.port = port
        self.db_name = db_name

    def select(self, sql):
        """
        执行sql命令
        :param sql: sql语句
        :return: 元祖
        """
        try:
            conn = pymysql.connect(
                host=self.host,
                user=self.user,
                passwd=self.passwd,
                port=self.port,
                database=self.db_name,
                charset='utf8',
                cursorclass=pymysql.cursors.DictCursor
            )
            cur = conn.cursor()  # 创建游标
            # conn.cursor()
            cur.execute(sql)  # 执行sql命令
            res = cur.fetchall()  # 获取执行的返回结果
            cur.close()
            conn.close()
            return res
        except Exception as e:
            print(e)
            return False

    def get_all_db(self):
        """
        获取所有数据库名
        :return: list
        """
        # 排除自带的数据库
        exclude_list = ["sys", "information_schema", "mysql", "performance_schema"]
        sql = "show databases"  # 显示所有数据库
        res = self.select(sql)
        # print(res)
        if not res:  # 判断结果非空
            return False

        db_list = []  # 数据库列表
        for i in res:
            db_name = i['Database']
            # 判断不在排除列表时
            if db_name not in exclude_list:
                db_list.append(db_name)
                # print(db_name)

        if not db_list:
            return False

        return db_list

    def get_user_list(self):
        """
        获取用户列表
        :return: list
        """
        # 排除自带的用户
        exclude_list = ["root", "mysql.sys", "mysql.session"]
        sql = "select User from mysql.user"
        res = self.select(sql)
        # print(res)
        if not res:  # 判断结果非空
            return False

        user_list = []
        for i in res:
            db_name = i['User']
            # 判断不在排除列表时
            if db_name not in exclude_list:
                user_list.append(db_name)

        if not user_list:
            return False

        return user_list

    def get_user_power(self):
        """
        获取用户权限
        :return: {}

        {
            "test":{  # 用户名
                "read":["db1","db2"],  # 只拥有读取权限的数据库
                "all":["db1","db2"],  # 拥有读写权限的数据库
            },
            ...
        }
        """
        info_dict = {}  # 最终结果字典
        # 获取用户列表
        user_list = self.get_user_list()
        if not user_list:
            return False

        # 查询每一个用户的权限
        for user in user_list:
            # print("user",user)
            sql = "show grants for {}".format(user)
            res = self.select(sql)
            if not res:
                return False

            for i in res:
                key = 'Grants for {}@%'.format(user)
                # print("key",key)
                # 判断key值存在时
                if i.get(key):
                    # print(i[key])
                    # 包含ALL或者SELECT时
                    if "ALL" in i[key] or "SELECT" in i[key]:
                        # print(i[key])
                        if not info_dict.get(user):
                            info_dict[user] = {"read": [], "all": []}

                        cut_str = i[key].split()  # 空格切割
                        # print(cut_str,len(cut_str))
                        power = cut_str[1]  # 权限，比如ALL，SELECT

                        if len(cut_str) == 6:  # 判断切割长度
                            # 去除左边的`
                            tmp_str = cut_str[3].lstrip("`")
                        else:
                            tmp_str = cut_str[4].lstrip("`")

                        # 替换字符串
                        tmp_str = tmp_str.replace('`.*', '')
                        value = tmp_str.replace('\_', '-')

                        # 判断权限为select 时
                        if power.lower() == "select":
                            if value not in info_dict[user].get("read"):
                                # 只读列表
                                info_dict[user]["read"].append(value)
                        else:
                            if value not in info_dict[user].get("all"):
                                # 所有权限列表
                                info_dict[user]["all"].append(value)

        # print(info_dict)
        return info_dict


if __name__ == '__main__':
    host = "192.168.10.10"
    user = "root"
    passwd = "123456"
    port = 3306
    db_name = "mysql"

    obj = Mysql(host, user, passwd, port, db_name)
    all_db_list = obj.get_all_db()
    user_power = obj.get_user_power()

    print("all_db_list",all_db_list)
    print("user_power",user_power)

关键字：