详细解读Python的web.py框架下的application.py模块

程序员文章站 2022-05-12 22:57:17

...

本文主要分析的是web.py库的application.py这个模块中的代码。总的来说，这个模块主要实现了WSGI兼容的接口，以便应用程序能够被WSGI应用服务器调用。WSGI是Web Server Gateway Interface的缩写，具体细节可以查看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，调用其内部定义函数process(processors)。

如果位置2判断处理器列表不为空，则进入if内部。

在位置3调用本次需要执行的处理器函数，参数为一个lambda函数，然后返回。

如果位置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的类型来做出不同的处理：