python开发总结-迁
gunicorn是一个python Wsgi http server,只支持在Unix系统上运行,来源于Ruby的unicorn项目。Gunicorn使用prefork master-worker模型(在gunicorn中,master被称为arbiter),能够与各种wsgi web框架协作。gunicorn的文档是比较完善的,这里也有部分中文翻译,不过还是建议直接读英文文档。
gunicorn的安装非常简单,pip install guncorn即可。后续如果用到异步的worker模型,还需要安装对应的模块(如gevent) 在装好gunicorn之后, 我们来看看gunicorn的hello world。代码来自官网,将下面的代码放到gunicorn_app.py中:def app(environ, start_response): data = b"Hello, World!\n" start_response("200 OK", [ ("Content-Type", "text/plain"), ("Content-Length", str(len(data))) ]) return iter([data])
可以看到app是非常标准的wsgi应用,然后我们启动gunicorn:gunicorn-w2gunicorn_app:app。 输出如下:
第一:启动了两个worker,这是通过"-w 2"指定(默认为1)
第二:worker的工作模型是sync(默认),后面会详细介绍worker模型 可以看出 worker进程(pid:19469, 19470)是master进程(pid:19464)的子进程。 新起一个terminal,用curl测试:curl127.0.0.1:8000 在该terminal输出“Hello,World!”# 目录 route.py plugins/cmd.py plugins/state.py # route.py from plugins import cmd action_list = { 'cmd': cmd.CMD, } module_obj = action_list['cmd'] print("module_obj", module_obj) # cmd.py class CMD: def __init__(self): super(CMD).__init__() print("cmd=====>") # 执行 ('module_obj', <class plugins.cmd.CMD at 0x7fef31f5bbb0>) loading from cmd.py
使用__import__的方法
# 平台端 import os class Platform(object): def __init__(self): self.loadPlugins() def sayHello(self, from_): print "hello from %s." % from_ def loadPlugins(self): for filename in os.listdir("plugins"): if not filename.endswith(".py") or filename.startswith("_"): continue print("loading plugins from %s" % filename) self.runPlugin(filename) def runPlugin(self, filename): pluginName=os.path.splitext(filename)[0] plugin=__import__("plugins."+pluginName, fromlist=[pluginName]) #Errors may be occured. Handle it yourself. plugin.run(self) if __name__=="__main__": platform=Platform() # cmd.py插件 class CMD(object): def __init__(self): super(CMD, self).__init__() print("cmd=====>") def run(platform): c = CMD() # 执行plugin类的方法 platform.sayHello("plugin1") # 执行 loading plugins from cmd.py cmd=====> hello from plugin1.
使用装饰器的方式:
# platform.py class TextProcessor(object): PLUGINS = {} def process(self, text, plugins=()): # print self.PLUGINS,"---------" if plugins is (): for plugin_name in self.PLUGINS.keys(): text = self.PLUGINS[plugin_name]().run(text) else: for plugin_name in plugins: text = self.PLUGINS[plugin_name]().run(text) return text @classmethod def plugin_register(cls, plugin_name): def wrapper(plugin): cls.PLUGINS.update({plugin_name:plugin}) return plugin return wrapper # route.py from platform import TextProcessor def test(): processor = TextProcessor() print(processor.PLUGINS) # {’plugin1': <class '__main__.CleanMarkdownBolds'>} print processor.process(text="**foo bar**", plugins=('plugin1',)) print processor.process(text="**foo bar**") print processor.process(text="**foo bar**", plugins=('plugin2',)) # run.py from my_route.route import test test() # plugins/__init__.py 导入插件文件 __all__ = ['plugin1'] # my_route/__init__.py from .plugins import * # plugin1.py from ..platform import TextProcessor # 插件 @TextProcessor.plugin_register('plugin1') class CleanMarkdownBolds1(object): def run(self, text): return text.replace('**', '') # 插件 @TextProcessor.plugin_register('plugin2') class CleanMarkdownBolds2(object): def run(self, text): return text.replace('**', '========')
app/__init__.py
与app/plugins/__init__.py
两个文件起了相互呼应的作用app/__init__.py
,不需要在别的任何地方显式地导入插件:from .plugins import *
或from .plugins import plugin1
app/plugins/__init__.py
文件为__all__ = ['plugin1', 'plugin2', 'plugin3']
执行
{'plugin1': <class 'my_route.plugins.plugin1.CleanMarkdownBolds1'>, 'plugin2': <class 'my_route.plugins.plugin1.CleanMarkdownBolds2'>} foo bar foo bar ========foo bar========
类 |
UhasApi - 驼峰法命名 BaseModule - 基类 AbsClsUser - 抽象类 instr_var - 成员变量 Entity - 类实体名称 _mod_or_protected_var - 模块变量或者protected变量 __private_var - 私有变量,以双下划线开头 |
函数 |
syntax - 语意 parser - 解析 pkg - 包 module - 模块 build - 组装组成 perform - 执行、工作,比如perform_create |
变量 |
special - 指定的 provided - 提供的 context - 内容的 ret - 结果 msg - 消息 logger - 日志 GLOBAL_VAR - 全局常量,大写字母,下划线分割 |
模型字段 |
create_at - 创建时间 create_by - 创建人 |
用户消息 | |