实现一个轻量插件系统 #python #插件

python-plugin.py

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author    :h1code2
# File      :test.py
# Email     :[email protected]
# Time      :2021/2/18 18:23
# Software  :PyCharm

"""
原文地址：https://mp.weixin.qq.com/s/Z_sQRgBWgYF2tIRI1ut-kg
"""

plugins = {}


def register(plugin):
    plugins[plugin.__name__] = plugin
    return plugin


@register
def transfer_age_to_int(doc):
    try:
        doc['age'] = int(doc['age'])
    except ValueError:
        doc['age'] = 'N/A'
    return doc


@register
def filter_date(doc):
    date = doc['date']
    if date < '2020-05-01':
        return None
    return doc


def read_from_redis():
    data_list = [
        {'age': 34, 'name': 'xxx', 'date': '2020-05-10'},
        {'age': 12, 'name': 'yyy', 'date': '2020-04-03'},
        {'age': '23', 'name': 'zzz', 'date': '2020-05-12'},
        {'age': 'aa', 'name': 'abc', 'date': '2020-05-10'},
        {'age': 89, 'name': 'def', 'date': '2020-02-10'},
        {'age': '', 'name': 'xyz', 'date': '2020-05-10'},
        {'age': 'xy', 'name': 'xxx', 'date': '2020-05-10'},
        {'age': 'mp', 'name': 'xxx', 'date': '2020-05-10'},
        {'age': 34, 'name': 'xxx', 'date': '2019-02-10'},
    ]
    for data in data_list:
        yield data


def write_to_mongo(doc):
    print(doc)


def parse():
    for doc in read_from_redis():
        for name, plugin in plugins.items():
            print(f"正常运行插件：{name}")
            doc = plugin(doc)
            if not doc:
                print(f"数据：{doc}, 被插件：{name} 过滤")
                break
        else:
            write_to_mongo(doc)


if __name__ == '__main__':
    parse()