Python MySQL 数据库连接 (三) mysqlclie

Mysql驱动介绍

MySQL-python(弃用)：也就是MySQLdb，是对C语言操作MySQL数据库的一个简单封装。遵循了Python DB API v2。但是只支持Python2，目前还不支持Python3。
mysqlclient(安装难,但django官方默认)：是MySQL-python的另外一个分支，支持Python3并且修复了一些bug。
pymysql：纯Python实现的一个驱动，因为是纯Python编写的，因此执行效率不如MySQL-python。并且也因为是纯Python编写的，因此可以和Python代码无缝衔接。
MySQL Connector/Python：MySQL官方推出的使用纯Python连接MySQL的驱动。因为是纯Python开发的，效率不高。

推荐参考

Python MySQL 数据库连接 (一) mysql-connector-python

Python MySQL 数据库连接 (二) pymysql

1 官方推荐安装(从来没有安装成功过)

pip install mysqlclient

2 手动下载安装(一般都可以安装上)

找自己需要安装的包 https://www.lfd.uci.edu/~gohlke/pythonlibs/#mysqlclient
这里使用 mysqlclient‑1.4.6‑cp38‑cp38‑win32.whl

pip install mysqlclient‑1.4.6‑cp38‑cp38‑win32.whl

使用案例

# 安装 mysqlclient 地址 https://www.lfd.uci.edu/~gohlke/pythonlibs/#mysqlclient

import logging
import MySQLdb
from MySQLdb import cursors, _mysql
from MySQLdb._exceptions import Error as MySQLdbError

class DBQuery:
    """
    1
    sqluser = DBQuery().ment("SELECT * FROM `user` ")  # 直接使用

    2
    sqldb = DBQuery()  # sql 初始化
    fimes = sqldb.ment("SELECT * FROM tables WHERE username=%s AND id=%s", (usernmae, 1), close=False)  # 不关当前链接

    3
    connectdata = {'host': '127.0.0.1'}
    sqldb = DBQuery(**connectdata) # sql 初始化
    fimes = sqldb.ment("SELECT * FROM tables WHERE username=%s AND id=%s", (usernmae, 1))  # 自动关闭当前链接

    """
    def __init__(self, **kwargs):
        self.mysql_config = {
            'user': '',
            'passwd': '',
            'host': '',
            'db': '',
            'port': 3306,
            'connect_timeout': 10,
            'cursorclass': cursors.DictCursor,
        }
        self.kwargs = kwargs
        self.replaceconfig()
        self.config = self.mysql_config.copy()
        self.dictionary = True  # 返回类型 dict, 默认类型[()]
        self.conn = None
        self.cur = None

    def replaceconfig(self):
        """ 替换参数 """
        for k, v in self.kwargs.items():
            self.mysql_config[k] = v

    def connect(self):
        """ 数据库连接函数,建立连接和游标 """
        if self.conn and self.cur:
            return True

        try:
            self.conn = MySQLdb.connect(**self.config)
        except MySQLdbError:
            logging.error("数据库连接失败",  exc_info=True)
            return False
        self.cur = self.conn.cursor()
        return True

    def ment(self, statement, parameter=None, close=True):
        """ 数据执行语句函数 ,statement:语句,parameter:参数,必须是 元组 或者 列表,返回执行结果"""
        # print(statement,parameter)
        if not self.connect():
            return False
        if parameter:
            if type(parameter) not in (tuple, list):
                logging.warning("查询的参数必须是元组 (...) 或者 ")
                return False
        try:
            self.cur.execute(statement, parameter)

            if 'SELECT' == statement.strip().upper()[0:6] or 'SHOW' == statement.strip().upper()[0:4]:
                result = self.cur.fetchall()
            elif 'CREATE TABLE' == statement.strip().upper()[0:12]:
                result = True
            else:
                result = self.conn.commit()
                if statement.upper()[0:6] in ['INSERT', 'UPDATE']:
                    result = self.rowcount()
        except MySQLdbError as e:
            logging.error('SQL_ERROR: %s, statement:%s' % (e, statement))
            self.conn.rollback()
            result =  False
        finally:
            self.mclose(close)
        return result

    def lastd(self):
        """ 返回 INSERT 语句的id, 应该在插入后立即使用 """
        try:
            return self.cur.lastrowid
        except:
            pass

    def rowcount(self):
        """ 返回 INSERT or UPDATE 语句的更新数目 """
        try:
            return self.cur.rowcount
        except:
            pass

    def mclose(self, close=True):
        if close:
            self.dbclose()

    def dbclose(self):
        try:
            if self.cur:
                self.cur.close()
            if self.conn:
                self.conn.close()
            return True
        except:
            pass

其他

• 官网

https://github.com/PyMySQL/mysqlclient-python