[Python] 两则装饰器

大约是三年前我写了优雅地使用python闭包 , 其实python的这颗装饰器“语法糖”本质上就是“闭包”，不熟悉的小伙伴可以先了解一下“闭包”哦。
使用python也已数年，确实感受到了这颗“语法糖”在开发中带给我的“红利”，以下就给大家分享我原创的两则装饰器吧。

1. sqlachemy查询装饰器

关于sqlachemy, 我之前也写过 简单粗暴使用SQLAlchemy, 平日的开发中对数据库的操作基本都是用的这个包。
但如果每次使用都需要建立连接、执行sql操作、提交操作、关闭连接的话，确实比较繁琐，所以这里我就想到了使用装饰器去完成这一操作，每次增删改查操作前只需要加个装饰器就可以了。
上代码：

class DBSession(object):
    """创建连接的session, [简单粗暴使用SQLAlchemy]中有写过"""
    def __init__(self, conf):
        user = conf.get('user')
        pwd = conf.get('pwd')
        host = conf.get('host')
        port = conf.get('port')
        db = conf.get('db')

        engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8'.format(user, pwd, host, port, db),
                               poolclass=NullPool, echo=False)
        _session = sessionmaker()
        _session.configure(bind=engine)

        self.session = _session(autocommit=True)

    def finalizer(self):
        self.session.flush()
        self.session.close()


def db_fixture(conf):
    """装饰器部分"""
    class DBFixture(object):
        def __init__(self):
            self.config = conf

        def __db_conn(self):
            """建立连接"""
            self.db = DBSession(self.config)
            self.db_session = self.db.session
            return self.db_session

        def __db_finalizer(self):
           """数据提交、关闭连接"""
            self.db_session.flush()
            self.db_session.close()

        def __call__(self, func):
            def wrapper(*args, **kwargs):
                session = self.__db_conn()      # 创建连接(session)
                try:
                    res = func(session=session, *args, **kwargs)    # 执行操作
                    return res
                except Exception as ex:
                    session.rollback()
                    logging.error(ex)
                    raise DBQueryError('db_error')
                finally:
                    self.__db_finalizer()

            return wrapper

   """返回实例化的DBFixture类, 会调用上述__call__方法"""
    return DBFixture()

2. API装饰器

目前我使用了django写后端接口，如果每个接口都单独使用 try except方法去捕获异常然后返回相应错误码的话，会造成代码的冗余，所以这里我写了一个通用的 request装饰器，如下：

def request_wrapper(func):
    """
    handle error for request
    """

    def handler(request):
        try:
            logging.info('[HTTP REQ] method={}, url={}, body={}'.format(request.method,
                                                                        request.get_full_path(),
                                                                        request.body.decode()))
            res = func(request)
            logging.info('[HTTP RES] result={}'.format(res.content.decode('unicode-escape')))
            return res

        except UnicodeDecodeError:
            res = func(request)
            logging.info('[HTTP RES] result={}'.format(res.content.decode()))
            return res

        except ParamInvalidError as ex:
            if str(ex):
                return make_error_result(code=ErrorCode.INVALID_PARAM, error=str(ex))
            return make_error_result(ErrorCode.INVALID_PARAM)

        except ParamMissingError as ex:
            if str(ex):
                return make_error_result(code=ErrorCode.MISSING_PARAM, error=str(ex))
            return make_error_result(ErrorCode.MISSING_PARAM)

        except StrOverflow:
            return make_error_result(ErrorCode.STR_OVERFLOW)

        except RecordDuplicate:
            return make_error_result(ErrorCode.RECORD_DUPLICATE)

        except RecordNotFound:
            return make_error_result(ErrorCode.RECORD_NOT_FOUND)

        except DBQueryError:
            record_error(request, ErrorCode.DB_ERROR.description)       #对于数据库错误，会有错误告警
            return make_error_result(ErrorCode.DB_ERROR)

        except OSGServiceError:
            record_error(request, ErrorCode.OSG_ERROR.description)
            return make_error_result(ErrorCode.OSG_ERROR)

        except NAGAServiceError:
            record_error(request, ErrorCode.NAGA_ERROR.description)
            return make_error_result(ErrorCode.NAGA_ERROR)

        except CommandError:
            record_error(request, ErrorCode.COMMAND_ERROR.description)
            return make_error_result(ErrorCode.COMMAND_ERROR)

        except ExecuteError as ex:
            if str(ex):
                return make_error_result(code=ErrorCode.EXECUTION_ERROR, error=str(ex))
            return make_error_result(ErrorCode.EXECUTION_ERROR)

        except Exception as ex:
            logging.error(ex)
            record_error(request, ex)
            return make_error_result(ErrorCode.SYSTEM_ERROR)

    return handler

具体使用方法如下 (无需再针对每个接口写异常方法了)：

@request_wrapper
def get_***_list(request):
    name = request.GET.get("name", '')
    strategy_id = request.GET.get("strategyId", '')
    page_id = int(request.GET.get("pageId", 1))
    page_size = int(request.GET.get("pageSize", 10))
    res = get_***_list_impl(name, strategy_id, page_id, page_size)
    return make_success_result(data=res)

以上代码以及文字皆为原创，如需转载，请表明出处。