您好,登錄后才能下訂單哦!
@TOC
flask中Manager源碼如下所示:
class Manager(object):
"""
Controller class for handling a set of commands.
Typical usage::
class Print(Command):
def run(self):
print "hello"
app = Flask(__name__)
manager = Manager(app)
manager.add_command("print", Print())
if __name__ == "__main__":
manager.run()
On command line::
python manage.py print
> hello
從上面我們我可以看到, 我們定義了一個類繼承Command, 并且重寫了run方法。就可以在命令行執行了, 也就是說Command為我們提供一個接口, run方法中內容都會被命令行所執行。
那么,現在看下Command又為我們做了什么, 源碼如下所示:
class Command(object):
"""
Base class for creating commands.
:param func: Initialize this command by introspecting the function.
"""
option_list = ()
help_args = None
def __init__(self, func=None):
if func is None:
if not self.option_list:
self.option_list = []
return
args, varargs, keywords, defaults = inspect.getargspec(func)
if inspect.ismethod(func):
args = args[1:]
options = []
# first arg is always "app" : ignore
defaults = defaults or []
kwargs = dict(izip(*[reversed(l) for l in (args, defaults)]))
for arg in args:
if arg in kwargs:
default = kwargs[arg]
if isinstance(default, bool):
options.append(Option('-%s' % arg[0],
'--%s' % arg,
action="store_true",
dest=arg,
required=False,
default=default))
else:
options.append(Option('-%s' % arg[0],
'--%s' % arg,
dest=arg,
type=text_type,
required=False,
default=default))
else:
options.append(Option(arg, type=text_type))
self.run = func
self.__doc__ = func.__doc__
self.option_list = options
@property
def description(self):
description = self.__doc__ or ''
return description.strip()
def add_option(self, option):
"""
Adds Option to option list.
"""
self.option_list.append(option)
def get_options(self):
"""
By default, returns self.option_list. Override if you
need to do instance-specific configuration.
"""
return self.option_list
def create_parser(self, *args, **kwargs):
func_stack = kwargs.pop('func_stack',())
parent = kwargs.pop('parent',None)
parser = argparse.ArgumentParser(*args, add_help=False, **kwargs)
help_args = self.help_args
while help_args is None and parent is not None:
help_args = parent.help_args
parent = getattr(parent,'parent',None)
if help_args:
from flask_script import add_help
add_help(parser,help_args)
for option in self.get_options():
if isinstance(option, Group):
if option.exclusive:
group = parser.add_mutually_exclusive_group(
required=option.required,
)
else:
group = parser.add_argument_group(
title=option.title,
description=option.description,
)
for opt in option.get_options():
group.add_argument(*opt.args, **opt.kwargs)
else:
parser.add_argument(*option.args, **option.kwargs)
parser.set_defaults(func_stack=func_stack+(self,))
self.parser = parser
self.parent = parent
return parser
def __call__(self, app=None, *args, **kwargs):
"""
Handles the command with the given app.
Default behaviour is to call ``self.run`` within a test request context.
"""
with app.test_request_context():
return self.run(*args, **kwargs)
def run(self):
"""
Runs a command. This must be implemented by the subclass. Should take
arguments as configured by the Command options.
"""
raise NotImplementedError
下面分析下執行過程:
1、其他函數是對options[列表]的內容進行增刪查操作
2、create_parser函數創建了命令行解析對象parser = argparse.ArgumentParser(*args, add_help=False, **kwargs)
,獲取options中獲取并保存options中數據和help_args中數據,以及parser.set_defaults(func_stack=func_stack+(self,))
將Command自己添加到parser中參數中。
3,在flask應用代碼中我們添加例如manager.add_command("db", Print())
的代碼,傳入了Command的實例對象, 而add_command創建了Command的實例對象并保存在slef._commands的namespace中或者key_value值中。
manager.add_command("db", MigrateCommand)
的代碼,傳入了Manager的實例對象MigrateCommand- - - -另外一個Manager對象(此對象,已經添加了添加了遷移等命令, 后面會和當前這個flask應用中manager對象建立關聯self.parent)4,call方法中顯示,當Command的實例對象被調用的時候,就會被執行(此時, 引入app實例的上下文, 并執行了run方法)。那么此時,我們就只要去尋找Command實例是何時被調用的
。
run方法中通過sys.argv接收了命令行參數,并把參數提交給slef.handle
執行。
而handle方法中創建app_parser = self.create_parser(prog)
(此函數獲取到Commad對象),獲取了所有的app_parser的信息(func和 args和config)。
此時而app_parser依然是argparse中ArgumentParser
對象。
依然是在Manage的create_parser方法中,執行了app_namespace, remaining_args = app_parser.parse_known_args(args)
, 方法內又調用了_parse_known_args
此處是重點:_parse_known_args中,內部函數consume_optional先調用self._option_string_actions通過string映射action類,
另一方面調用了內部函數take_action,創建action的實例對象(如上面Command的實例對象就是此時被創建)。
app_namespace.dict中可以獲取到func_stack,
最后遍歷for handle in func_stack:
,并執行handle, handle可能是Command實例對象,調用后,執行call方法,執行run方法,也可能是func函數,直接被執行。
def run(self, commands=None, default_command=None):
"""
Prepares manager to receive command line input. Usually run
inside "if __name__ == "__main__" block in a Python script.
:param commands: optional dict of commands. Appended to any commands
added using add_command().
:param default_command: name of default command to run if no
arguments passed.
"""
if commands:
self._commands.update(commands)
# Make sure all of this is Unicode
argv = list(text_type(arg) for arg in sys.argv)
if default_command is not None and len(argv) == 1:
argv.append(default_command)
try:
result = self.handle(argv[0], argv[1:])
except SystemExit as e:
result = e.code
sys.exit(result or 0)
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。