python 模版引擎 mako 源码阅读

Mako 是用Python编写的模板引擎。从概念上讲，mako是一种嵌入式Python（即Python Server Page）语言，模版被编译成Python代码，使用python解释器执行。mako用于外网热门网站 reddit.com ，同时也是Pylons和Pyramid Web框架默认模板语言。学习mako，可以帮助我们加深对python编译和执行的理解。本文包括下面几个部分:

• 基础知识

• 抽象语法树 AST

• 动态编译 compile

• mako 项目结构

• Template API 介绍

• 模版解析

• 模版编译

• 模版渲染

• 小结

• 小技巧

基础知识

mako 模版引擎使用了一些编译原理相关的知识：语法解析，代码生成和编译执行等，了解这些基础知识才可以更好的读懂 mako 代码。

抽象语法树 AST

在计算机科学中，抽象语法树（Abstract Syntax Tree，AST），或简称语法树（Syntax tree），是源代码语法结构的一种抽象表示。它以树状的形式表现编程语言的语法结构，树上的每个节点都表示源代码中的一种结构。之所以说语法是“抽象”的，是因为这里的语法并不会表示出真实语法中出现的每个细节。

python标准库中 ast 模块可以帮助生成AST:

import ast

expr = """x,y =1,2\nprint(x+y)"""
ast_tree = ast.parse(expr)
print(type(ast_tree))
print(ast.dump(ast_tree))  # python 3.9 可以很好的格式化输出ast

生成的AST结构如下:

Module(
 body=[
  Assign(
   targets=[Tuple(elts=[Name(id='x', ctx=Store()), Name(id='y', ctx=Store())], ctx=Store())], 
   value=Tuple(elts=[Constant(value=1, kind=None), Constant(value=2, kind=None)], ctx=Load()), 
   type_comment=None), 
  Expr(
   value=Call(func=Name(id='print', ctx=Load()), args=[BinOp(left=Name(id='x', ctx=Load()), op=Add(), right=Name(id='y', ctx=Load()))], keywords=[])
  )
 ], 
 type_ignores=[]
)

使用参考链接中的可视化工具，这颗AST大概长这样：

ast和python版本有关，图中内容是python2版本，所以和日志中的内容略有不同

动态编译 compile

python支持对文本进行动态编译执行，请看下面示例:

import types
source = """def add(x,y):\n    return x+y"""  # 函数定义的表达式
mod = types.ModuleType("test")  # 动态创建模块
print(mod, type(mod))
code = compile(source, "test", "exec")  # 编译表达式
print(code, type(code))
exec(code, mod.__dict__, mod.__dict__)  # 动态执行代码
result = mod.add(1, 2)  # 调用动态生成的add函数
print(result, type(result))

执行日志:

<module 'test'> <class 'module'>
<code object <module> at 0x7ff3b02649d0, file "test", line 1> <class 'code'>
3 <class 'int'>

结合日志可以知道, 动态编译执行主要是下面3步:

• 使用 types.ModuleType 创建模块
• 使用 compile 编译表达式链接到模块
• 使用 exec 执行编译后的byte-code

理解上面代码后，就可以知道模版引擎的工作就是将模版的代码解析转换成python文本，再动态的编译执行。

mako 项目结构

本文选择的mako源码版本是 1.1.0，不算ext扩展包，接近7000行代码，代码量比较大，相对也复杂一些，会有一点点挑战。源码目录如下:

Template API

按照惯例，从API示例作为入口进入项目:

mytemplate = Template("hello, ${name}!")  # 1. 创建模版对象
print(mytemplate._code)
result=mytemplate.render(name="shawn")  # 2. 渲染模版对象
print(result)

Template的构造函数不算复杂，摘剪示例相关代码如下:

class Template(object):
    lexer_cls = Lexer  # 词法分析类
    def __init__(
        self,
        text=None,  # 模版文本
        filename=None,
        uri=None,
        ...
    ):
    if uri:
        ...
    else:
        self.module_id = "memory:" + hex(id(self))
        self.uri = self.module_id
    if text is not None:
        (code, module) = _compile_text(self, text, filename)  # 编译模版文本
        self._code = code
        self._source = text
    self.module = module
    self.callable_ = self.module.render_body  # 注意 render_body 函数

模版文本编译主要过程是这样的:

def _compile(template, text, filename, generate_magic_comment):
    lexer = template.lexer_cls(
        text,
        filename,
        disable_unicode=template.disable_unicode,
        input_encoding=template.input_encoding,
        preprocessor=template.preprocessor,
    )
    node = lexer.parse()  # 词法解析出节点
    source = codegen.compile( 
        node,
        template.uri,
        filename,
        ...
    )  # 生成python代码
    return source, lexer

