web.py源码分析: application

Dear 丶 2022-08-10 05:49 147阅读 0赞

本文主要分析的是web.py库的application.py这个模块中的代码。总的来说，这个模块主要实现了WSGI兼容的接口，以便应用程序能够被WSGI应用服务器调用。WSGI是Web Server Gateway Interface的缩写，具体细节可以查看[WSGI的WIKI页面][WSGI_WIKI]

# 接口的使用 #

## 使用web.py自带的HTTP Server ##

下面这个例子来自官方文档的[Hello World][]，这个代码一般是应用入口的代码：

import web urls = ("/.*", "hello") app = web.application(urls, globals()) class hello: def GET(self): return 'Hello, world!' if __name__ == "__main__": app.run()

上面的例子描述了一个web.py应用最基本的组成元素：

*  URL路由表
 *  一个`web.application`实例`app`
 *  调用`app.run()`

其中，`app.run()`的调用是初始化各种WCGI接口，并启动一个内置的HTTP服务器和这些接口对接，代码如下：

def run(self, *middleware): return wsgi.runwsgi(self.wsgifunc(*middleware))

## 与WSGI应用服务器对接 ##

如果你的应用要与WSGI应用服务器对接，比如uWSGI，gunicorn等，那么应用入口的代码就要换一种写法了：

import web class hello: def GET(self): return 'Hello, world!' urls = ("/.*", "hello") app = web.application(urls, globals()) application = app.wsgifunc()

在这种场景下，应用的代码不需要启动HTTP服务器，而是实现一个WSGI兼容的接口供WSGI服务器调用。web.py框架为我们实现了这样的接口，你只需要调用`application = app.wsgifunc()`就可以了，这里所得到的`application`变量就是WSGI接口（后面分析完代码你就会知道了）。

# WSGI接口的实现分析 #

分析主要围绕着下面两行代码进行：

app = web.application(urls, globals())
    application = app.wsgifunc()

## web.application实例化 ##

初始化这个实例需要传递两个参数：URL路由元组和`globals()`的结果。

另外，还可以传递第三个变量：autoreload，用来指定是否需要自动重新导入Python模块，这在调试的时候很有用，不过我们分析主要过程的时候可以忽略。

`application`类的初始化代码如下：

class application: def __init__(self, mapping=(), fvars={}, autoreload=None): if autoreload is None: autoreload = web.config.get('debug', False) self.init_mapping(mapping) self.fvars = fvars self.processors = [] self.add_processor(loadhook(self._load)) self.add_processor(unloadhook(self._unload)) if autoreload: ...

其中，autoreload相关功能的代码略去了。其他的代码主要作了如下几个事情：

*  `self.init_mapping(mapping)`：初始化URL路由映射关系。
 *  `self.add_processor()`：添加了两个处理器。

### 初始化URL路由映射关系 ###

def init_mapping(self, mapping): self.mapping = list(utils.group(mapping, 2))

这个函数还调用了一个工具函数，效果是这样的：

urls = ("/", "Index",
            "/hello/(.*)", "Hello",
            "/world", "World")

如果用户初始化时传递的元组是这样的，那么调用`init_mapping`之后：

self.mapping = [["/", "Index"],
                    ["/hello/(.*)", "Hello"],
                    ["/world", "World"]]

后面框架在进行URL路由时，就会遍历这个列表。

### 添加处理器 ###

self.add_processor(loadhook(self._load))
        self.add_processor(unloadhook(self._unload))

这两行代码添加了两个处理器：`self._load`和`self._unload`，而且还对这两个函数进行了装饰。处理器的是用在HTTP请求处理前后的，它不是真正用来处理一个HTTP请求，但是可以用来作一些额外的工作，比如官方教程里面有提到的给子应用添加session的做法，就是使用了处理器：

def session_hook(): web.ctx.session = session app.add_processor(web.loadhook(session_hook))

处理器的定义和使用都是比较复杂的，后面专门讲。

## wsgifunc函数 ##

wsgifunc的执行结果是返回一个WSGI兼容的函数，并且该函数内部实现了URL路由等功能。

