阿里云-云小站(无限量代金券发放中)
【腾讯云】云服务器、云数据库、COS、CDN、短信等热卖云产品特惠抢购

基于PyMySQL的数据库操作类使用

184次阅读
没有评论

共计 3555 个字符,预计需要花费 9 分钟才能阅读完成。

一 简介
    Python 和 MySQL 交互的模块有 MySQLdb 和 PyMySQL(pymysql),MySQLdb 是基于 C 语言编写的,而且 Python3 不在支持 MySQLdb。PyMySQL 是一个纯 Python 写的 MySQL 客户端,它的目标是替代 MySQLdb,可以在 CPython、PyPy、IronPython 和 Jython 环境下运行,PyMySQL 在 MIT 许可下发布。

    在开发基于 Python 语言的项目中,为了以后系统能兼容 Python3, 我们使用了 PyMySQL 替换了 MySQLdb。下面我们来熟悉一下 pymysql 的使用。
 
二 安装方式
  pymsql 的源码 https://github.com/PyMySQL/PyMySQL,目前还在持续更新。

安装要求:
 
Python — one of the following:
 
    CPython >= 2.6 or >= 3.3

    PyPy >= 4.0

    IronPython 2.7

MySQL Server — one of the following:

    MySQL >= 4.1 (tested with only 5.5~)

    MariaDB >= 5.1
安装
  pip install PyMySQL
  相关文档 , 强烈建议大家仔细阅读一遍。其用法和 MySQLdb 相差无几,核心用法一致。这样使用 pymysql 替换 mysqldb 的成本极小。

三 基于 pymysql 的数据库交互

 

#!/usr/bin/env python
 
# encoding: utf-8

“””
 
author: yangyi@youzan

time: 2015/6/8 上午 11:34

func: 基于 pymysql 的数据库交互类,支持事务提交和回滚,返回结果记录行数,和 insert 的最新 id

“””

import pymysql

from warnings import filterwarnings

filterwarnings(‘ignore’, category=pymysql.Warning)

CONNECT_TIMEOUT = 100

IP = ‘localhost’

PORT = 3306

USER = ‘root’

PASSSWORD = ”

 

class QueryException(Exception):

    “””
 
    “””

 

 

class ConnectionException(Exception):

    “””
 
    “””

class MySQL_Utils():

    def __init__(

            self, ip=IP, port=PORT, user=USER, password=PASSSWORD,

            connect_timeout=CONNECT_TIMEOUT, remote=False, socket=”, dbname=’test’):

        self.__conn = None

        self.__cursor = None

        self.lastrowid = None

        self.connect_timeout = connect_timeout

        self.ip = ip

        self.port = port

        self.user = user

        self.password = password

        self.mysocket = socket

        self.remote = remote

        self.db = dbname

        self.rows_affected = 0

 

 

    def __init_conn(self):

        try:
            conn = pymysql.connect(

                    host=self.ip,

                    port=int(self.port),

                    user=self.user,

                    db=self.db,

                    connect_timeout=self.connect_timeout,

                    charset=’utf8′, unix_socket=self.mysocket)

        except pymysql.Error as e:

            raise ConnectionException(e)

        self.__conn = conn

 

 

    def __init_cursor(self):

        if self.__conn:

            self.__cursor = self.__conn.cursor(pymysql.cursors.DictCursor)

 

 

    def close(self):

        if self.__conn:

            self.__conn.close()

            self.__conn = None

 

    #专门处理 select 语句
    def exec_sql(self, sql, args=None):

        try:

            if self.__conn is None:

                self.__init_conn()

                self.__init_cursor()

            self.__conn.autocommit = True

            self.__cursor.execute(sql, args)

            self.rows_affected = self.__cursor.rowcount

            results = self.__cursor.fetchall()

            return results

        except pymysql.Error as e:

            raise pymysql.Error(e)

        finally:

            if self.__conn:

                self.close()

 

    # 专门处理 dml 语句 delete,updete,insert 
    def exec_txsql(self, sql, args=None):

        try:

            if self.__conn is None:

                self.__init_conn()

                self.__init_cursor()

            if self.__cursor is None:

                self.__init_cursor()

 

 

            self.rows_affected=self.__cursor.execute(sql, args)
            self.lastrowid = self.__cursor.lastrowid
            return self.rows_affected
        except pymysql.Error as e:

            raise pymysql.Error(e)

        finally:

            if self.__cursor:

                self.__cursor.close()

                self.__cursor = None

 

    # 提交
    def commit(self):

        try:

            if self.__conn:

                self.__conn.commit()

        except pymysql.Error as e:

            raise pymysql.Error(e)

        finally:

            if self.__conn:

                self.close()

 

    #回滚操作
    def rollback(self):

        try:

            if self.__conn:

                self.__conn.rollback()
        except pymysql.Error as e:

            raise pymysql.Error(e)

        finally:

            if self.__conn:

                self.close()

    # 适用于需要获取插入记录的主键自增 id

    def get_lastrowid(self):

        return self.lastrowid

    #获取 dml 操作影响的行数
    def get_affectrows(self):
        return self.rows_affected
      #MySQL_Utils 初始化的实例销毁之后,自动提交
    def __del__(self):
        self.commit()
 
四 小结
  前几天刚刚将我们的系统中的 MySQLdb 替换为 PyMySQL, 还未遇到问题。欢迎大家使用测试上述脚本,有问题欢迎和我讨论。

本文永久更新链接地址:http://www.linuxidc.com/Linux/2017-06/144689.htm

正文完
星哥玩云-微信公众号
post-qrcode
 0
星锅
版权声明:本站原创文章,由 星锅 于2022-01-22发表,共计3555字。
转载说明:除特殊说明外本站文章皆由CC-4.0协议发布,转载请注明出处。
【腾讯云】推广者专属福利,新客户无门槛领取总价值高达2860元代金券,每种代金券限量500张,先到先得。
阿里云-最新活动爆款每日限量供应
评论(没有评论)
验证码
【腾讯云】云服务器、云数据库、COS、CDN、短信等云产品特惠热卖中