使用项目框架管理工具Pyouter的实际应用

文章目录

  • pip包的安装
  • 项目目录结构
  • 项目实现
  • main.py
  • options.py
  • config.py
  • 二级叶子结点test实现
  • 三级叶子结点server实现
  • 项目结构细节
  • 无论是个人开发还是团队协作,在进行代码开发的时候,统一的代码规范是十分重要的。

    这里说的代码规范指的不是变量或者函数的命名规范,而是项目结构的规范。

    在这里,我要介绍一个由@幻灰龙和@ccat开发的开源项目pyouter。这个项目很好的为开发者提供了一个规范的项目框架,不管是个人开发还是团队开发,都可以通过使用这套框架来共同规范起开发逻辑。

    pyouter的内部原理实现可以参考https://blog.csdn.net/huanhuilong/article/details/121481377或者https://github.com/fanfeilong/pyouter,这篇文章只做项目的实践介绍。

    测试项目的github地址为pyouter_test

    pip包的安装

    # https://pypi.org/project/pyouter/0.0.1
    pip install pyouter
    

    项目目录结构

    假设我们开发的项目名为pyouter_test,我们的项目目录应该如下所示:

  • pyouter_test
  • src
  • data(可自定义)
  • test(可自定义)
  • server(可自定义)
  • config
  • config.py
  • main.py
  • options.py
  • 项目实现

    定义好我们的目录结构之后,首先进行main.py和options.py的实现。
    main.py是我们整个项目的入口,options.py是实现整个项目结构的逻辑

    main.py

    from pyouter.app import App
    from pyouter.router import Router
    
    from options import get_args_parser
    from config.config import load_config
    
    def main_dispatch():
        """ 主路由
    
        """
    
    
        router = Router(
            init = lambda config,options: print('server_ip:{}'.format(config['server_ip'])),
        )
    
        return router
    
    
    def run():
        """ 入口函数
        
        """
    
        # 获取命令行参数
        args_parser = get_args_parser()
        options = args_parser.parse_args()
        # 加载配置文件
        config = load_config(options)
    
        app = App(
            config=config,
            parser=args_parser
        )
    
        app.use(
            router=Router(
                pyouter_test = main_dispatch()
            )
        )
    
        app.run()
    
    
    if __name__=="__main__":
        run()
    

    options.py

    from argparse import ArgumentParser
    from pyouter.default import create_parser
    
    
    def add_custom_arguments(parser: ArgumentParser):
        ''' 用户定制化参数选项
        
        * --cluster: 开发环境还是线上环境
        '''
        parser.add_argument(
            "--cluster",
            dest="cluster",
            help="cluster dev or pro",
            default="dev",
            nargs='?',
            type=str,
            metavar="CLUSTER"
        )
    
    
    
    def get_args_parser():
        parser = create_parser("pyouter test")
        add_custom_arguments(parser)
    
        return parser
    
    
    def show_help():
        """
        命令行选项说明:
        ==
        """
    
        help = '\n'.join([
            show_help.__doc__,
            add_custom_arguments.__doc__
        ])
    
        print(help)
    

    我们在main函数里面配置好了整个项目的入口,在options.py里面设置了一个参数"–cluster"。
    我们还在main函数里面调用了config/config.py里面的load_config函数,这个函数load了一个字典格式的config,方便我们对项目的环境进行配置,类似以下代码:

    config.py

    def load_config(options):
        if options.cluster == 'dev':
            config = {'server_ip':'127.0.0.1'}
        elif options.cluster == 'pro':
            config = {'server_ip':'192.168.0.1'}
        return config
    

    写完上面三个python代码后可以在src目录下运行命令

    python main.py pyouter_test.init --cluster dev
    

    会打印结果

    如果运行命令

    python main.py pyouter_test.init --cluster pro
    

    则会打印结果

    二级叶子结点test实现

    综上,项目的基本框架已经实现了,接下来实现叶子结点的函数。
    例如,我们在src/test/test_hello.py有一个函数

    def test_helloworld(config,options):
    	print('hello world!')
    

    我们要怎么直接运行这个函数呢?
    首先,我们需要在src/test/下面定义一个__init__.py

    from pyouter.router import Router
    
    def dispatch():
        from test.test_hello import test_helloworld
    
        router=Router(
            test_hello = test_helloworld
        )
        return router
    

    然后在main.py里面修改main_dispatch()函数

    def main_dispatch():
        """ 主路由
    
        """
        from test import dispatch as test_dispatch
    
        router = Router(
            init = lambda config,options: print('server_ip:{}'.format(config['server_ip'])),
            test = test_dispatch()
        )
    
        return router
    

    接下来运行命令

    python main.py pyouter_test.test.test_hello --cluster dev
    

    则会直接运行test_helloworld函数,终端打印

    三级叶子结点server实现

    上面的test_helloworld函数只是二级的叶子结点实现,假如我们要进行一个三级的叶子结点实现可以参考以下的server实现。
    假如我们在src/data/目录下有一个data.txt文件,里面只有三行数据

    test
    server
    data
    

    我们在src/server/data/read_data.py里面有个函数read_data_txt函数

    def read_data_txt(config,options):
        data_path = './data/data.txt'
        with open(data_path) as f:
            for line in f:
                print(line + ' ')
    

    如果我们要直接运行这个函数,我们还需要分别在src/server/src/server/read_data/下面分别定义一个__init__.py
    src/server/data/下面的__init__.py实现如下:

    from pyouter.router import Router
    
    def dispatch():
        from server.data.read_data import read_data_txt
    
        router=Router(
            read_data_txt = read_data_txt
        )
        return router
    

    src/server/下面的__init__.py实现如下:

    from pyouter.router import Router
    
    def dispatch():
        from server.data import dispatch as data_dispatch
    
        router=Router(
            data = data_dispatch()
        )
        return router
    

    然后修改main.py里面的main_dispatch()函数:

    def main_dispatch():
        """ 主路由
    
        """
        from test import dispatch as test_dispatch
        from server import dispatch as server_dispatch
    
        router = Router(
            init = lambda config,options: print('server_ip:{}'.format(config['server_ip'])),
            test = test_dispatch(),
            server = server_dispatch()
        )
    
        return router
    

    接下来运行命令

    python main.py pyouter_test.server.data.read_data_txt --cluster dev
    

    终端会打印结果

    项目结构细节

    上述实现的整体项目类似于一颗树,根目录是src/main.pytest_helloworldread_data_txt函数分别是二级叶子结点和三级叶子结点,如下图所示:

    在运行命令的时候用.来隔开不同层级结点

    python main.py pyouter_test.server.data.read_data_txt
    

    在运行命令的时候可以在后面添加定制化参数如--cluster

    python main.py pyouter_test.server.data.read_data_txt --cluster dev
    

    通过options.cluster可以获得该参数的值"dev",还可以在src/options.py里面的def add_custom_arguments(parser: ArgumentParser)函数新增定制化参数,比如新增--input_file--output_file参数:

    def add_custom_arguments(parser: ArgumentParser):
        ''' 用户定制化参数选项
        
        * --cluster: 开发环境还是线上环境
        '''
    
        parser.add_argument(
            "--cluster",
            dest="cluster",
            help="cluster dev or pro",
            default="dev",
            nargs='?',
            type=str,
            metavar="CLUSTER"
        )
    
        parser.add_argument(
            "--input_file",
            dest="input_file",
            help="input_file path",
            default="",
            nargs='?',
            type=str,
            metavar="INPUT_FILE"
        )
    
        parser.add_argument(
            "--output_file",
            dest="output_file",
            help="output_file path",
            default="",
            nargs='?',
            type=str,
            metavar="OUTPUT_FILE"
        )
    

    执行以下命令的时候可以给input_fileoutput_file参数赋值

    python main.py pyouter_test.server.data.read_data_txt --cluster dev --input_file './data/data_txt' --output_file './data/res.txt'
    

    值得注意的是,无论是命令行输入的相对路径还是函数里面实现的路径都是针对/src/main.py的相对路径。

    作者:又三郎丶

    物联沃分享整理
    物联沃-IOTWORD物联网 » 使用项目框架管理工具Pyouter的实际应用

    发表回复