pecan源码阅读(二)

    xiaoxiao2021-08-21  221

    config文件

    新建pecan工程的默认config.py文件(上一篇博文讲过,该文件是从scaffolds中的模板copy的)

    // 启动的默认的host和port server = { 'port': '8080', 'host': '0.0.0.0' } // app的配置 app = { // root项很重要,指定开始路由的处理器 'root': 'pecanstudy.controllers.root.RootController', 'modules': ['pecanstudy'], 'static_root': '%(confdir)s/public', 'template_path': '%(confdir)s/pecanstudy/templates', 'debug': True, 'errors': { // abort(404)时会转发到该路由 404: '/error/404', '__force_dict__': True } } // python格式的日志配置 logging = { 'root': {'level': 'INFO', 'handlers': ['console']}, 'loggers': { 'pecanstudy': {'level': 'DEBUG', 'handlers': ['console'], 'propagate': False}, 'pecan': {'level': 'DEBUG', 'handlers': ['console'], 'propagate': False}, 'py.warnings': {'handlers': ['console']}, '__force_dict__': True }, 'handlers': { 'console': { 'level': 'DEBUG', 'class': 'logging.StreamHandler', 'formatter': 'color' } }, 'formatters': { 'simple': { 'format': ('%(asctime)s %(levelname)-5.5s [%(name)s]' '[%(threadName)s] %(message)s') }, 'color': { '()': 'pecan.log.ColorFormatter', 'format': ('%(asctime)s [%(padded_color_levelname)s] [%(name)s]' '[%(threadName)s] %(message)s'), '__force_dict__': True } } }

    路由分析

    app是一个Pecan实例,先来看看初始化Pecan实例时做了哪些工作

    // core.py文件 class Pecan(PecanBase): def __new__(cls, *args, **kw): if kw.get('use_context_locals') is False: self = super(Pecan, cls).__new__(ExplicitPecan, *args, **kw) self.__init__(*args, **kw) return self return super(Pecan, cls).__new__(cls) def __init__(self, *args, **kw): self.init_context_local(kw.get('context_local_factory')) //调用了PecanBase的初始化方法 super(Pecan, self).__init__(*args, **kw) ...... ...... class PecanBase(object): def __init__(self, root, default_renderer='mako', template_path='templates', hooks=lambda: [], custom_renderers={}, extra_template_vars={}, force_canonical=True, guess_content_type_from_ext=True, context_local_factory=None, request_cls=Request, response_cls=Response, **kw): // 这里将配置文件中的root配置项导入为一个python对象 if isinstance(root, six.string_types): root = self.__translate_root__(root) self.root = root // url中的路径,比如:/v1/books self.request_cls = request_cls // webob的Request类,所以pecan以来WebOb self.response_cls = response_cls //webob的Response类 self.renderers = RendererFactory(custom_renderers, extra_template_vars) // 渲染器 self.default_renderer = default_renderer // 初始化钩子程序 if six.callable(hooks): hooks = hooks() self.hooks = list(sorted( hooks, key=operator.attrgetter('priority') )) self.template_path = template_path self.force_canonical = force_canonical self.guess_content_type_from_ext = guess_content_type_from_ext

    在core.py中定义了一个全局变量state,它的生命周期和整个请求的生命周期一致,保存了请求过程中各种参数状态值 state.hooks:存储钩子程序的list state.request:请求对象 state.response:响应对象 state.controller:处理器(程序中被expose装饰的方法) state.arguments:参数 state.app:Peacn对象

    当一个请求从wsgiserver转发过来,首先处理的是Pecan中的__call__方法

    // core.py文件 class Pecan(PecanBase): def __call__(self, environ, start_response): try: state.hooks = [] state.app = self state.controller = None state.arguments = None // 调用了PecanBase的__call__方法 return super(Pecan, self).__call__(environ, start_response) finally: del state.hooks del state.request del state.response del state.controller del state.arguments del state.app class PecanBase(object): def __call__(self, environ, start_response): // WebOb的Request和Response req = self.request_cls(environ) resp = self.response_cls() state = RoutingState(req, resp, self) environ['pecan.locals'] = { 'request': req, 'response': resp } controller = None internal_redirect = False try: req.context = environ.get('pecan.recursive.context', {}) req.pecan = dict(content_type=None) // 路由方法,对象分发路由机制,传入state,记录整个过程中的状态 controller, args, kwargs = self.find_controller(state) // 调用处理方法 self.invoke_controller(controller, args, kwargs, state) except Exception as e: ...... ...... self._handle_empty_response_body(state) // 返回结果 return state.response(environ, start_response)

    主要调用了find_controller和invoke_controller方法。find_controller根据对象分发机制找到url的处理方法,如果没找到,则抛出异常,由后面的except代码块处理,找到了就调用invoke_controller执行该处理方法,将处理结果保存到state中。

    class PecanBase(object): def find_controller(self, state): req = state.request pecan_state = req.pecan pecan_state['routing_path'] = path = req.path_info // 处理钩子程序 self.handle_hooks(self.hooks, 'on_route', state) ...... ...... // 具体路由方法 controller, remainder = self.route(req, self.root, path) ...... ...... // 根据路由结果处理参数,如果没有路由到,则该方法会抛出异常 args, varargs, kwargs = self.get_args( state, params.mixed(), remainder, cfg['argspec'], im_self ) state.arguments = Arguments(args, varargs, kwargs) // 处理钩子程序 self.handle_hooks(self.determine_hooks(controller), 'before', state) return controller, args + varargs, kwargs

    钩子程序分为4种,路由前(on_route),路由后处理前(before),处理后(after),发生错误(on_error),钩子程序可在app.py中自定义,需要继承PecanHook类(在hooks.py中定义)

    route(req, self.root, path): req:WebOb的Request对象,存储了请求的信息 self.root:是第一个处理对象(config.py中定义的root对象) path:路径信息,如:/v1/books

    def route(self, req, node, path): path = path.split('/')[1:] // 转化后的路径 path:['v1','boos'] try: // 调用了routing.py文件中的lookup_controller方法 node, remainder = lookup_controller(node, path, req) return node, remainder except NonCanonicalPath as e: if self.force_canonical and \ not _cfg(e.controller).get('accept_noncanonical', False): if req.method == 'POST': raise RuntimeError( "You have POSTed to a URL '%s' which " "requires a slash. Most browsers will not maintain " "POST data when redirected. Please update your code " "to POST to '%s/' or set force_canonical to False" % (req.pecan['routing_path'], req.pecan['routing_path']) ) redirect(code=302, add_slash=True, request=req) return e.controller, e.remainder // routing.py文件 def lookup_controller(obj, remainder, request=None): ...... ...... // 存储在obj中未找到处理方法时的_default,_lookup notfound_handlers = [] while True: try: obj, remainder = find_object(obj, remainder, notfound_handlers, request) handle_security(obj) return obj, remainder except (exc.HTTPNotFound, exc.HTTPMethodNotAllowed, PecanNotFound) as e: if isinstance(e, PecanNotFound): e = exc.HTTPNotFound() while notfound_handlers: name, obj, remainder = notfound_handlers.pop() if name == '_default': return obj, remainder else: result = handle_lookup_traversal(obj, remainder) if result: if ( remainder == [''] and len(obj._pecan['argspec'].args) > 1 ): raise e obj_, remainder_ = result return lookup_controller(obj_, remainder_, request) else: raise e

    lookup_controller针对每一个controller对象,在其中查找对应的处理方法,如果没找到,则会继续找_default,如果没定义_default,则找_lookup,然后继续循环调用lookup_controller,直到找到对应的方法,或notfound_handlers 为空抛出异常

    obj, remainder = find_object(obj, remainder, notfound_handlers, request)具体查找: obj:当前的controller对象 remainder:路由信息,如[‘v1’, ‘books’] notfound_handlers:该controller中没找到时,存储_default或者_lookup request:请求信息

    def find_object(obj, remainder, notfound_handlers, request): prev_obj = None while True: if obj is None: raise PecanNotFound // 如果传入的obj直接时一个处理方法(被expsoe装饰),直接返回 if iscontroller(obj): if getattr(obj, 'custom_route', None) is None: return obj, remainder // 处理自定义路由信息 _detect_custom_path_segments(obj) // 根据自定义路由名找到处理方法 if remainder: custom_route = __custom_routes__.get((obj.__class__, remainder[0])) if custom_route: return getattr(obj, custom_route), remainder[1:] cross_boundary(prev_obj, obj) // 如果根据默认和自定义路由都没找到,则找该controller中的index方法 // 如果有路由:/v1/books//best(不标准的路径),那么就只能路由到/v1/books,后面的就没法路由 try: next_obj, rest = remainder[0], remainder[1:] if next_obj == '': index = getattr(obj, 'index', None) if iscontroller(index): return index, rest except IndexError: index = getattr(obj, 'index', None) if iscontroller(index): raise NonCanonicalPath(index, []) // 存储_default方法到notfound_handlers default = getattr(obj, '_default', None) if iscontroller(default): notfound_handlers.append(('_default', default, remainder)) // 则存储_lookup方法到notfound_handlers lookup = getattr(obj, '_lookup', None) if iscontroller(lookup): notfound_handlers.append(('_lookup', lookup, remainder)) // 根据自定义的_route方法来处理路由(pecan允许开发者在controller中自定义_route方法,让开发者完全掌控路由方式) route = getattr(obj, '_route', None) if iscontroller(route): if len(getargspec(route).args) == 2: warnings.warn( ( "The function signature for %s.%s._route is changing " "in the next version of pecan.\nPlease update to: " "`def _route(self, args, request)`." % ( obj.__class__.__module__, obj.__class__.__name__ ) ), DeprecationWarning ) next_obj, next_remainder = route(remainder) else: next_obj, next_remainder = route(remainder, request) cross_boundary(route, next_obj) return next_obj, next_remainder if not remainder: raise PecanNotFound prev_remainder = remainder prev_obj = obj remainder = rest // 根据方法名(或者属性名)(默认的路由方式)找到处理方法 try: obj = getattr(obj, next_obj, None) except UnicodeEncodeError: obj = None if not obj and not notfound_handlers and hasattr(prev_obj, 'index'): if request.method in _cfg(prev_obj.index).get('generic_handlers', {}): return prev_obj.index, prev_remainder

    find_object 首先会处理自定义的路由信息,然后存储_default和_lookup,最后处理默认路由(个人觉得可以先处理默认路由信息,然后根据是否配置route装饰进行取舍,这样可能处理更高效)

    routing.py中的lookup_controller 和 find_object是核心路由方式的实现,从代码中可以看出,最终找到处理方法的方式是根据路径(/v1/books)中每一个segment来查找对应的对象,然后根据当前对象再查找下一个对象,所以pecan的路由机制叫做对象分发

    下一篇将学习pecan的异常抛出,根据方法路由等内容


    最新回复(0)