python的pymysql使用

参考：https://www.cnblogs.com/liubinsh/p/7568423.html

在python3.x中，可以使用pymysql连接数据库并进行对应操作

pymysql安装

pip3 install pymysql

数据库连接

测试使用的数据库和表

CREATE DATABASE lianxi;
CREATE TABLE `coupon_info` (
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
  `enabled` tinyint(3) unsigned NOT NULL DEFAULT '1' COMMENT '0无意义;1有效;2无效',
  `user_id` varchar(64) COLLATE utf8_bin NOT NULL DEFAULT '' COMMENT '用户ID',
  `coupon_id` varchar(64) COLLATE utf8_bin NOT NULL COMMENT '优惠券ID',
  `coupon_name` varchar(128) COLLATE utf8_bin NOT NULL DEFAULT '' COMMENT '优惠券名称',
  `start_time` datetime NOT NULL DEFAULT '1000-10-10 10:00:00' COMMENT '优惠券最早使用时间',
  `end_time` datetime NOT NULL DEFAULT '1000-10-10 10:00:00' COMMENT '优惠券最晚使用时间',
  `create_time` datetime NOT NULL DEFAULT '1000-10-10 10:00:00' COMMENT '优惠券信息创建时间',
  `modify_time` datetime NOT NULL DEFAULT '1000-10-10 10:00:00' COMMENT '优惠券信息修改时间',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_open_id` (`user_id`,`coupon_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin COMMENT='用户优惠券信息表';

数据库连接、关闭

    def __init__(self,host='127.0.0.1', user='root',password='1234567', database='lianxi'):
        #连接数据库
        self.conn = pymysql.connect(
            host=host,
            user=user,
            password=password,
            database=database,
            charset='utf8'
        )
        #创建一个光标
        self.cursor = self.conn.cursor()

    def close(self):
        """关闭光标对象、数据库连接
        """
        self.cursor.close()
        self.conn.close()

数据库操作

增删改操作

以插入数据为例（删、改类似）

       def insert(self,sql):
        """将数据插入数据库
        :param sql:
        :return:是否执行成功
        """
        try:
            #执行sql并返回受影响的行数
            update_row = self.cursor.execute(sql)
            #提交数据
            self.conn.commit()
        except Exception as e:
            #插入失败回滚
            self.conn.rollback()
            #设置执行失败
            update_row = False
            logging.error("数据库插入执行失败sql:"+str(sql))
        #根据受影响的行数判断数据是否插入成功
        if update_row > 0:
            update_row = True
        else:
            update_row = False
        #获取最后的主键(表中有自增主键时)
        #last_id = self.cursor.lastrowid
        #print(last_id)
        return update_row

数据库查询

    def select_all(self,sql):
        #返回所有数据
        self.cursor.execute(sql)
        return self.cursor.fetchall()

    def select_chmany(self,sql,num):
        #返回指定数量的数据
        self.cursor.execute(sql)
        return self.cursor.fetchmany(num)

    def select_one(self, sql):
        #返回第一条数据
        self.cursor.execute(sql)
        return self.cursor.fetchone()

    def select_a(self,sql):
        self.cursor.execute(sql)
        res = self.cursor.fetchall()
        print(res)
        #设置光标，如果这里不正确设置光标，下面的res1会返回空
        #absolute=绝对位置，relative=相对位置
        self.cursor.scroll(0,mode='absolute')
        # self.cursor.scroll(-3,mode='relative')
        res1 = self.cursor.fetchall()
        print(res1)

以上方法返回结果：

#执行代码：
    cursor = Cursor()
    sql = "select user_id, coupon_id, coupon_name from coupon_info where coupon_id like 'ada%'"
    coupon_data1 = cursor.select_all(sql)
    coupon_data2 = cursor.select_one(sql)
    coupon_data3 = cursor.select_chmany(sql,2)
    print(coupon_data1)
    print(coupon_data2)
    print(coupon_data3)
    cursor.close()
#返回结果：
#返回所有
(('123456', 'adads', '测试优惠券'), ('123456', 'adadaa', '测试优惠券'), ('123456', 'adadaa1', '测试优惠券'), ('123456', 'adad', '测试优惠券'))
#返回第一条
('123456', 'adads', '测试优惠券')
#返回指定条数
(('123456', 'adads', '测试优惠券'), ('123456', 'adadaa', '测试优惠券'))

总结：

• execute() 执行单条sql，返回受影响的行数
• executemany() 执行多条sql，返回受影响的行数
• commit() 增删改后提交数据
• rollback() 回滚操作
• lastrowid 返回最后的自增主键(新增时使用)
• fetchall() 查询时返回所有数据
• fetchone() 查询时返回第一条数据
• fetchmany(n) 查询时返回指定条数
• scroll(0,mode='absolute') 光标设置（absolute绝对位置，relative相对位置）

多条sql写法参考：

sql = "INSERT INTO userinfo(username,passwd) VALUES(%s,%s)"
cursor.executemany(sql,[("tom","123"),("alex",'321')])

整体代码：

class Cursor:
    """
    数据库操作
    """
    def __init__(self,host='127.0.0.1', user='root',password='1234567', database='lianxi'):
        self.conn = pymysql.connect(
            host=host,
            user=user,
            password=password,
            database=database,
            charset='utf8'
        )
        self.cursor = self.conn.cursor()

    def insert(self,sql):
        """将数据插入数据库
        :param sql:
        :return:是否执行成功
        """
        try:
            update_row = self.cursor.execute(sql)
            self.conn.commit()
        except Exception as e:
            self.conn.rollback()
            update_row = False
            logging.error("数据库插入执行失败sql:"+str(sql))
        if update_row > 0:
            update_row = True
        else:
            update_row = False
        # last_id = self.cursor.lastrowid
        # print(last_id)
        return update_row

    def select_all(self,sql):
        self.cursor.execute(sql)
        return self.cursor.fetchall()

    def select_chmany(self,sql,num):
        self.cursor.execute(sql)
        return self.cursor.fetchmany(num)

    def select_one(self, sql):
        self.cursor.execute(sql)
        return self.cursor.fetchone()

    def select_a(self,sql):
        self.cursor.execute(sql)
        res = self.cursor.fetchall()
        print(res)
        self.cursor.scroll(0,mode='absolute')
        # self.cursor.scroll(-3,mode='relative')
        res = self.cursor.fetchall()
        print(res)

    def close(self):
        """关闭光标对象、数据库连接
        """
        self.cursor.close()
        self.conn.close()

if __name__ == '__main__':
    cursor = Cursor()
    sql = "select user_id, coupon_id, coupon_name from coupon_info where coupon_id like 'ada%'"
    coupon_data1 = cursor.select_all(sql)
    coupon_data2 = cursor.select_one(sql)
    coupon_data3 = cursor.select_chmany(sql,2)
    print(coupon_data1)
    print(coupon_data2)
    print(coupon_data3)
    cursor.close()