基础语法：
import pymysql #导入模块
conn = pymysql.connect(host=‘localhost‘,user=‘root‘, passwd=‘123456‘, db=‘test‘, port=3306, charset=‘utf8‘,cursorclass = pymysql.cursors.DictCursor) #连接数据库，cursorclass = pymysql.cursors.DictCursor表示把查询的返回值变为字典格式
cur = conn.cursor() #建立指针
cur.execute("insert into users (username,password,email) values (%s,%s,%s)",("老齐","9988","[email protected]")) #插入单条数据
cur.executemany("INSERT DELAYED INTO users (username,pwd,mail) VALUES (%s,%s,%s)",(‘lili‘, ‘ffff‘, ‘[email protected]‘),(‘lili‘, ‘sds‘, ‘[email protected]‘)) #插入多条数据
conn.commit() #提交保存
以下是在接口测试中的实际应用
# -*- coding:utf-8 -*-
import pymysql,logging,os
class OperationDb_interface(object):
def __init__(self):
self.conn=pymysql.connect(host=‘localhost‘,
user=‘root‘, passwd=‘123456‘, db=‘test‘, port=3306, charset=‘utf8‘,cursorclass = pymysql.cursors.DictCursor) #创建数据库连接
self.cur=self.conn.cursor() #创建游标
#定义单条数据操作，增删改
def op_sql(self,param):
try:
self.cur.execute(param) #游标下执行sql语句
self.conn.commit() #提交数据
return True
except pymysql.Error as e:
print(‘Mysql Error %d:%s‘ % (e.args[0], e.args[1]))
logging.basicConfig(filename = os.path.join(os.getcwd(), ‘./log.txt‘),
level = logging.DEBUG, format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘)
logger = logging.getLogger(__name__)
logger.exception(e)
return False
#查询表中单条数据
def selectOne(self,condition):
try:
self.cur.execute(condition) #执行sql语句
results = self.cur.fetchone() #获取一条结果
except pymysql.Error as e:
results =‘spl0001‘
print(‘Mysql Error %d:%s‘ %(e.args[0], e.args[1]))
logging.basicConfig(filename = os.path.join(os.getcwd(), ‘./log.txt‘),
level = logging.DEBUG, format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘)
logger = logging.getLogger(__name__)
logger.exception(e)
finally:
return results
#查询表中多条数据
def selectAll(self,condition):
try:
self.cur.execute(condition)
self.cur.scroll(0, mode=‘absolute‘) #游标里的光标回到初始位置
results = self.cur.fetchall() #返回游标中所有结果
except pymysql.Error as e:
results=‘spl0001‘
print(‘Mysql Error %d:%s‘ %(e.args[0], e.args[1]))
logging.basicConfig(filename = os.path.join(os.getcwd(), ‘./log.txt‘),
level = logging.DEBUG, format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘)
logger = logging.getLogger(__name__)
logger.exception(e)
finally:
return results
#定义表中插入多条数据操作
def insertMore(self, condition,argsall):
try:
self.cur.executemany(condition,argsall)
self.conn.commit()
return True
except pymysql as e:
results =‘spl0001‘ #数据库执行错误
print(‘Mysql Error %d:%s‘ % (e.args[0], e.args[1]))
logging.basicConfig(filename = os.path.join(os.getcwd(), ‘./log.txt‘),
level = logging.DEBUG, format=‘%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s‘)
logger = logging.getLogger(__name__)
logger.exception(e)
return False
#数据库关闭
def __del__(self):
if self.cur != None:
self.cur.close()
if self.conn != None:
self.conn.close()
if __name__ == ‘__main__‘:
test=OperationDb_interface() #实例化类
sql1="insert into users (username,pwd,mail) VALUES (%s,%s,%s)"
argsall=[(‘lilyu‘,‘1234‘,‘[email protected]‘),(‘lilu‘,‘124‘,‘[email protected]‘),(‘lil‘,‘14‘,‘[email protected]‘)]
result=test.insertMore(sql1,argsall)
print(result)
原文：http://www.cnblogs.com/lansan0701/p/7454745.html