Python数据库工具类MySQLdb使用

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

# -*- coding: utf-8 -*-      
#mysqldb      
import time, MySQLdb      
     
#连接      
conn=MySQLdb.connect(host="localhost",user="root",passwd="root",db="test",charset="utf8")    
cursor = conn.cursor()      
 
#删除表  
sql = "drop table if exists user"  
cursor.execute(sql)  
 
#创建  
sql = "create table if not exists user(name varchar(128) primary key, created int(10))"  
cursor.execute(sql)  
 
#写入      
sql = "insert into user(name,created) values(%s,%s)"    
param = ("aaa",int(time.time()))      
n = cursor.execute(sql,param)      
print 'insert',n      
     
#写入多行      
sql = "insert into user(name,created) values(%s,%s)"    
param = (("bbb",int(time.time())), ("ccc",33), ("ddd",44) )  
n = cursor.executemany(sql,param)      
print 'insertmany',n      
 
#更新      
sql = "update user set name=%s where name='aaa'"    
param = ("zzz")      
n = cursor.execute(sql,param)      
print 'update',n      
     
#查询      
n = cursor.execute("select * from user")      
for row in cursor.fetchall():      
    print row  
    for r in row:      
        print r      
     
#删除      
sql = "delete from user where name=%s"    
param =("bbb")      
n = cursor.execute(sql,param)      
print 'delete',n      
 
#查询      
n = cursor.execute("select * from user")      
print cursor.fetchall()      
 
cursor.close()      
     
#提交      
conn.commit()  
#关闭 
conn.close()
     

封装类操作

此处ConfigUtils工具类为Python配置工具类ConfigParser使用提供。在使用数据库连接时，建议每次调用利用try finally机制，做好资源回收。在对于异常处理时，其实不建议像如此处理。

 class DButils(object):
    def __init__(self,filename,section):
        super(DButils, self).__init__()
        #read config
        cfg = ConfigUtils(filename).config
        self.cfg = cfg
        self.section = section
        #init mysql connection
        self.conn= MySQLdb.connect(
            host=cfg.get(section,'host'),
            port = cfg.getint(section,'port'),
            user=cfg.get(section,'user'),
            passwd=cfg.get(section,'passwd'),
            db=cfg.get(section,'db'),
            connect_timeout=cfg.getint(section,'connect_timeout')
        )
        self.cur = self.conn.cursor()    
     
    def fetchmany(self,sql):
        sql = sql.replace('{$db}',self.cfg.get(self.section,'db'))
        try:
            return self.cur.fetchmany(self.cur.execute(sql))
        except Exception, e:
            print traceback.print_exc()
            print sql
   
    def fetchone(self,sql):
        sql = sql.replace('{$db}',self.cfg.get(self.section,'db'))
        try:
            self.cur.execute(sql)
            return self.cur.fetchone()
        except Exception, e:
            print traceback.print_exc()
            print sql    def create(self,sql):
        try:
            self.cur.execute(sql)
            self.conn.commit()
        except Exception, e:
            print traceback.print_exc()    def is_table_exit(self,tableName):
        show_sql = 'show tables;'
        try:
            return tableName in self.cur.fetchmany(self.cur.execute(show_sql))
        except Exception,e:
            print traceback.print_exc()
    def close_db(self):
        self.cur.close()
        self.conn.close()db    = DButils('ini.cfg','src_db')
try:
    db.fetchone('select * from table limit 1')
finally:
    db.close_db()