文档章节

利用反射实现Python职责链模式

木头释然
 木头释然
发布于 2018/07/09 10:56
字数 903
阅读 101
收藏 0

    在数据处理中常常会遇到数据处理需要使用多个处理流程,不同数据使用的处理流程常有重复,如果对于每一种数据都单独做一套处理流程不仅耗费时间,也不利于代码的维护,故而可以将不同的处理流程分别作为模块,对于不同数据通过定义好输入输出格式,或者在必要时在不同模块中间添加格式转换模块来实现对于数据的链式处理流程。

    首先将基础处理流程定义为BaseProcessor,所有流程都将从该类继承。

class BaseProcessor(object):
    
    def __init__(self, **kwargs):
        self.config = kwargs['config']
    
    def doProcess(self, *args, **kwargs):
        raise NotImplementedError

假设现在我们有两个处理流程分别为AProcessor,BProcessor,CProcessor分别继承自BaseProcess,有两种数据,数据1需要A,B两种处理方式,数据2需要AC两种处理方式,那么配置文件如下:

{
    "types": [
        {
            "name": "数据1",
            "processors": [
                {
                    "module_name": "abc.processors.a_processor",
                    "class_name": "APorcessor"
                },
                {
                    "module_name": "abc.processors.b_processor",
                    "class_name": "BPorcessor"
                }
            ]
        },
        {
            "name": "数据2",
            "processors": [
                {
                    "module_name": "abc.processors.a_processor",
                    "class_name": "APorcessor"
                },
                {
                    "module_name": "abc.processors.c_processor",
                    "class_name": "CPorcessor"
                }
            ]
        }
    ],
    "processors": {
        "APorcessor": {
            "bin_path": "APorcessor.bin",
            "class_types":["yes","no"]
        },
        "BPorcessor": {
            "bin_path": "BPorcessor.bin"
        },
        "CPorcessor": {
            "plus":1
        }
    }
}

types定义了两种数据类型:数据1和数据2以及他们分别的处理流程,processors定义了每种处理流程一些必要的参数。我们还需要一个读取json文件的工具,在这里定义为单例模式:

import json

class Singleton(object):
    _instance = None
    def __new__(cls, *args, **kw):
        if not cls._instance:
            cls._instance = super(Singleton, cls).__new__(cls)
        return cls._instance

class JsonUtil(Singleton):

    def __init__(self):
        self.config={}
    
    def load(self, filePath):
        with open(filePath, 'r', encoding='utf-8') as stream:
            self.config = json.load(stream)  
        return self.config

好了,准备工作已经完成了,下面就要实现将不同处理流程串成链,然后调用这个链。在这里需要用到getattr函数,首先说明一下getattr函数的使用方法:它接收2个参数,前面的是一个对象或者模块,后面的是一个字符串。

import logging
import os
import sys
from abc.utils.json_util import JsonUtil


class Application(object):

    def __init__(self):
        jsonPath = os.path.join(os.path.dirname(__file__), 'config.json')
        self.config = JsonUtil().load(jsonPath)
        self.processors = dict()
        self.baseProcessorModule = 'abc.processors'

    def getAllAnalysisType(self):
        types = []
        if self.config is not None:
            types = self.config['types']
        return mvp_types

    def getPorcessorsByTypeName(self, typeName):
        types = self.getAllAnalysisType()
        processors = None
        for t in types:
            if t['name'] == typeName:
                processors = t['processors']
        return processors

    def makeProcessor(self, typeName, processorInfo):
        pname = '{}.{}'.format(processorInfo['class_name'], typeName)
        if self.processors.__contains__(pname):
            processor = self.processors[pname]
        else:
            className = processorInfo['class_name']
            moduleName = processorInfo['module_name']
            fullModuleName = '{}.{}'.format(
                self.baseProcessorModule, moduleName)
            module = __import__(fullModuleName, fromlist=(moduleName))
            classTypeMeta = getattr(module, className)
            config = self.getProcessorConfig(className)
            processor = classTypeMeta(config)
            self.processors[pname] = processor
        return processor

    def getProcessorConfig(self, processorName):
        processorConfig = {}
        processors = self.config['processors']
        if processors is not None:
            processorConfig = processors[processorName]
        return processorConfig

    def runAnalysis(self, typeName, content):
        pinfos = self.getPorcessorsByTypeName(typeName)
        context = content
        if pinfos is None or len(pinfos) == 0:
            return -1
        else:
            for p in pinfos:
                processor = self.makeProcessor(typeName, p)
                context = processor.doProcess(context)
            return context

