#coding:utf-8
import MySQLdb,re,logging,os
class OperationDb_interface(object): #定义类
def __init__(self):
#创建数据库的的连接
self.conn=MySQLdb.connect(host='localhost',user='root',
passwd='123456',db='world',port='3306',charset='utf8')
#创建游标
self.cur=self.conn.cursor()
#单条数据进行增删改
def op_sql(self,param):
try:
self.cur.execute(param) #执行sql语句
self.conn.commit()
return True
except MySQLdb.Error, 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)
results= self.cur.fetchone()
except MySQLdb.Error, e:
results= 'sql0001'
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 MySQLdb.Error, e:
results = 'sql0001'
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)
except:
return False
#插入多条数据
def insertMore(self,condition):
try:
self.cur.executemany(condition)
self.conn.commit()
return True
except MySQLdb.Error, 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 __del__(self):
if self.cur is not None:
self.cur.close()
if self.conn is not None:
self.conn.close()
if __name__ == "__main__":
test=OperationDb_interface() #实例化类
#results=test.selectOne('select * from city where id=1') # 查询单条数据
#results=test.selectAll('select * from city') #查询多条
'''results= test.insertMore("insert into city values(%s,%s,%s,%s)",
[(1001,'shanghai', 'EGY '', '50017'),
(1002,'beijing', 'ABW', '1000'),
(1003, 'tianjin', 'IDN', '2483556')]''' #插
入多条数据
# results = test.op_sql('update city set Population="15000" where
id=1003') # 更新
print results
但是插入多条数据结果并没有报错,但是数据库里并没有真正插入成功
网友评论