def wsgifunc(self, *middleware): """Returns a WSGI-compatible function for this application.""" ... for m in middleware: wsgi = m(wsgi) return wsgi

除开内部函数的定义，wsgifunc的定义就是这么简单，如果没有实现任何中间件，那么就是直接返回其内部定义的`wsgi`函数。

### wsgi函数 ###

该函数实现了WSGI兼容接口，同时也实现了URL路由等功能。

def wsgi(env, start_resp): # clear threadlocal to avoid inteference of previous requests self._cleanup() self.load(env) try: # allow uppercase methods only if web.ctx.method.upper() != web.ctx.method: raise web.nomethod() result = self.handle_with_processors() if is_generator(result): result = peep(result) else: result = [result] except web.HTTPError, e: result = [e.data] result = web.safestr(iter(result)) status, headers = web.ctx.status, web.ctx.headers start_resp(status, headers) def cleanup(): self._cleanup() yield '' # force this function to be a generator return itertools.chain(result, cleanup()) for m in middleware: wsgi = m(wsgi) return wsgi

下面来仔细分析一下这个函数：

self._cleanup()
        self.load(env)

`self._cleanup()`内部调用`utils.ThreadedDict.clear_all()`，清除所有的thread local数据，避免内存泄露（因为web.py框架的很多数据都会保存在thread local变量中）。

`self.load(env)`使用`env`中的参数初始化`web.ctx`变量，这些变量涵盖了当前请求的信息，我们在应用中有可能会使用到，比如`web.ctx.fullpath`。

try: # allow uppercase methods only if web.ctx.method.upper() != web.ctx.method: raise web.nomethod() result = self.handle_with_processors() if is_generator(result): result = peep(result) else: result = [result] except web.HTTPError, e: result = [e.data]

这一段主要是调用`self.handle_with_processors()`，这个函数会对请求的URL进行路由，找到合适的类或子应用来处理该请求，也会调用添加的处理器来做一些其他工作（关于处理器的部分，后面专门讲）。对于处理的返回结果，可能有三种方式：

*  返回一个可迭代对象，则进行安全迭代处理。
 *  返回其他值，则创建一个列表对象来存放。
 *  如果抛出了一个HTTPError异常（比如我们使用raise web.OK("hello, world")这种方式来返回结果时），则将异常中的数据`e.data`封装成一个列表。

result = web.safestr(iter(result)) status, headers = web.ctx.status, web.ctx.headers start_resp(status, headers) def cleanup(): self._cleanup() yield '' # force this function to be a generator return itertools.chain(result, cleanup())

接下来的这段代码，会对前面返回的列表`result`进行字符串化处理，得到HTTP Response的body部分。然后根据WSGI的规范作如下两个事情：

*  调用`start_resp`函数。
 *  将`result`结果转换成一个迭代器。

现在你可以看到，之前我们提到的`application = app.wsgifunc()`就是将`wsgi`函数赋值给`application`变量，这样应用服务器就可以采用WSGI标准和我们的应用对接了。

## 处理HTTP请求 ##

前面分析的代码已经说明了web.py框架如何实现WSGI兼容接口的，即我们已经知道了HTTP请求到达框架以及从框架返回给应用服务器的流程。那么框架内部是如何调用我们的应用代码来实现一个请求的处理的呢？这个就需要详细分析刚才忽略掉的处理器的添加和调用过程。

### loadhook和unloadhook装饰器 ###

这两个函数是真实处理器的函数的装饰器函数（虽然他的使用不是采用装饰器的@操作符），装饰后得到的处理器分别对应请求处理之前（loadhook）和请求处理之后（unloadhook）。

#### loadhook ####

def loadhook(h): def processor(handler): h() return handler() return processor

这个函数返回一个函数`processor`，它会确保先调用你提供的处理器函数`h`，然后再调用后续的操作函数`handler`。

#### unloadhook ####