def _compile_text(template, text, filename):
    identifier = template.module_id
    source, lexer = _compile(
        template,
        text,
        filename,
        generate_magic_comment=template.disable_unicode,
    )

    cid = identifier
    module = types.ModuleType(cid)
    code = compile(source, cid, "exec")  # 编译byte-code
    exec(code, module.__dict__, module.__dict__)  # 执行后，动态创建函数链接到module
    return (source, module)

1. 使用词法解析器将模版解析成node
2. 使用代码生成器将node转换成python的源码
3. 使用compile将源码编译成byte-code
4. 使用exec执行byte-code

模版解析

词法分析可以这样使用:

from mako.lexer import Lexer
lexer = Lexer("hello, ${name}!")
node = lexer.parse()
print(lexer.template)

日志显示模版解析后得到2个 Text 节点和1个 Expression 节点，每个节点包括节点类型，名称及在模版文件中的行列位置:

TemplateNode({}, [Text('hello, ', (1, 1)), Expression('name', [], (1, 8)), Text('!', (1, 15))])

Lexer构造函数:

class Lexer(object):
    def __init__(
        self,
        text,
        filename=None,
        disable_unicode=False,
        input_encoding=None,
        preprocessor=None,
    ):
        self.text = text  # 文本
        self.filename = filename
        self.template = parsetree.TemplateNode(self.filename) # 模版节点
        self.matched_lineno = 1
        self.matched_charpos = 0
        self.lineno = 1
        self.match_position = 0
        self.tag = []  # 节点
        self.control_line = []
        self.ternary_stack = []
        self.disable_unicode = disable_unicode
        self.encoding = input_encoding
        ...

对模版文本进行解析:

def parse(self):

    self.textlength = len(self.text)

    while True:  # 解析模版文件
        if self.match_position > self.textlength:
            break
        if self.match_end():
            break
        if self.match_expression():  # 匹配表达式
            continue
        if self.match_control_line(): # 匹配控制语句
            continue
        if self.match_comment(): # 匹配注释
            continue
        if self.match_tag_start(): # 标签起点
            continue
        if self.match_tag_end(): # 标签终点
            continue
        if self.match_python_block(): # 匹配代码块
            continue
        if self.match_text():  # 匹配文本
            continue

    return self.template

配合下面的模版文件更容易理解parse过程:

<%inherit file="base.html"/>
<%
    rows = [[v for v in range(0,10)] for row in range(0,10)]
%>
<table>
    % for row in rows: 
        ${makerow(row)}  # expression
    % endfor
</table>

<%def name="makerow(row)">
    <tr>
    % for name in row:
        <td>${name}</td>
    % endfor
    </tr>
</%def>

第一个示例中得到2个Txt和1个expression，其中解析文本函数全文:

def match_text(self):
    match = self.match(
        r"""
            (.*?)         # anything, followed by:
            (
             (?<=\n)(?=[ \t]*(?=%|\#\#)) # an eval or line-based
                                         # comment preceded by a
                                         # consumed newline and whitespace
             |
             (?=\${)      # an expression
             |
             (?=</?[%&])  # a substitution or block or call start or end
                          # - don't consume
             |
             (\\\r?\n)    # an escaped newline  - throw away
             |
             \Z           # end of string
            )""",
        re.X | re.S,
    )

    if match:
        text = match.group(1)  # 只解析一个
        if text:
            self.append_node(parsetree.Text, text)  #  生成Text节点
        return True
    else:
        return False

match是通用的解析函数，主要是正则查找目标对象，并且进行游标移位操作:

def match(self, regexp, flags=None):
    mp = self.match_position

    match = reg.match(self.text, self.match_position) # 解析
    if match:
        (start, end) = match.span()
        if end == start:
            self.match_position = end + 1 # 移位
        else:
            self.match_position = end
        ...
    return match

解析表达式全文:

def match_expression(self):
    match = self.match(r"\${")
    if match:
        line, pos = self.matched_lineno, self.matched_charpos
        text, end = self.parse_until_text(True, r"\|", r"}")
        if end == "|":
            escapes, end = self.parse_until_text(True, r"}")
        else:
            escapes = ""
        text = text.replace("\r\n", "\n")
        self.append_node(
            parsetree.Expression,  
            text,
            escapes.strip(),
            lineno=line,
            pos=pos,
        ) # 生成Expression节点
        return True
    else:
        return False

