Python封装MySQL类

0x00 安装

有两种,一个是MySQLdb,一个是pymysql

  • 下载MySQL-python然后安装
  • sudo pip install pymysql(推荐,因为py3已经不支持MySQLdb了)

0x01 简单表设计如下

insert into mysql.user(Host,User,Password) values('%','python','123456');

drop database if exists python;
create database python;
use python;

drop table if exists msg;
create table msg(
id int not null auto_increment primary key,
ip varchar(40) not null default '127.0.0.1' comment 'ip地址',
domain varchar(100) not null default 'www.xx.com' comment '域名'
);

grant all privileges on python.* to 'python'@'%' identified by '123456';
flush privileges;

0x02 MySQLdb封装代码

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import MySQLdb

class mysql(object):
    """docstring for mysql"""
    def __init__(self, dbconfig):
        self.host = dbconfig['host']
        self.port = dbconfig['port']
        self.user = dbconfig['user']
        self.passwd = dbconfig['passwd']
        self.dbname = dbconfig['dbname']
        self.charset = dbconfig['charset']
        self._conn = None
        self._connect()
        self._cursor = self._conn.cursor()

    def _connect(self):
        try:
            self._conn = MySQLdb.connect(host=self.host,
                port = self.port,
                user=self.user,
                passwd=self.passwd,
                db=self.dbname,
                charset=self.charset)
        except MySQLdb.Error,e:
            print e
            
    def query(self, sql):
        try:
            result = self._cursor.execute(sql)
        except MySQLdb.Error, e:
            print e
            result = False
        return result

    def select(self, table, column='*', condition=''):
        condition = ' where ' + condition if condition else None
        if condition:
            sql = "select %s from %s %s" % (column,table,condition)
        else:
            sql = "select %s from %s" % (column,table)
        self.query(sql)
        return self._cursor.fetchall()

    def insert(self, table, tdict):
        column = ''
        value = ''
        for key in tdict:
            column += ',' + key
            value += "','" + tdict[key]
        column = column[1:]
        value = value[2:] + "'"
        sql = "insert into %s(%s) values(%s)" % (table,column,value)
        try:
            self._cursor.execute(sql)
            self._conn.commit()
        except:
            self.rollback()
        return self._cursor.lastrowid #返回最后的id

    def update(self, table, tdict, condition=''):
        if not condition:
            print "must have id"
            exit()
        else:
            condition = 'where ' + condition
        value = ''
        for key in tdict:
            value += ",%s='%s'" % (key,tdict[key])
        value = value[1:]
        sql = "update %s set %s %s" % (table,value,condition)
        try:
            self._cursor.execute(sql)
        except:
            self.rollback()
        return self.affected_num() #返回受影响行数

    def delete(self, table, condition=''):
        condition = 'where ' + condition if condition else None
        sql = "delete from %s %s" % (table,condition)
        try:
            self._cursor.execute(sql)
            self._conn.commit()
        except:
            self.rollback()
        return self.affected_num() #返回受影响行数

    def rollback(self):
        self._conn.rollback()

    def affected_num(self):
        return self._cursor.rowcount

    def __del__(self):
        try:
            self._cursor.close()
            self._conn.close()
        except:
            pass

    def close(self):
        self.__del__()

if __name__ == '__main__':
    dbconfig = {
        'host':'192.168.188.134',
        'port':3306,
        'user':'python',
        'passwd':'123456',
        'dbname':'python',
        'charset':'utf8'
    }
    db = mysql(dbconfig)

    # print db.select('msg','id,ip,domain')
    # print db.select('msg','id,ip,domain','id>2')
    # print db.affected_num()

    # tdict = {
    #     'ip':'111.13.100.91',
    #     'domain':'baidu.com'
    # }
    # print db.insert('msg', tdict)
    
    # tdict = {
    #     'ip':'111.13.100.91',
    #     'domain':'aaaaa.com'
    # }
    # print db.update('msg', tdict, 'id=5')

    # print db.delete('msg', 'id>3')

    db.close()

0x03 pymysql封装代码

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import urlparse
import pymysql
import contextlib

class MySQLX(object):
    def __init__(self, mysql_uri):
        super(MySQLX, self).__init__()
        self.mysql_uri = mysql_uri
        self.mysql_info = self.mysql_parse_uri()

    def mysql_parse_uri(self):
        p = urlparse.urlparse(self.mysql_uri)
        host = p.hostname
        port = p.port
        user = p.username
        password = p.password
        dbname = p.path.strip('/')
        charset = urlparse.parse_qs(p.query)['charset'][0]

        return {
            'host': host,
            'port': port,
            'user': user,
            'password': password,
            'db': dbname,
            'charset': charset,
            'cursorclass': pymysql.cursors.DictCursor,
        }

    @contextlib.contextmanager
    def init(self):
        dbconn = pymysql.connect(**self.mysql_info)
        cursor = dbconn.cursor()
        # dbconn = pymysql.connect(
        #     host=mysql_info.get('host'),
        #     user=mysql_info.get('user'),
        #     password=mysql_info.get('password'),
        #     db=mysql_info.get('db'),
        #     charset=mysql_info.get('charset')
        #     )
        # cursor = dbconn.cursor(cursor=pymysql.cursors.DictCursor)

        try:
            yield cursor #这里就是with返回的
        finally:
            dbconn.commit()
            cursor.close()
            dbconn.close()

    def query(self, sql, arg=''):
        try:
            with self.init() as cursor:
                if arg:

                    cursor.execute(sql,arg) #返回受影响行数
                else:
                    cursor.execute(sql)
                result = cursor.fetchall() #返回数据格式是[{},{}]
                # result = cursor.fetchone() #返回数据格式是{}
                return result
        except Exception as e:
            print sql,str(e)

if __name__ == '__main__':
    mysql_uri = "mysql+pymysql://root:root@localhost:3306/rtest?charset=utf8mb4"
    sqlconn = MySQLX(mysql_uri)
    sql = "select * from `msg` where id=%s"
    result = sqlconn.query(sql,"2")
    print result
若未作声明则文章版权归本人(@reber)所有,转载请注明原文链接:

可以返回顶部