def unloadhook(h): def processor(handler): try: result = handler() is_generator = result and hasattr(result, 'next') except: # run the hook even when handler raises some exception h() raise if is_generator: return wrap(result) else: h() return result def wrap(result): def next(): try: return result.next() except: # call the hook at the and of iterator h() raise result = iter(result) while True: yield next() return processor

这个函数也返回一个`processor`，它会先调用参数传递进来的`handler`，然后再调用你提供的处理器函数。

### handle\_with\_processors函数 ###

def handle_with_processors(self): def process(processors): try: if processors: p, processors = processors[0], processors[1:] return p(lambda: process(processors)) else: return self.handle() except web.HTTPError: raise except (KeyboardInterrupt, SystemExit): raise except: print >> web.debug, traceback.format_exc() raise self.internalerror() # processors must be applied in the resvere order. (??) return process(self.processors)

这个函数挺复杂的，最核心的部分采用了递归实现（我感觉不递归应该也能实现同样的功能）。为了说明清晰，采用实例说明。

前面有提到，初始化`application`实例的时候，会添加两个处理器到`self.processors`：

self.add_processor(loadhook(self._load))
        self.add_processor(unloadhook(self._unload))

所以，现在的`self.processors`是下面这个样子的：

self.processors = [loadhook(self._load), unloadhook(self._unload)]
    # 为了方便后续说明，我们缩写一下：
    self.processors = [load_processor, unload_processor]

当框架开始执行`handle_with_processors`的时候，是逐个执行这些处理器的。我们还是来看代码分解，首先简化一下`handle_with_processors`函数：

def handle_with_processors(self): def process(processors): try: if processors: # 位置2 p, processors = processors[0], processors[1:] return p(lambda: process(processors)) # 位置3 else: return self.handle() # 位置4 except web.HTTPError: raise ... # processors must be applied in the resvere order. (??) return process(self.processors) # 位置1

1.  函数执行的起点是位置1，调用其内部定义函数`process(processors)`。
2.  如果位置2判断处理器列表不为空，则进入`if`内部。
3.  在位置3调用本次需要执行的处理器函数，参数为一个lambda函数，然后返回。
4.  如果位置2判断处理器列表为空，则执行`self.handle()`，该函数真正的调用我们的应用代码（下面会讲到）。

以上面的例子来说，目前有两个处理器：

self.processors = [load_processor, unload_processor]

从位置1进入代码后，在位置2会判断还有处理器要执行，会走到位置3，此时要执行代码是这样的：

return load_processor(lambda: process([unload_processor]))

`load_processor`函数是一个经过`loadhook`装饰的函数，因此其定义在执行时是这样的：

def load_processor(lambda: process([unload_processor])): self._load() return process([unload_processor]) # 就是参数的lambda函数

会先执行`self._load()`，然后再继续执行`process`函数，依旧会走到位置3，此时要执行的代码是这样的：

return unload_processor(lambda: process([]))

`unload_processor`函数是一个经过`unloadhook`装饰的函数，因此其定义在执行时是这样的：

def unload_processor(lambda: process([])): try: result = process([]) # 参数传递进来的lambda函数 is_generator = result and hasattr(result, 'next') except: # run the hook even when handler raises some exception self._unload() raise if is_generator: return wrap(result) else: self._unload() return result

现在会先执行`process([])`函数，并且走到位置4（调用`self.handle()`的地方），从而得到应用的处理结果，然后再调用本处理器的处理函数`self._unload()`。

总结一下执行的顺序：

*  `self._load()`
    
     *  `self.handle()`
 *  `self._unload()`

如果还有更多的处理器，也是按照这种方法执行下去，对于`loadhook`装饰的处理器，先添加的先执行，对于`unloadhook`装饰的处理器，后添加的先执行。

### handle函数 ###

讲了这么多，才讲到真正要调用我们写的代码的地方。在所有的load处理器执行完之后，就会执行`self.handle()`函数，其内部会调用我们写的应用代码。比如返回个hello, world之类的。`self.handle`的定义如下：

def handle(self): fn, args = self._match(self.mapping, web.ctx.path) return self._delegate(fn, self.fvars, args)