添加节点函数，比较复杂，主要逻辑是生成各种Node对象：

def append_node(self, nodecls, *args, **kwargs):
    kwargs.setdefault("source", self.text)
    kwargs.setdefault("lineno", self.matched_lineno)
    kwargs.setdefault("pos", self.matched_charpos)
    kwargs["filename"] = self.filename
    node = nodecls(*args, **kwargs)  # 生成节点对象
    if len(self.tag):  # tag是互相匹配的
        self.tag[-1].nodes.append(node)
    else:
        self.template.nodes.append(node)
    if self.control_line:  # 条件语句
        ...
    if isinstance(node, parsetree.Tag): # tag
        ...
    elif isinstance(node, parsetree.ControlLine): 
        ...

Node节点根类:

class Node(object):

    def __init__(self, source, lineno, pos, filename):
        self.source = source  # 源码
        self.lineno = lineno  # 行
        self.pos = pos # 列
        self.filename = filename

    def get_children(self):
        return []

    def accept_visitor(self, visitor):
        def traverse(node):  # 递归节点
            for n in node.get_children():
                n.accept_visitor(visitor)

        method = getattr(visitor, "visit" + self.__class__.__name__, traverse)  # 动态函数
        method(self)

class TemplateNode(Node):

    def __init__(self, filename):
        super(TemplateNode, self).__init__("", 0, 0, filename)
        self.nodes = []  # 所有节点

    def get_children(self):
        return self.nodes
    ...

节点类型清单:

• ControlLine
• Text
• Code
• Comment
• Expression
• Tag
• IncludeTag
• NamespaceTag
• TextTag
• DefTag
• BlockTag
• CallTag
• CallNamespaceTag
• InheritTag
• PageTag

我们选择文本节点和表达式节点进行学习。文本节点非常简单，仅存储文本内容:

class Text(Node):

    def __init__(self, content, **kwargs):
        super(Text, self).__init__(**kwargs)
        self.content = content
    ...

表达式节点相对复杂一些:

class Expression(Node):

    def __init__(self, text, escapes, **kwargs):
        super(Expression, self).__init__(**kwargs)
        self.text = text  # 文本
        ...
        self.code = ast.PythonCode(text, **self.exception_kwargs)  # 代码片段

python代码片段:

class PythonCode(object):

    def __init__(self, code, **exception_kwargs):
        self.code = code

         # represents all identifiers which are assigned to at some point in
        # the code
        self.declared_identifiers = set()  # 重要的集合，没想好中文，请看英文

        # represents all identifiers which are referenced before their
        # assignment, if any
        self.undeclared_identifiers = set() # 同上

        if isinstance(code, compat.string_types):
            expr = pyparser.parse(code.lstrip(), "exec", **exception_kwargs)

        f = pyparser.FindIdentifiers(self, **exception_kwargs)
        f.visit(expr)  # 递归解析ast

每个代码片段都会解析成python的ast:

# pyparser

def parse(code, mode="exec", **exception_kwargs):
    return _ast_util.parse(code, "<unknown>", mode)

# _ast_util
def parse(expr, filename="<unknown>", mode="exec"):
    """Parse an expression into an AST node."""
    return compile(expr, filename, mode, PyCF_ONLY_AST)

再看一下示例2:

import ast
expr = """hello, """  # 文本节点
ast_tree = ast.parse(expr)
print(ast.dump(ast_tree))
# 输出
Module(body=[Expr(value=Tuple(elts=[Name(id='hello', ctx=Load())], ctx=Load()))], type_ignores=[])

识别标志:

# pyparser
class FindIdentifiers(_ast_util.NodeVisitor):

    def __init__(self, listener, **exception_kwargs):
        self.listener = listener  # PythonCode对象

    def visit_Name(self, node):
        if isinstance(node.ctx, _ast.Store): # 值
            self._add_declared(node.id)
        elif (
            node.id not in reserved
            and node.id not in self.listener.declared_identifiers
            and node.id not in self.local_ident_stack
        ):  # 变量
            self.listener.undeclared_identifiers.add(node.id)

class FindTuple(_ast_util.NodeVisitor):
    def __init__(self, listener, code_factory, **exception_kwargs):
        self.listener = listener
        self.exception_kwargs = exception_kwargs
        self.code_factory = code_factory

    def visit_Tuple(self, node):
        for n in node.elts:
            ...
            self.listener.args.append(ExpressionGenerator(n).value())  # 生成python代码
            ...