好了,通过以上代码就可以实现数据的链式处理,processors会将之前所有使用过的流程实例保存,这样做的好处是有些处理流程需要加载大文件且速度较慢,避免重复实例化浪费时间。

需要注意的是,定义处理流程时必须定义好上下处理流程输出输入关系,否则会造成上一个处理流程输出结果与下一个处理流程输入不一致,导致数据丢失。

© 著作权归作者所有

木头释然
粉丝 17
博文 9
码字总数 5035
作品 0
西青
其他
私信 提问
《大话设计模式》Python版代码实现

 上一周把《大话设计模式》看完了,对面向对象技术有了新的理解,对于一个在C下写代码比较多、偶尔会用到一些脚本语言写脚本的人来说,很是开阔眼界。《大话设计模式》的代码使用C#写成的,...

若虚道人
2014/08/14
396
2
python面试中较常问及的知识点梳理---高级特性

版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。 https://blog.csdn.net/Nikki0126/article/details/97793753 • 高级特性 1.函数装饰器有什么...

等我想个好名字
08/28
0
0
MSSQL · 最佳实践 · RDS SDK实现数据库迁移上阿里云RDS SQL Server

title: MSSQL · 最佳实践 · RDS SDK实现数据库迁移上阿里云RDS SQL Server author: 风移 摘要 至今,我们完成了SQL Server备份还原专题系列七篇月报分享:三种常见的数据库备份、备份策略的...

风移
2018/07/18
0
0
职责链

1.场景问题 1.1 申请聚餐费用 来考虑这样一个功能:申请聚餐费用的管理。 很多公司都有这样的福利,就是项目组或者是部门可以向公司申请一些聚餐费用,用于组织项目组成员或者是部门成员进行...

ciyo_yang
2017/07/12
0
0
一款通过寻找和串联Gadget来构建ROP漏洞利用的工具

  今天给大家介绍的是一款名叫ROPGenerator的ROP漏洞利用开发工具,它能够自动寻找gadget,并构建ROP漏洞利用链,该工具当前版本支持x86和x64源码。       工具介绍   ROPGenerator使...

FreeBuf
2018/07/28
0
0

没有更多内容

加载失败,请刷新页面

加载更多

无回路有向图的拓扑排序

因公司业务需要,在表单中每个字段都会配置自动计算,但自动计算公式中会引用到其他字段中的值。所以希望可以根据计算公式,优先计算引用的公式。所以最终使用了无回路有向图的扩扑排序来实现...

兜兜毛毛
今天
6
0
如何抢占云栖大会C位?史上最强强强攻略来了

点击观看视频: APSARA云栖大会开发者情怀 原文链接 本文为云栖社区原创内容,未经允许不得转载。

阿里云官方博客
今天
6
0
Kubernetes 从懵圈到熟练:集群服务的三个要点和一种实现

作者 | 声东 阿里云售后技术专家 文章来源:Docker,点击查看原文。 以我的经验来讲,理解 Kubernetes 集群服务的概念,是比较不容易的一件事情。尤其是当我们基于似是而非的理解,去排查服务...

阿里巴巴云原生
今天
11
0
PHP7.3的新特性

2018年12月6日,PHP7.3正式版发布,在PHP7.2基础上进行了大量错误修复和安全优化,性能提升10%! 从目前的更新说明来看,PHP 7.3 并不是一个主打新特性的版本,包含更多的是 bug 修复。PHP 7...

迅睿CMS-PHP开源CMS程序
今天
8
0
Tomcat 应用中并行流带来的类加载问题

本文首发于 vivo互联网技术 微信公众号 链接:https://mp.weixin.qq.com/s/f-X3n9cvDyU5f5NYH6mhxQ 作者:肖铭轩、王道环 随着 Java8 的不断流行,越来越多的开发人员使用并行流(parallel)...

vivo互联网技术
今天
8
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部