这个函数就很好理解了，第一行调用的`self._match`是进行路由功能，找到对应的类或者子应用，第二行的`self._delegate`就是调用这个类或者传递请求到子应用。

#### \_match函数 ####

`_match`函数的定义如下：

def _match(self, mapping, value): for pat, what in mapping: if isinstance(what, application): # 位置1 if value.startswith(pat): f = lambda: self._delegate_sub_application(pat, what) return f, None else: continue elif isinstance(what, basestring): # 位置2 what, result = utils.re_subm('^' + pat + '$', what, value) else: # 位置3 result = utils.re_compile('^' + pat + '$').match(value) if result: # it's a match return what, [x for x in result.groups()] return None, None

该函数的参数中`mapping`就是`self.mapping`，是URL路由映射表；`value`则是`web.ctx.path`，是本次请求路径。该函数遍历`self.mapping`，根据映射关系中处理对象的类型来处理：

*  位置1，处理对象是一个`application`实例，也就是一个子应用，则返回一个匿名函数，该匿名函数会调用`self._delegate_sub_application`进行处理。
 *  位置2，如果处理对象是一个字符串，则调用`utils.re_subm`进行处理，这里会把`value`（也就是`web.ctx.path`）中的和`pat`匹配的部分替换成`what`（也就是我们指定的一个URL模式的处理对象字符串），然后返回替换后的结果以及匹配的项（是一个re.MatchObject实例）。
 *  位置3，如果是其他情况，比如直接指定一个类对象作为处理对象。

如果`result`非空，则返回处理对象和一个参数列表（这个参数列表就是传递给我们实现的GET等函数的参数）。

#### \_delegate函数 ####

从`_match`函数返回的结果会作为参数传递给`_delegate`函数：

fn, args = self._match(self.mapping, web.ctx.path)
    return self._delegate(fn, self.fvars, args)

其中：

*  `fn`：是要处理当前请求的对象，一般是一个类名。
 *  `args`：是要传递给请求处理对象的参数。
 *  `self.fvars`：是实例化`application`时的全局名称空间，会用于查找处理对象。

`_delegate`函数的实现如下：

def _delegate(self, f, fvars, args=[]): def handle_class(cls): meth = web.ctx.method if meth == 'HEAD' and not hasattr(cls, meth): meth = 'GET' if not hasattr(cls, meth): raise web.nomethod(cls) tocall = getattr(cls(), meth) return tocall(*args) def is_class(o): return isinstance(o, (types.ClassType, type)) if f is None: raise web.notfound() elif isinstance(f, application): return f.handle_with_processors() elif is_class(f): return handle_class(f) elif isinstance(f, basestring): if f.startswith('redirect '): url = f.split(' ', 1)[1] if web.ctx.method == "GET": x = web.ctx.env.get('QUERY_STRING', '') if x: url += '?' + x raise web.redirect(url) elif '.' in f: mod, cls = f.rsplit('.', 1) mod = __import__(mod, None, None, ['']) cls = getattr(mod, cls) else: cls = fvars[f] return handle_class(cls) elif hasattr(f, '__call__'): return f() else: return web.notfound()

这个函数主要是根据参数`f`的类型来做出不同的处理：

*  `f`为空，则返回302 Not Found.
 *  `f`是一个`application`实例，则调用子应用的`handle_with_processors()`进行处理。
 *  `f`是一个类对象，则调用内部函数`handle_class`。
 *  `f`是一个字符串，则进行重定向处理，或者获取要处理请求的类名后，调用`handle_class`进行处理（我们写的代码一般是在这个分支下被调用的）。
 *  `f`是一个可调用对象，直接调用。
 *  其他情况返回302 Not Found.

本文由 [diabloneo][] 创作，http://segmentfault.com/a/1190000002586349

[WSGI_WIKI]: http://en.wikipedia.org/wiki/Web_Server_Gateway_Interface
[Hello World]: http://webpy.org/cookbook/helloworld
[diabloneo]: http://segmentfault.com/u/diabloneo