mako/util.py - third_party/mako - Git at Google

 # mako/util.py
 # Copyright (C) 2006-2015 the Mako authors and contributors <see AUTHORS file>
 #
 # This module is part of Mako and is released under
 # the MIT License: http://www.opensource.org/licenses/mit-license.php

 import re
 import collections
 import codecs
 import os
 from mako import compat
 import operator


 def update_wrapper(decorated, fn):
     decorated.__wrapped__ = fn
     decorated.__name__ = fn.__name__
     return decorated


 class PluginLoader(object):

     def __init__(self, group):
         self.group = group
         self.impls = {}

     def load(self, name):
         if name in self.impls:
             return self.impls[name]()
         else:
             import pkg_resources
             for impl in pkg_resources.iter_entry_points(
                     self.group,
                     name):
                 self.impls[name] = impl.load
                 return impl.load()
             else:
                 from mako import exceptions
                 raise exceptions.RuntimeException(
                     "Can't load plugin %s %s" %
                     (self.group, name))

     def register(self, name, modulepath, objname):
         def load():
             mod = __import__(modulepath)
             for token in modulepath.split(".")[1:]:
                 mod = getattr(mod, token)
             return getattr(mod, objname)
         self.impls[name] = load


 def verify_directory(dir):
     """create and/or verify a filesystem directory."""

     tries = 0

     while not os.path.exists(dir):
         try:
             tries += 1
             os.makedirs(dir, compat.octal("0775"))
         except:
             if tries > 5:
                 raise


 def to_list(x, default=None):
     if x is None:
         return default
     if not isinstance(x, (list, tuple)):
         return [x]
     else:
         return x


 class memoized_property(object):

     """A read-only @property that is only evaluated once."""

     def __init__(self, fget, doc=None):
         self.fget = fget
         self.__doc__ = doc or fget.__doc__
         self.__name__ = fget.__name__

     def __get__(self, obj, cls):
         if obj is None:
             return self
         obj.__dict__[self.__name__] = result = self.fget(obj)
         return result


 class memoized_instancemethod(object):

     """Decorate a method memoize its return value.

     Best applied to no-arg methods: memoization is not sensitive to
     argument values, and will always return the same value even when
     called with different arguments.

     """

     def __init__(self, fget, doc=None):
         self.fget = fget
         self.__doc__ = doc or fget.__doc__
         self.__name__ = fget.__name__

     def __get__(self, obj, cls):
         if obj is None:
             return self

         def oneshot(*args, **kw):
             result = self.fget(obj, *args, **kw)
             memo = lambda *a, **kw: result
             memo.__name__ = self.__name__
             memo.__doc__ = self.__doc__
             obj.__dict__[self.__name__] = memo
             return result
         oneshot.__name__ = self.__name__
         oneshot.__doc__ = self.__doc__
         return oneshot


 class SetLikeDict(dict):

     """a dictionary that has some setlike methods on it"""

     def union(self, other):
         """produce a 'union' of this dict and another (at the key level).

         values in the second dict take precedence over that of the first"""
         x = SetLikeDict(**self)
         x.update(other)
         return x


 class FastEncodingBuffer(object):

     """a very rudimentary buffer that is faster than StringIO,
     but doesn't crash on unicode data like cStringIO."""

     def __init__(self, encoding=None, errors='strict', as_unicode=False):
         self.data = collections.deque()
         self.encoding = encoding
         if as_unicode:
             self.delim = compat.u('')
         else:
             self.delim = ''
         self.as_unicode = as_unicode
         self.errors = errors
         self.write = self.data.append

     def truncate(self):
         self.data = collections.deque()
         self.write = self.data.append

     def getvalue(self):
         if self.encoding:
             return self.delim.join(self.data).encode(self.encoding,
                                                      self.errors)
         else:
             return self.delim.join(self.data)


 class LRUCache(dict):

     """A dictionary-like object that stores a limited number of items,
     discarding lesser used items periodically.

     this is a rewrite of LRUCache from Myghty to use a periodic timestamp-based
     paradigm so that synchronization is not really needed.  the size management
     is inexact.
     """

     class _Item(object):

         def __init__(self, key, value):
             self.key = key
             self.value = value
             self.timestamp = compat.time_func()

         def __repr__(self):
             return repr(self.value)

     def __init__(self, capacity, threshold=.5):
         self.capacity = capacity
         self.threshold = threshold

     def __getitem__(self, key):
         item = dict.__getitem__(self, key)
         item.timestamp = compat.time_func()
         return item.value

     def values(self):
         return [i.value for i in dict.values(self)]

     def setdefault(self, key, value):
         if key in self:
             return self[key]
         else:
             self[key] = value
             return value

     def __setitem__(self, key, value):
         item = dict.get(self, key)
         if item is None:
             item = self._Item(key, value)
             dict.__setitem__(self, key, item)
         else:
             item.value = value
         self._manage_size()

     def _manage_size(self):
         while len(self) > self.capacity + self.capacity * self.threshold:
             bytime = sorted(dict.values(self),
                             key=operator.attrgetter('timestamp'), reverse=True)
             for item in bytime[self.capacity:]:
                 try:
                     del self[item.key]
                 except KeyError:
                     # if we couldn't find a key, most likely some other thread
                     # broke in on us. loop around and try again
                     break

 # Regexp to match python magic encoding line
 _PYTHON_MAGIC_COMMENT_re = re.compile(
     r'[ \t\f]* \# .* coding[=:][ \t]*([-\w.]+)',
     re.VERBOSE)


 def parse_encoding(fp):
     """Deduce the encoding of a Python source file (binary mode) from magic
     comment.

     It does this in the same way as the `Python interpreter`__

     .. __: http://docs.python.org/ref/encodings.html

     The ``fp`` argument should be a seekable file object in binary mode.
     """
     pos = fp.tell()
     fp.seek(0)
     try:
         line1 = fp.readline()
         has_bom = line1.startswith(codecs.BOM_UTF8)
         if has_bom:
             line1 = line1[len(codecs.BOM_UTF8):]

         m = _PYTHON_MAGIC_COMMENT_re.match(line1.decode('ascii', 'ignore'))
         if not m:
             try:
                 import parser
                 parser.suite(line1.decode('ascii', 'ignore'))
             except (ImportError, SyntaxError):
                 # Either it's a real syntax error, in which case the source
                 # is not valid python source, or line2 is a continuation of
                 # line1, in which case we don't want to scan line2 for a magic
                 # comment.
                 pass
             else:
                 line2 = fp.readline()
                 m = _PYTHON_MAGIC_COMMENT_re.match(
                     line2.decode('ascii', 'ignore'))

         if has_bom:
             if m:
                 raise SyntaxError(
                     "python refuses to compile code with both a UTF8"
                     " byte-order-mark and a magic encoding comment")
             return 'utf_8'
         elif m:
             return m.group(1)
         else:
             return None
     finally:
         fp.seek(pos)


 def sorted_dict_repr(d):
     """repr() a dictionary with the keys in order.

     Used by the lexer unit test to compare parse trees based on strings.

     """
     keys = list(d.keys())
     keys.sort()
     return "{" + ", ".join(["%r: %r" % (k, d[k]) for k in keys]) + "}"


 def restore__ast(_ast):
     """Attempt to restore the required classes to the _ast module if it
     appears to be missing them
     """
     if hasattr(_ast, 'AST'):
         return
     _ast.PyCF_ONLY_AST = 2 << 9
     m = compile("""\
 def foo(): pass
 class Bar(object): pass
 if False: pass
 baz = 'mako'
 1 + 2 - 3 * 4 / 5
 6 // 7 % 8 << 9 >> 10
 11 & 12 ^ 13 | 14
 15 and 16 or 17
 -baz + (not +18) - ~17
 baz and 'foo' or 'bar'
 (mako is baz == baz) is not baz != mako
 mako > baz < mako >= baz <= mako
 mako in baz not in mako""", '<unknown>', 'exec', _ast.PyCF_ONLY_AST)
     _ast.Module = type(m)

     for cls in _ast.Module.__mro__:
         if cls.__name__ == 'mod':
             _ast.mod = cls
         elif cls.__name__ == 'AST':
             _ast.AST = cls

     _ast.FunctionDef = type(m.body[0])
     _ast.ClassDef = type(m.body[1])
     _ast.If = type(m.body[2])

     _ast.Name = type(m.body[3].targets[0])
     _ast.Store = type(m.body[3].targets[0].ctx)
     _ast.Str = type(m.body[3].value)

     _ast.Sub = type(m.body[4].value.op)
     _ast.Add = type(m.body[4].value.left.op)
     _ast.Div = type(m.body[4].value.right.op)
     _ast.Mult = type(m.body[4].value.right.left.op)

     _ast.RShift = type(m.body[5].value.op)
     _ast.LShift = type(m.body[5].value.left.op)
     _ast.Mod = type(m.body[5].value.left.left.op)
     _ast.FloorDiv = type(m.body[5].value.left.left.left.op)

     _ast.BitOr = type(m.body[6].value.op)
     _ast.BitXor = type(m.body[6].value.left.op)
     _ast.BitAnd = type(m.body[6].value.left.left.op)

     _ast.Or = type(m.body[7].value.op)
     _ast.And = type(m.body[7].value.values[0].op)

     _ast.Invert = type(m.body[8].value.right.op)
     _ast.Not = type(m.body[8].value.left.right.op)
     _ast.UAdd = type(m.body[8].value.left.right.operand.op)
     _ast.USub = type(m.body[8].value.left.left.op)

     _ast.Or = type(m.body[9].value.op)
     _ast.And = type(m.body[9].value.values[0].op)

     _ast.IsNot = type(m.body[10].value.ops[0])
     _ast.NotEq = type(m.body[10].value.ops[1])
     _ast.Is = type(m.body[10].value.left.ops[0])
     _ast.Eq = type(m.body[10].value.left.ops[1])

     _ast.Gt = type(m.body[11].value.ops[0])
     _ast.Lt = type(m.body[11].value.ops[1])
     _ast.GtE = type(m.body[11].value.ops[2])
     _ast.LtE = type(m.body[11].value.ops[3])

     _ast.In = type(m.body[12].value.ops[0])
     _ast.NotIn = type(m.body[12].value.ops[1])


 def read_file(path, mode='rb'):
     fp = open(path, mode)
     try:
         data = fp.read()
         return data
     finally:
         fp.close()


 def read_python_file(path):
     fp = open(path, "rb")
     try:
         encoding = parse_encoding(fp)
         data = fp.read()
         if encoding:
             data = data.decode(encoding)
         return data
     finally:
         fp.close()
	# mako/util.py
	# Copyright (C) 2006-2015 the Mako authors and contributors <see AUTHORS file>
	#
	# This module is part of Mako and is released under
	# the MIT License: http://www.opensource.org/licenses/mit-license.php

	import re
	import collections
	import codecs
	import os
	from mako import compat
	import operator


	def update_wrapper(decorated, fn):
	decorated.__wrapped__ = fn
	decorated.__name__ = fn.__name__
	return decorated


	class PluginLoader(object):

	def __init__(self, group):
	self.group = group
	self.impls = {}

	def load(self, name):
	if name in self.impls:
	return self.impls[name]()
	else:
	import pkg_resources
	for impl in pkg_resources.iter_entry_points(
	self.group,
	name):
	self.impls[name] = impl.load
	return impl.load()
	else:
	from mako import exceptions
	raise exceptions.RuntimeException(
	"Can't load plugin %s %s" %
	(self.group, name))

	def register(self, name, modulepath, objname):
	def load():
	mod = __import__(modulepath)
	for token in modulepath.split(".")[1:]:
	mod = getattr(mod, token)
	return getattr(mod, objname)
	self.impls[name] = load


	def verify_directory(dir):
	"""create and/or verify a filesystem directory."""

	tries = 0

	while not os.path.exists(dir):
	try:
	tries += 1
	os.makedirs(dir, compat.octal("0775"))
	except:
	if tries > 5:
	raise


	def to_list(x, default=None):
	if x is None:
	return default
	if not isinstance(x, (list, tuple)):
	return [x]
	else:
	return x


	class memoized_property(object):

	"""A read-only @property that is only evaluated once."""

	def __init__(self, fget, doc=None):
	self.fget = fget
	self.__doc__ = doc or fget.__doc__
	self.__name__ = fget.__name__

	def __get__(self, obj, cls):
	if obj is None:
	return self
	obj.__dict__[self.__name__] = result = self.fget(obj)
	return result


	class memoized_instancemethod(object):

	"""Decorate a method memoize its return value.

	Best applied to no-arg methods: memoization is not sensitive to
	argument values, and will always return the same value even when
	called with different arguments.

	"""

	def __init__(self, fget, doc=None):
	self.fget = fget
	self.__doc__ = doc or fget.__doc__
	self.__name__ = fget.__name__

	def __get__(self, obj, cls):
	if obj is None:
	return self

	def oneshot(args, *kw):
	result = self.fget(obj, args, *kw)
	memo = lambda a, *kw: result
	memo.__name__ = self.__name__
	memo.__doc__ = self.__doc__
	obj.__dict__[self.__name__] = memo
	return result
	oneshot.__name__ = self.__name__
	oneshot.__doc__ = self.__doc__
	return oneshot


	class SetLikeDict(dict):

	"""a dictionary that has some setlike methods on it"""

	def union(self, other):
	"""produce a 'union' of this dict and another (at the key level).

	values in the second dict take precedence over that of the first"""
	x = SetLikeDict(**self)
	x.update(other)
	return x


	class FastEncodingBuffer(object):

	"""a very rudimentary buffer that is faster than StringIO,
	but doesn't crash on unicode data like cStringIO."""

	def __init__(self, encoding=None, errors='strict', as_unicode=False):
	self.data = collections.deque()
	self.encoding = encoding
	if as_unicode:
	self.delim = compat.u('')
	else:
	self.delim = ''
	self.as_unicode = as_unicode
	self.errors = errors
	self.write = self.data.append

	def truncate(self):
	self.data = collections.deque()
	self.write = self.data.append

	def getvalue(self):
	if self.encoding:
	return self.delim.join(self.data).encode(self.encoding,
	self.errors)
	else:
	return self.delim.join(self.data)


	class LRUCache(dict):

	"""A dictionary-like object that stores a limited number of items,
	discarding lesser used items periodically.

	this is a rewrite of LRUCache from Myghty to use a periodic timestamp-based
	paradigm so that synchronization is not really needed. the size management
	is inexact.
	"""

	class _Item(object):

	def __init__(self, key, value):
	self.key = key
	self.value = value
	self.timestamp = compat.time_func()

	def __repr__(self):
	return repr(self.value)

	def __init__(self, capacity, threshold=.5):
	self.capacity = capacity
	self.threshold = threshold

	def __getitem__(self, key):
	item = dict.__getitem__(self, key)
	item.timestamp = compat.time_func()
	return item.value

	def values(self):
	return [i.value for i in dict.values(self)]

	def setdefault(self, key, value):
	if key in self:
	return self[key]
	else:
	self[key] = value
	return value

	def __setitem__(self, key, value):
	item = dict.get(self, key)
	if item is None:
	item = self._Item(key, value)
	dict.__setitem__(self, key, item)
	else:
	item.value = value
	self._manage_size()

	def _manage_size(self):
	while len(self) > self.capacity + self.capacity * self.threshold:
	bytime = sorted(dict.values(self),
	key=operator.attrgetter('timestamp'), reverse=True)
	for item in bytime[self.capacity:]:
	try:
	del self[item.key]
	except KeyError:
	# if we couldn't find a key, most likely some other thread
	# broke in on us. loop around and try again
	break

	# Regexp to match python magic encoding line
	_PYTHON_MAGIC_COMMENT_re = re.compile(
	r'[ \t\f]* \# .* coding[=:][ \t]*([-\w.]+)',
	re.VERBOSE)


	def parse_encoding(fp):
	"""Deduce the encoding of a Python source file (binary mode) from magic
	comment.

	It does this in the same way as the `Python interpreter`__

	.. __: http://docs.python.org/ref/encodings.html

	The ``fp`` argument should be a seekable file object in binary mode.
	"""
	pos = fp.tell()
	fp.seek(0)
	try:
	line1 = fp.readline()
	has_bom = line1.startswith(codecs.BOM_UTF8)
	if has_bom:
	line1 = line1[len(codecs.BOM_UTF8):]

	m = _PYTHON_MAGIC_COMMENT_re.match(line1.decode('ascii', 'ignore'))
	if not m:
	try:
	import parser
	parser.suite(line1.decode('ascii', 'ignore'))
	except (ImportError, SyntaxError):
	# Either it's a real syntax error, in which case the source
	# is not valid python source, or line2 is a continuation of
	# line1, in which case we don't want to scan line2 for a magic
	# comment.
	pass
	else:
	line2 = fp.readline()
	m = _PYTHON_MAGIC_COMMENT_re.match(
	line2.decode('ascii', 'ignore'))

	if has_bom:
	if m:
	raise SyntaxError(
	"python refuses to compile code with both a UTF8"
	" byte-order-mark and a magic encoding comment")
	return 'utf_8'
	elif m:
	return m.group(1)
	else:
	return None
	finally:
	fp.seek(pos)


	def sorted_dict_repr(d):
	"""repr() a dictionary with the keys in order.

	Used by the lexer unit test to compare parse trees based on strings.

	"""
	keys = list(d.keys())
	keys.sort()
	return "{" + ", ".join(["%r: %r" % (k, d[k]) for k in keys]) + "}"


	def restore__ast(_ast):
	"""Attempt to restore the required classes to the _ast module if it
	appears to be missing them
	"""
	if hasattr(_ast, 'AST'):
	return
	_ast.PyCF_ONLY_AST = 2 << 9
	m = compile("""\
	def foo(): pass
	class Bar(object): pass
	if False: pass
	baz = 'mako'
	1 + 2 - 3 * 4 / 5
	6 // 7 % 8 << 9 >> 10
	11 & 12 ^ 13 \| 14
	15 and 16 or 17
	-baz + (not +18) - ~17
	baz and 'foo' or 'bar'
	(mako is baz == baz) is not baz != mako
	mako > baz < mako >= baz <= mako
	mako in baz not in mako""", '<unknown>', 'exec', _ast.PyCF_ONLY_AST)
	_ast.Module = type(m)

	for cls in _ast.Module.__mro__:
	if cls.__name__ == 'mod':
	_ast.mod = cls
	elif cls.__name__ == 'AST':
	_ast.AST = cls

	_ast.FunctionDef = type(m.body[0])
	_ast.ClassDef = type(m.body[1])
	_ast.If = type(m.body[2])

	_ast.Name = type(m.body[3].targets[0])
	_ast.Store = type(m.body[3].targets[0].ctx)
	_ast.Str = type(m.body[3].value)

	_ast.Sub = type(m.body[4].value.op)
	_ast.Add = type(m.body[4].value.left.op)
	_ast.Div = type(m.body[4].value.right.op)
	_ast.Mult = type(m.body[4].value.right.left.op)

	_ast.RShift = type(m.body[5].value.op)
	_ast.LShift = type(m.body[5].value.left.op)
	_ast.Mod = type(m.body[5].value.left.left.op)
	_ast.FloorDiv = type(m.body[5].value.left.left.left.op)

	_ast.BitOr = type(m.body[6].value.op)
	_ast.BitXor = type(m.body[6].value.left.op)
	_ast.BitAnd = type(m.body[6].value.left.left.op)

	_ast.Or = type(m.body[7].value.op)
	_ast.And = type(m.body[7].value.values[0].op)

	_ast.Invert = type(m.body[8].value.right.op)
	_ast.Not = type(m.body[8].value.left.right.op)
	_ast.UAdd = type(m.body[8].value.left.right.operand.op)
	_ast.USub = type(m.body[8].value.left.left.op)

	_ast.Or = type(m.body[9].value.op)
	_ast.And = type(m.body[9].value.values[0].op)

	_ast.IsNot = type(m.body[10].value.ops[0])
	_ast.NotEq = type(m.body[10].value.ops[1])
	_ast.Is = type(m.body[10].value.left.ops[0])
	_ast.Eq = type(m.body[10].value.left.ops[1])

	_ast.Gt = type(m.body[11].value.ops[0])
	_ast.Lt = type(m.body[11].value.ops[1])
	_ast.GtE = type(m.body[11].value.ops[2])
	_ast.LtE = type(m.body[11].value.ops[3])

	_ast.In = type(m.body[12].value.ops[0])
	_ast.NotIn = type(m.body[12].value.ops[1])


	def read_file(path, mode='rb'):
	fp = open(path, mode)
	try:
	data = fp.read()
	return data
	finally:
	fp.close()


	def read_python_file(path):
	fp = open(path, "rb")
	try:
	encoding = parse_encoding(fp)
	data = fp.read()
	if encoding:
	data = data.decode(encoding)
	return data
	finally:
	fp.close()