参考：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()