# _ast_util.py
class NodeVisitor(object):

    def get_visitor(self, node):
        method = "visit_" + node.__class__.__name__
        return getattr(self, method, None)  # 动态获取不同ast对象的方法

    def visit(self, node):
        """Visit a node."""
        f = self.get_visitor(node)
        if f is not None:
            return f(node)
        return self.generic_visit(node)  # 递归

    def generic_visit(self, node):
        """Called if no explicit visitor function exists for a node."""
        for field, value in iter_fields(node):
            if isinstance(value, list):
                for item in value:
                    if isinstance(item, AST):
                        self.visit(item)
            elif isinstance(value, AST):
                self.visit(value)

模版解析成node节点，每个node节点包含python标准的的ast对象。

模版编译

前面lexer处理模版文件得到node，node再被转换成python代码:

from mako import codegen

source = codegen.compile(node, "a", default_filters=[])
print(source)

日志如下:

 1  from mako import runtime, filters, cache
 2  UNDEFINED = runtime.UNDEFINED
 3  STOP_RENDERING = runtime.STOP_RENDERING
 4  __M_dict_builtin = dict
 5  __M_locals_builtin = locals
 6  _magic_number = 10
 7  _modified_time = 1615385051.364234
 8  _enable_loop = True
 9  _template_filename = None
10  _template_uri = 'a'
11  _source_encoding = None
12  _exports = []
13
14
15  def render_body(context,**pageargs):
16      __M_caller = context.caller_stack._push_frame()
17      try:
18          __M_locals = __M_dict_builtin(pageargs=pageargs)
19          name = context.get('name', UNDEFINED)  # 从context中获取name属性值
20          __M_writer = context.writer()  # 重定向输出流
21          __M_writer('hello, ')  # 输出文本
22          __M_writer(name) # 输出变量
23          __M_writer('!')
24          return ''
25      finally:
26          context.caller_stack._pop_frame()
27
28
29  """
30  __M_BEGIN_METADATA
31  {"filename": null, "uri": "a", "source_encoding": null, "line_map": {"15": 0, "21": 1, "22": 1, "23": 1, "29": 23}}
32  __M_END_METADATA
33  """                        

source是一段python代码：

• 定义依赖 from mako import ...
• 定义常量 UNDEFINED 和 STOP_RENDERING
• 定义一些私有的变量 _template_filename 等
• 定义了一个关键的 render_body 函数，函数名称和我们在Template中看到 self.callable_ = self.module.render_body 一致
• 定义了一段注释

模版的编译过程是这样的：

def compile(
    node,
    uri,
    ...
):
    buf = util.FastEncodingBuffer()  # 重定向的流
    printer = PythonPrinter(buf)  # printer
    _GenerateRenderMethod(  # 生成render函数
        printer,
        _CompileContext(  # 编译上下文
            uri,
            filename,
            ...
        ),
        node,
    )
    return buf.getvalue()  # 获取buffer的值:

buffer实现比较简单

class FastEncodingBuffer(object):

    def __init__(self, encoding=None, errors="strict", as_unicode=False):
        self.data = collections.deque()  # 使用双端队列模拟输出流
        self.delim = ""
        ...
        self.write = self.data.append

    def getvalue(self):
        ...
        return self.delim.join(self.data)

重点就是_GenerateRenderMethod函数

class _GenerateRenderMethod(object):

    """A template visitor object which generates the
       full module source for a template.

    """

    def __init__(self, printer, compiler, node):
        self.printer = printer
        self.compiler = compiler
        self.node = node
        ...
        # 生成render函数
        self.write_render_callable(
            pagetag or node, name, args, buffered, filtered, cached
        )

write_render_callable的实现:

def write_render_callable(
    self, node, name, args, buffered, filtered, cached
):

    self.printer.start_source(node.lineno)
    self.printer.writelines(
        "def %s(%s):" % (name, ",".join(args)),
        # push new frame, assign current frame to __M_caller
        "__M_caller = context.caller_stack._push_frame()",
        "try:",
    )
    ...

    self.write_variable_declares(self.identifiers, toplevel=True)

    for n in self.node.nodes:
        n.accept_visitor(self) # 遍历节点树

    self.write_def_finish(self.node, buffered, filtered, cached)
    self.printer.writeline(None)
    self.printer.write_blanks(2)

这里配合前面生成的python源码看，就容易理解

将node转换成python源码:

# parsetree
def accept_visitor(self, visitor):
    def traverse(node):
        for n in node.get_children():
            n.accept_visitor(visitor)

    method = getattr(visitor, "visit" + self.__class__.__name__, traverse)
    method(self)

