python3连接MySQL的工具类

Love The Way You Lie 2022-11-05 08:40 154阅读 0赞

**目录**

一、环境准备

1、python3安装

2、MySQL安装

二、安装驱动

1、pymogo驱动安装

2、pymogo驱动测试

三、py工具类

1、工具脚本

--------------------

## 一、环境准备 ##

### 1、python3安装 ###

windows离线安装python3.6.8环境：[https://blog.csdn.net/qq262593421/article/details/111309116][https_blog.csdn.net_qq262593421_article_details_111309116]

### 2、MySQL安装 ###

Linux 离线安装 python 3.6.10：[https://blog.csdn.net/qq262593421/article/details/105395101][https_blog.csdn.net_qq262593421_article_details_105395101]

win10安装MySQL5.7.22 32位：[https://blog.csdn.net/qq262593421/article/details/108732403][https_blog.csdn.net_qq262593421_article_details_108732403]

## 二、安装驱动 ##

### 1、pymogo驱动安装 ###

pip install pymysql

### 2、pymogo驱动测试 ###

python
    import pymysql

## 三、py工具类 ##

### 1、工具脚本 ###

# -*- encoding: utf-8 -*-
    
    import pymysql
    
    class MySQLUtil:
        """
        MySQL工具类
        """
        def __init__(self, host="127.0.0.1", user=None, passwd=None, db=None, charset="utf8", *args, **kwargs):
            """构造函数"""
            self.__host = host
            self.__user = user
            self.__passwd = passwd
            self.__db = db
            self.__conn = pymysql.connect(host=host, user=user, passwd=passwd, db=db, charset=charset, *args, **kwargs)
            self.__cursor = self.__conn.cursor()
    
        def __del__(self):
            """析构函数"""
            self.__cursor.close()
            self.__conn.close()
    
        def get_conn(self):
            """获取连接"""
            return self.__conn
    
        def get_cursor(self, cursor=None):
            """获取游标"""
            return self.__conn.cursor(cursor)
    
        def select_db(self, db):
            """选择数据库"""
            self.__conn.select_db(db)
    
        def list_databases(self, args=None):
            """查询所有数据库"""
            self.__cursor.execute("SHOW DATABASES", args)
            dbs = []
            for db in self.__cursor.fetchall():
                dbs.append(db[0])
            return dbs
    
        def list_tables(self, args=None):
            """查询所有表"""
            self.__cursor.execute("SHOW TABLES", args)
            tables = []
            for table in self.__cursor.fetchall():
                tables.append(table[0])
            return tables
    
        def execute(self, sql, args=None):
            """获取SQL执行结果"""
            self.__cursor.execute(sql, args)
            return self.__cursor.fetchall()
    
        def get_version(self, args=None):
            """获取MySQL版本"""
            self.__cursor.execute("SELECT VERSION()", args)
            version = self.__cursor.fetchone()
            print("MySQL Version : %s" % version)
            return version
    
        def list_table_metadata(self, args=None):
            """查询所有表的元数据信息"""
            sql = "SELECT * FROM information_schema.TABLES WHERE TABLE_TYPE !='SYSTEM VIEW' AND TABLE_SCHEMA NOT IN ('sys','mysql','information_schema','performance_schema')"
            self.__cursor.execute(sql, args)
            return self.__cursor.fetchall()
    
        def get_table_fields(self, db, table, args=None):
            """获取表字段信息"""
            sql = 'SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE table_schema="'+db+'" AND table_name="'+table+'"'
            self.__cursor.execute(sql, args)
            fields = []
            for field in self.__cursor.fetchall():
                fields.append(field[0])
            return fields
    
        def table_metadata(self, db, table, args=None):
            """查询表字段的元数据信息"""
            db = "'" + db + "'"
            table = "'" + table + "'"
            sql = """
            SELECT 
                column_name,column_type,ordinal_position,column_comment,column_default 
            FROM 
                information_schema.COLUMNS 
            WHERE 
                table_schema = %s AND table_name = %s;
            """ % (db, table)
            self.__cursor.execute(sql, args)
            return self.__cursor.fetchall()
    
    
    if __name__ == "__main__":
        mysqlUtil = MySQLUtil("127.0.0.1", "root", "123456", "test")
        mysqlUtil = MySQLUtil(host="127.0.0.1", user="root", passwd="123456", db="test")
        mysqlUtil.get_version()
        dbs = mysqlUtil.list_databases()
        print(dbs)
        conn = mysqlUtil.get_conn()
        mysqlUtil.select_db("test")
        print(type(conn.db), conn.db)
        databases = mysqlUtil.list_databases()
        print(type(databases), databases)
        tables = mysqlUtil.list_tables()
        print(type(tables), tables)
        sql = "SELECT * FROM t_user"
        result = mysqlUtil.execute(sql)
        for i in result:
            print(i)
        result = mysqlUtil.table_metadata("test", "t_user")
        result = mysqlUtil.get_table_fields("test", "t_user")
        for i in result:
            print(i)

**执行结果**

> MySQL Version : 5.7.22  
> <class 'bytes'> b'test'  
> (1, '科比', 10)  
> (2, '麦迪', 20)

[https_blog.csdn.net_qq262593421_article_details_111309116]: https://blog.csdn.net/qq262593421/article/details/111309116
[https_blog.csdn.net_qq262593421_article_details_105395101]: https://blog.csdn.net/qq262593421/article/details/105395101
[https_blog.csdn.net_qq262593421_article_details_108732403]: https://blog.csdn.net/qq262593421/article/details/108732403