从零开始,贝博集团论坛

快捷导航
广告联系qq1031180668贝博集团
查看: 50|回复: 0
打印 上一主题 下一主题

[python] Python使用装饰器实现事务的控制

[复制链接]
  • ta_mind

    2019-8-30 13:16
  • classn_01: 73 classn_02

    [LV.6]常住居民II

    1041

    主题

    1763

    帖子

    3623

    积分

    管理员

    Rank: 9Rank: 9Rank: 9

    积分
    3623

    最佳新人活跃会员热心会员推广达人宣传达人灌水之王突出贡献优秀版主荣誉管理论坛元老

    跳转到指定楼层
    楼主
    发表于 2019-8-31 17:23:06 | 只看该作者 回帖奖励 |倒序浏览 |阅读模式
    自己写的一个db操作类依赖
    • import pymysql
    • from DBUtils.PersistentDB import PersistentDB
    • import logging as log(可选)
    [Python] syntaxhighlighter_viewsource syntaxhighlighter_copycode
    class Db:
        """
        数据库连接池-数据库操作工具
        """
    
        def __init__(self):
            self.pool = PersistentDB(
                creator=pymysql,  # 使用链接数据库的模块
                maxusage=None,
                setsession=[],  # 开始会话前执行的命令列表。如:["set datestyle to ...", "set time zone ..."]
                ping=0,
                closeable=False,
                threadlocal=None,
                host=config.DB_HOST,
                port=config.DB_PROT,
                user=config.DB_USER,
                password=config.DN_PASS,
                database=config.DB_NAME,
                charset='utf8'
            )
    
        def select(self, sql):
            """
            执行查询方法 返回list
            :param sql: sql
            :return:
            """
            conn = self.pool.connection()
            cursor = conn.cursor()
            cursor.execute(sql)
            result = cursor.fetchall()
            index_dict = get_index_dict(cursor)
            res = []
            for rowS in result:
                dataDict = dict()
                for index in index_dict:
                    dataDict[index] = rowS[index_dict[index]]
                res.append(dataDict)
            cursor.close()
            conn.close()
            return res
    
        def executeConn(self, conn, sql):
            """
            执行sql 返回影响行数,自动提交并回滚,需要搭配@tran使用
            :param conn:
            :param sql:
            :return:
            """
            cursor = conn.cursor()
            return cursor.execute(sql)
    
        def execute(self, sql):
            """
            执行sql 返回影响行数
            :param sql:
            :return:
            """
            conn = self.pool.connection()
            cursor = conn.cursor()
            try:
                # 执行SQL语句
                num = cursor.execute(sql)
                # 提交到数据库执行
                conn.commit()
                return num
            except Exception as e:
                log.error('数据库执行异常:{sql} \n 错误:{e}'.format(sql=sql, e=e))
                # 发生错误时回滚
                conn.rollback()
            finally:
                cursor.close()
                conn.close()



    事务核心处理
    [Python] syntaxhighlighter_viewsource syntaxhighlighter_copycode
    from functools import wraps
    from support.util import db
    import logging as log
    def tran(connName='_YUXUAN_TRAN_CONN_', error={}):
        """
        事务控制
        :param connName: conn的名字
        :param error: 错误返回
        :return:
        """
    
        def wrapper(func):
            @wraps(func)
            def inner_wrapper(*args, **kwargs):
                conn = db.pool.connection()
                kwargs[connName] = conn
                result = error
                try:
                    result = func(*args, **kwargs)
                    conn.commit()
                except Exception as e:
                    log.error(e)
                    conn.rollback()
                finally:
                    conn.close()
    
                return result
    
            return inner_wrapper
    
        return wrapper


    @tran
    参数1 connName=获取连接的名称
    参数2 error=执行失败的时候默认返回值

    不积跬步,无以至千里
    游客
    贝博官网下载 回复
    您需要登录后才可以回帖 登录 | 立即注册

    手机版|Archiver|小黑屋|sitemap| 从零开始,贝博集团论坛 - 一个单纯的贝博集团学习交流论坛 ( 豫ICP备15032706号 )

    GMT+8, 2019-9-11 20:44 , Processed in 1.093396 second(s), 23 queries .

    Powered by Discuz! X3.4

    ? 2001-2013 Comsenz Inc.

    快速回复 返回顶部 返回列表