# _ast_utl  
class SourceGenerator(NodeVisitor):

    def visit_Name(self, node): # 写name
        self.write(node.id)

    def visit_FunctionDef(self, node): # 写函数
        self.newline(n=2)
        self.decorators(node)
        self.newline()
        self.write("def %s(" % node.name)
        self.signature(node.args)
        self.write("):")
        self.body(node.body)
    ...

模版渲染

使用这样的伪代码，就可以执行之前生成的 render—body 函数:

context = {"name":"shawn"}
self.callable_(context)

render正是做这样的工作，只是过程更复杂一些:

def render(self, *args, **data):
    return runtime._render(self, self.callable_, args, data)

def _render(template, callable_, args, data, as_unicode=False):
    buf = util.FastEncodingBuffer(
        as_unicode=as_unicode,
        encoding=template.output_encoding,
        errors=template.encoding_errors,
    )  # 重定向输出

    context = Context(buf, **data)  # 执行上下文

    _render_context(
        template,
        callable_,
        context,
        *args,
        **_kwargs_for_callable(callable_, data)
    )  # 执行渲染

    return context._pop_buffer().getvalue()  # 获取渲染结果

FastEncodingBuffer在前面代码生成的时候已经介绍，我们只需要了解runtime.Context类和_render_context方法。runtime.Context涉及的代码:

# runtime.py

class Context(object):

    def __init__(self, buffer, **data):
        self._buffer_stack = [buffer]
        self._data = data
        ....

     def __getitem__(self, key): # 代理data的值
        if key in self._data:
            return self._data[key]
        else:
            return compat_builtins.__dict__[key]

    def writer(self):
        return self._buffer_stack[-1].write # 获取输出流

    def _pop_buffer(self):
        return self._buffer_stack.pop()

_render_context的主要逻辑和我们推断的伪代码一样，就是执行callable_函数:

def _render_context(tmpl, callable_, context, *args, **kwargs):
    ...
    callable_(context, *args, **kwargs)

小结

mako 模版功能较多，我们只是学习了最简单的变量格式输出: "hello, ${name}!".format(name="shawn") 过程。更复杂的条件分支，函数，循环等都没有深入研究。通过这个简单的示例，我们已经知道mako模版的核心逻辑流程:

1. 使用lexer对模版解析生成node节点
2. 将node节点转换成python源码并编译执行得到渲染函数
3. 调用渲染函数完成模版渲染

由于时间紧张，能力有限，mako更深入的解析并未完成，希望下次会深入更多细节实现。

小技巧

使用 StringIO 可以重定向print的输出:

import io

output = io.StringIO()
output.write('First line.\n')
print('Second line.', file=output)

# Retrieve file contents -- this will be
# 'First line.\nSecond line.\n'
contents = output.getvalue()

# Close object and discard memory buffer --
# .getvalue() will now raise an exception.
output.close()

使用 inspect 可以反射函数的参数信息:

import inspect

def inspect_getargspec(func):

    if inspect.ismethod(func):
        func = func.__func__
    if not inspect.isfunction(func):
        raise TypeError("{!r} is not a Python function".format(func))

    co = func.__code__
    if not inspect.iscode(co):
        raise TypeError("{!r} is not a code object".format(co))

    nargs = co.co_argcount
    names = co.co_varnames
    nkwargs = co.co_kwonlyargcount if py3k else 0
    args = list(names[:nargs])

    nargs += nkwargs
    varargs = None
    if co.co_flags & inspect.CO_VARARGS:
        varargs = co.co_varnames[nargs]
        nargs = nargs + 1
    varkw = None
    if co.co_flags & inspect.CO_VARKEYWORDS:
        varkw = co.co_varnames[nargs]

    return ArgSpec(args, varargs, varkw, func.__defaults__)


def _kwargs_for_include(callable_, data, **kwargs):
    argspec = compat.inspect_getargspec(callable_)  # 反射获取函数参数
    namedargs = argspec[0] + [v for v in argspec[1:3] if v is not None]
    for arg in namedargs:
        if arg != "context" and arg in data and arg not in kwargs:
            kwargs[arg] = data[arg]
    return kwargs

参考链接

• https://docs.python.org/zh-cn/3/library/ast.html
• https://vpyast.appspot.com/
• https://zh.wikipedia.org/wiki/%E8%AF%8D%E6%B3%95%E5%88%86%E6%9E%90
• https://www.makotemplates.org/

本文由哈喽比特于3年以前收录，如有侵权请联系我们。
文章来源：https://mp.weixin.qq.com/s/ncOpjI894n38AjStkSruHw