Python元类

[Metaclasses] are deeper magic than 99% of users should ever worry about. If you wonder whether you need them, you don’t (the people who actually need them know with certainty that they need them, and don’t need an explanation about why). The Python core developer Time Peters said.

元类是创建类的类。新风格(new-style)类是type类的实例。元类是type类的派生类,通过重载type类的__new____init__方法,重新定义类创建协议,来实现类创建的定制化。

元类模型

(对象)实例是通过类创建,类是通过type类创建,元类是type类的派生类。

  • 类型(自定义)是通过type类或type派生元类创建
  • 元类是type类的派生类
  • 用户自定义类是type类的实例
  • 用户自定义类可以生成自己的实例

类声明协议

Python解释器在类声明语句结束时,调用type类型创建,class = type(classname, superclasses, attributedict)

1
2
type.__new__(meta, classname, superclasses, attributedict)
type.__init__(cls, classname, superclasses, attributedict)

声明元类

Py3与Py2声明元类的方式不一样:

Py3声明元类

1
2
3
4
5
6
7
8
9
>>> class Metaclass(type):
... def __new__(meta, classname, superclasses, attributedict):
... return type(classname, superclasses, attributedict)
... def __init__(cls, classname, superclasses, attributedict):
... pass
...
>>> class Dummy(object, metaclass=Metaclass):
... pass
...

Py2声明元类

1
2
3
4
5
6
7
8
9
>>> class Metaclass(type):
... def __new__(meta, classname, superclasses, attributedict):
... return type(classname, superclasses, attributedict)
... def __init__(cls, classname, superclasses, attributedict):
... pass
...
>>> class Dummy(object):
... __metaclass__ = Metaclass
...

继承和实例

  • 元类继承于type类
  • 元类的声明可以被派生类继承
  • 元类的属性不能被类的实例继承
  • 元类的属性可以被类获取

元类继承于type类

元类重载type类的__new____init__方法,定制类的创建和初始化。

元类的声明可以被派生类继承

元类的属性不能被类实例继承

类是元类的实例,元类的行为可以被类访问,当类不能被类的实例访问。

元类的属性可以被类获取

继承

Python继承算法

  1. 对于一个实例,先搜索这个实例,再搜索实例的类,再搜索超类
    a. 先搜索实例的__dict__
    b. 再搜索该实例的类的__mro__对应类的__dict__
  2. 对于一个类,先搜索类,再搜索超类,再搜索元类
    a. 根据__mro__搜索类的__dict__
    b. 再搜索元类的__dict__
  3. 规则1和2中,再b阶段中,数据描述的优先级高
  4. 规则1和2中,对于内置的运算符,跳过a阶段

数据描述符继承算法

对于定义了__set__拦截赋值的描述符就是数据描述符。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
>>> class D(object):
... def __get__(self, ins, _type):
... print('call D.__get__')
... def __set__(self, ins, value):
... print('call D.__set__')
...
>>>
>>> class Dummy(object):
... d = D()
...
>>> ins = Dummy()
>>> ins.d
call D.__get__
>>> ins.d = 'spam'
call D.__set__

未定义__set__的描述符

1
2
3
4
5
6
7
8
9
10
11
12
13
>>> class D(object):
... def __get__(self, ins, value):
... print('call D.__get__')
...
>>> class Dummy(object):
... d = D()
...
>>> ins = Dummy()
>>> ins.d
call D.__get__
>>> ins.d = 'spam'
>>> ins.d
'spam

Python 的继承算法

  1. 对于实例I,先搜索实例,再搜索类,再搜索超类
    a. 根据类的__mro__搜索超类的__dict__
    b. If 如果在a阶段发现了数据描述,调用该数据描述,完成后退出
    c. Else 返回该实例__dict__中的值
    d. Else 调用非数据描述符,并放回结果
  2. 对于类C,搜索类,再搜索超类,再搜索元类
    a. 搜索类的__mro__依次搜索类的__dict__
    b. If 如果在a阶段发现了数据描述符,调用该数据描述符,完成后退出
    c. Else 返回该类__dict__中值g
    d. Else 调用非数据描述符,返回结果

Note, 数据描述符的优先级 > 普通属性 > 非数据描述符

元类与类装饰器

TODO

示例

django ORM

ripozo API

参考

  • Learning Python 5th Edition

Python描述符

如果一个对象定义了以下任意方法,这个对象就是一个描述符。给描述符下个定义,描述符就是绑定了行为属性的对象。

object.__get__(self, instance, owner)

object.__set__(self, instance, value)

object.__delete__(self, instance)

属性访问的默认行为就是从一个对象字典中获取、设置和删除属性。比如,a.x首先会搜索a.__dict__['x'],其次type(a).__dict__['x'],最后所有type(a)的元类。如果要查找的值一个包含描述器方法的对象,Python会用调用描述器方法代替默认行为。

Note:只有new-style class会调用描述符的对象的方法。

描述符是一个强大的通用协议。Python内建的property, staticmethod, classmethod, super背后的实现机制都是描述符协议。

Descriptor Protocol

object.__get__(self, ins, _type=None)

object.__set__(self, ins, value)

object.__del__(self, ins)

如果一个对象包含上面任意一个方法,就可以被看作是一个描述符。如果一个对象定义了__get____set__两个方法,该对象可以被看作一个数据描述符,如果一个对象只定义了__get__,该对象就是non-data描述符。

数据描述符与非数据描述符的区别在于,描述符与对象实例entry调用优先级。如果一个实例的字典有一个entry和数据描述符的名字相同,数据描述符的调用的优先级高。如果一个实例有一个entry和非数据描述符的名字相同个,实例entry的调用的优先级高。

Invoking Descriptors

obj.d查找obj的字典是否包含d,如果d定义了__get__方法,d.__get__(obj, type(obj))就会被调用。

1
2
3
4
5
6
def __getattribute__(self, key):
"Emulate type_getattro() in Objects/typeobject.c"
v = object.__getattribute__(self, key)
if hasattr(v, '__get__'):
return v.__get__(None, self)
return v

super()返回的对象有一个定制化的__getattribute__方法,用于调用描述符。super(B, self).m先会搜索obj.__class__.__mro__查找基类A,如果是一个数据描述符,则会调用A.__dict__['m'].__get__(obj, B),如果是一个非数据描述符,返回结果不会改变。

Built-in Descriptors

Property

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class C(object):
def getx(self): return self.__x
def setx(self, value): self.__x = value
def delx(self): del self.__x
x = property(getx, setx, delx, "I'm the 'x' property.")


class Property(object):
"Emulate PyProperty_Type() in Objects/descrobject.c"

def __init__(self, fget=None, fset=None, fdel=None, doc=None):
self.fget = fget
self.fset = fset
self.fdel = fdel
if doc is None and fget is not None:
doc = fget.__doc__
self.__doc__ = doc

def __get__(self, obj, objtype=None):
if obj is None:
return self
if self.fget is None:
raise AttributeError("unreadable attribute")
return self.fget(obj)

def __set__(self, obj, value):
if self.fset is None:
raise AttributeError("can't set attribute")
self.fset(obj, value)

def __delete__(self, obj):
if self.fdel is None:
raise AttributeError("can't delete attribute")
self.fdel(obj)

def getter(self, fget):
return type(self)(fget, self.fset, self.fdel, self.__doc__)

def setter(self, fset):
return type(self)(self.fget, fset, self.fdel, self.__doc__)

def deleter(self, fdel):
return type(self)(self.fget, self.fset, fdel, self.__doc__)

Staticmethod

1
2
3
4
5
6
7
8
class StaticMethod(object):
"Emulate PyStaticMethod_Type() in Objects/funcobject.c"

def __init__(self, f):
self.f = f

def __get__(self, obj, objtype=None):
return self.f

Classmethod

1
2
3
4
5
6
7
8
9
10
11
12
class ClassMethod(object):
"Emulate PyClassMethod_Type() in Objects/funcobject.c"

def __init__(self, f):
self.f = f

def __get__(self, obj, klass=None):
if klass is None:
klass = type(obj)
def newfunc(*args):
return self.f(klass, *args)
return newfunc

References

Python上下文管理器

上下管理器是一个对象,定义了执行with语句时需要创建的上下文。context manager的__enter__()__exit__()方法分别在进入、退出with语句时被调用。

object.__enter__(self)

object.__exit__(self, exc_type, exc_value, traceback)

with statement

1
2
with_stmt ::=  "with" with_item ("," with_item)* ":" suite
with_item ::= expression ["as" target]

with语句执行数据流:

  1. 评估上下文表达式是否包含上下文管理器
  2. 加载上下文管理器的__exit__方法
  3. 调用上下文管理的__enter__方法
  4. 如果target包含在with语句中,将__enter__方法的返回值赋给target
  5. 执行suite
  6. 调用__exit__()方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
mgr = (EXPR)
exit = type(mgr).__exit__ # Not calling it yet
value = type(mgr).__enter__(mgr)
exc = True
try:
VAR = value # Only if "as VAR" is present
BLOCK
except:
# The exceptional case is handled here
exc = False
if not exit(mgr, *sys.exc_info()):
raise
# The exception is swallowed if exit() returns true
finally:
# The normal and non-local-goto cases are handled here
if exc:
exit(mgr, None, None, None)

contextlib

contextlib提供了快速定义支持上下文管理器的函数对象。

定义一个生成器函数,contextmanager装饰后就变成一个支持上下文管理器的函数对象。yield之前语句子在代码块之前被执行,yield之后语句在代码执行完之后被执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
>>> from contextlib import contextmanager
>>>
>>> @contextmanager
... def tag(name):
... print('<%s>' % name)
... yield
... print('</%s>' % name)
...
>>> with tag('h1'):
... print('hotbaby')
...
<h1>
hotbaby
</h1>

context decorator

contextmanager是一个函数装饰器,装饰只包含一个yield语句的生成器函数,返回一个支持上下文管理器的函数对象。

1
2
3
4
5
def contextmanager(func):
@wraps(func)
def helper(*args, **kwds):
return GeneratorContextManager(func(*args, **kwds))
return helper

Note: 被装饰的生成器函数变成生成器作为参数传递到GeneratorContextManager对象中。

生成器上下文管理器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class GeneratorContextManager(object):
"""Helper for @contextmanager decorator."""

def __init__(self, gen):
self.gen = gen

def __enter__(self):
try:
return self.gen.next()
except StopIteration:
raise RuntimeError("generator didn't yield")

def __exit__(self, type, value, traceback):
if type is None:
try:
self.gen.next()
except StopIteration:
return
else:
raise RuntimeError("generator didn't stop")
else:
if value is None:
# Need to force instantiation so we can reliably
# tell if we get the same exception back
value = type()
try:
self.gen.throw(type, value, traceback)
raise RuntimeError("generator didn't stop after throw()")
except StopIteration, exc:
# Suppress the exception *unless* it's the same exception that
# was passed to throw(). This prevents a StopIteration
# raised inside the "with" statement from being suppressed
return exc is not value
except:
# only re-raise if it's *not* the exception that was
# passed to throw(), because __exit__() must not raise
# an exception unless __exit__() itself failed. But throw()
# has to raise the exception to signal propagation, so this
# fixes the impedance mismatch between the throw() protocol
# and the __exit__() protocol.
#
if sys.exc_info()[1] is not value:
raise

references

Python Itertools

itertools.imap()

imap(func, *iterables) 和iter和map混合体。

工作原理

1
2
3
4
5
6
7
8
9
>>> def myimap(func, *iterables):
iterables = map(iter,iterables)
while True:
args = [next(it) for it in iterables]
yield func(*args)


>>> [item for item in myimap(pow, (2,2), (3,3))]
[8, 8]

itertools.chain()

chain(*iterables) 返回一个迭代器,该迭代器依次返回可迭代对象中没一个元素。

工作原理

1
2
3
4
5
6
7
>>> def mychain(*iterables):
for iter_ in iterables:
for item in iter_:
yield item

>>> [item for item in mychain('abc', 'def')]
['a', 'b', 'c', 'd', 'e', 'f']

References

Python PyPI

PyPI(Python Package Index)是Python软件仓库。

pip是Python包管理工具,默认使用pypi.python.org作为PyPI的镜像。

pip安装Python软件包经常会出现”链接pypi.python.org失败”。为了优化包的管理,考虑替换pypi.python.org,转而使用国内的PyPI镜像。

PyPI mirror list

Mirror Location
pypi.python.org San Francisco, California US
pypi.douban.com Beijing, Beijing CN
pypi.fcio.net Oberhausen, Nordrhein-Westfalen DE

Linux(Debian) 替换PyPI镜像

touch ~/.pip/pip.conf

1
2
3
[global]
index-url = https://pypi.douban.com/simple
format = columns

参考

Lisp

Lisp语言从诞生的时候就包含9种思想.其中一些我们今天已经习以为常,另一些则刚刚在其他高级语言中出现,至今还有2种是Lisp独有的.按照大众的接受程度,这9种思想依次如下排列:

  1. 条件结构.现在大家都觉得这是理所当然的,但是Fortran I就没有这个结构,它只有底层机器的goto结构.
  2. 函数也是一种数据类型.在Lisp语言中,函数与整数或字符串一样,也属于数据类型的一种.它有自己的字面表示形式(literal representation),能狗存储在变量中,也能当作参数传递.一种数据类型应该有的功能,它都有.
  3. 递归.Lisp是第一个支持递归函数的高级语言.
  4. 变量的动态类型.在Lisp语言中,所有变量实际上都是指针,所指向的值有类型之分,而变量本身没有.复制变量就是相当于复制指针,而不是复制它们指向的数据.
  5. 垃圾回收机制.
  6. 程序由表达式组成.Lisp程序是一些表达树的集合,每个表达式都返回一个值.这与Fortran和大多数后来的语言都截然不同,他们的程序都由表达式和语句组成.区分表达式与语句在Fortran I中是自然的,因为它不支持语句嵌套.所以,如果你需要用数学式子计算一个值,那就只有表达式返回这个值,没有其他语法结构可用,否则就无法处理这个值.后来,新的编程语言支持块结构,这种限制当然就不存在了.但是为时已晚,表达式和语句的区分已经根深蒂固.它从Fortran扩散到它们两者的后继语言.
  7. 符号类型.符号实际上是一种指针,指向存储在散列表中字符串.所以,比较两个符号是否相等,只要看它们的指针是否一样就可以了,不用逐个字符比较.
  8. 代码使用符号和常量组成的树形表示法.
  9. 无论什么时候,整个语言都是可用的.Lisp并不真正区分读取,编译期和运行期.你可以在读取期编译或运行代码,也可以在编译期读取和运行代码,还可以在运行期读取或编译代码.在读取期运行代码,使得用户可以重新调整Lisp的语法,在编译期运行代码,则是Lisp宏的工作基础,在运行期编译代码,使得Lisp可以在Emacs这样的程序中充当扩展语言(extension language),在运行期读取代码,使得程序之间可以用S表达式通信,近来XML格式的出现使得这个概念被重新”发明”出来了.

Lisp语言刚出现的时候,这些思想与其他编程语言大相径庭,后者的设计思想主要由50年代后期的硬件决定.随着时间流逝,流行的编程语言不断更新换代,语言设计思想逐渐向Lisp靠拢.思想(1)到思想(5)已经被广泛接受,思想(6)开始在主流编程语言中出现,思想(7)在Python语言中有所实现,不过似乎没有专用的语法.

思想(8)可能是最有意思的一点.它与思想(9)只是由于偶然的原因成为Lisp语言的一部分,因为它们不属于麦卡锡的原始构想,是由拉塞尔自行添加的.它们从此使得Lisp语言看上去很古怪,但也成为了这种语言最独一无二的特点.说Lisp语法古怪不是因为它的语法很古怪,而是因为它根本就没有语法,程序直接以解析树(parse tree)的形式表达出来.在其他语言中,这种形式只是经过解析在后台产生,但是Lisp直接采用它作为表达式形式.它由列表构成,而列表则是Lisp的基本数据结构.

用一种语言自己的数据结构来表达该语言是非常强大的功能.思想(8)和思想(9),意味着你可以写出一种能够自己编程的程序.

Django数据库路由

django ORM数据模型配置数据库.

django支持多个数据库,通过django ORM定义数据模型,比如class User(Model),无法通过class Meta配置管理该数据模型对应的数据库,只能使用默认数据库default

django ConnectionRouter解决数据模型与数据库映射.

实现DB router

db_router.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import logging
from django.conf import settings

_logger = logging.getLogger('django')

class DatabaseRouter(object):
"""
Database router to control the models for differrent db.
"""

DEFAULT_DB = 'default'

def _db(self, model, **hints):
db = getattr(model, '_database', None)
if not db:
return self.DEFAULT_DB

if db in settings.DATABASES.keys():
return db
else:
_logger.warn('%s not exist' % db)
return self.DEFAULT_DB

def db_for_read(self, model, **hints):
return self._db(model, **hints)

def db_for_write(self, model, **hints):
return self._db(model, **hints)

配置DB routers

DATABASE_ROUTERS = ['db_router.DatabaseRouter']

为Model制定数据库

1
2
3
4
5
class User(Model)
_database = 'user_db'
class Meta:
db_table = 'user'
...

实现原理

django通过ConnectionRouter管理数据库路由

django/db/utils.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class ConnectionRouter(object):

@cached_property
def routers(self):
if self._routers is None:
self._routers = settings.DATABASE_ROUTERS
routers = []
for r in self._routers:
if isinstance(r, six.string_types):
router = import_string(r)()
else:
router = r
routers.append(router)
return routers

def _router_func(action):
def _route_db(self, model, **hints):
chosen_db = None
for router in self.routers:
try:
method = getattr(router, action)
except AttributeError:
# If the router doesn't have a method, skip to the next one.
pass
else:
chosen_db = method(model, **hints)
if chosen_db:
return chosen_db
instance = hints.get('instance')
if instance is not None and instance._state.db:
return instance._state.db
return DEFAULT_DB_ALIAS
return _route_db

db_for_read = _router_func('db_for_read')
db_for_write = _router_func('db_for_write')

router初始化

router = ConnectionRouter()

router引用

django/db/models/query.py

1
2
3
4
5
6
7
8
class QuerySet(object):

@property
def db(self):
"Return the database that will be used if this query is executed now"
if self._for_write:
return self._db or router.db_for_write(self.model, **self._hints)
return self._db or router.db_for_read(self.model, **self._hints)

Python WSGI

WSGI(Web Server Gateway Interface) Web服务网关接口。

WSGI的目的替代CGI。CGI进程(类似Python解释器)针对每个请求进行创建,完成请求后退出。如果应程序接收树钱个请求,创建大量的语言解释器进程就会很快导致服务器宕机。

其目标在Web服务器与Web框架层之间提供一个通用的API标准,减少之间互操作性,并形成统一的调用方式。

WSGI 应用

根据WSGI的定义,其应用是可调用的对象,其参数固定为两个:

  • 含有服务器环境变量的字典
  • 可调用的对象, 该对象使用HTTP状态码和会返回客户端的HTTP头来初始化响应
1
2
3
4
5
def simple_wsgi_app(environment, start_response):
status = '200 OK'
headers = ['Content-Type': 'text/plain']
start_response(status, headers)
return ['Hello world!']

environment 包含一些环境变量,如HTTP_HOST, HTTP_USER, HTTP_AGENT, SERVER_PROTOCOL等。‘

start_response()是一个可调用的对象,必须在应用执行,生成最终发送回客户端的响应

werkzeug中start_response定义:

1
2
3
4
5
6
7
8
9
10
11
def start_response(status, response_headers, exc_info=None):
if exc_info:
try:
if headers_sent:
reraise(*exc_info)
finally:
exc_info = None
elif headers_set:
raise AssertionError('Headers already set')
headers_set[:] = [status, response_headers]
return write

WSGI服务器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import StringIO

def run_wsgi_app(app, environment):
body = StringIO.StringIO()

def start_response(status, headers):
body.write('Status: %s\r\n' % status)
for header in headers:
body.write('%s: %s\r\n' % header)
return body.write

iterable = app(environment, start_response)
try:
body.write('\r\n%s\r\n' % '\r\n'.join(line for line in iterable))
finally:
if hasattr(iterable, 'close') and callable(iterable.close):
iterable.close()

Flask WSGI 应用和服务实现

flask WSGI app

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
# Flask:wsgi_app

def wsgi_app(self, environ, start_response):
"""The actual WSGI application. This is not implemented in
`__call__` so that middlewares can be applied without losing a
reference to the class. So instead of doing this::

app = MyMiddleware(app)

It's a better idea to do this instead::

app.wsgi_app = MyMiddleware(app.wsgi_app)

Then you still have the original application object around and
can continue to call methods on it.

.. versionchanged:: 0.7
The behavior of the before and after request callbacks was changed
under error conditions and a new callback was added that will
always execute at the end of the request, independent on if an
error occurred or not. See :ref:`callbacks-and-errors`.

:param environ: a WSGI environment
:param start_response: a callable accepting a status code,
a list of headers and an optional
exception context to start the response
"""
ctx = self.request_context(environ)
ctx.push()
error = None
try:
try:
response = self.full_dispatch_request()
except Exception as e:
error = e
response = self.handle_exception(e)
return response(environ, start_response)
finally:
if self.should_ignore_error(error):
error = None
ctx.auto_pop(error)

# Flask:__call__

def __call__(self, environ, start_response):
"""Shortcut for :attr:`wsgi_app`."""
return self.wsgi_app(environ, start_response)

# BaseResponse:__call__

def __call__(self, environ, start_response):
"""Process this response as WSGI application.

:param environ: the WSGI environment.
:param start_response: the response callable provided by the WSGI
server.
:return: an application iterator
"""
app_iter, status, headers = self.get_wsgi_response(environ)
start_response(status, headers)
return app_iter

flask WSGI Server

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def run_wsgi(self):
if self.headers.get('Expect', '').lower().strip() == '100-continue':
self.wfile.write(b'HTTP/1.1 100 Continue\r\n\r\n')

self.environ = environ = self.make_environ()
headers_set = []
headers_sent = []

def write(data):
assert headers_set, 'write() before start_response'
if not headers_sent:
status, response_headers = headers_sent[:] = headers_set
try:
code, msg = status.split(None, 1)
except ValueError:
code, msg = status, ""
self.send_response(int(code), msg)
header_keys = set()
for key, value in response_headers:
self.send_header(key, value)
key = key.lower()
header_keys.add(key)
if 'content-length' not in header_keys:
self.close_connection = True
self.send_header('Connection', 'close')
if 'server' not in header_keys:
self.send_header('Server', self.version_string())
if 'date' not in header_keys:
self.send_header('Date', self.date_time_string())
self.end_headers()

assert isinstance(data, bytes), 'applications must write bytes'
self.wfile.write(data)
self.wfile.flush()

def start_response(status, response_headers, exc_info=None):
if exc_info:
try:
if headers_sent:
reraise(*exc_info)
finally:
exc_info = None
elif headers_set:
raise AssertionError('Headers already set')
headers_set[:] = [status, response_headers]
return write

def execute(app):
application_iter = app(environ, start_response)
try:
for data in application_iter:
write(data)
if not headers_sent:
write(b'')
finally:
if hasattr(application_iter, 'close'):
application_iter.close()
application_iter = None

try:
execute(self.server.app)
except (socket.error, socket.timeout) as e:
self.connection_dropped(e, environ)
except Exception:
if self.server.passthrough_errors:
raise
from werkzeug.debug.tbtools import get_current_traceback
traceback = get_current_traceback(ignore_system_exceptions=True)
try:
# if we haven't yet sent the headers but they are set
# we roll back to be able to set them again.
if not headers_sent:
del headers_set[:]
execute(InternalServerError())
except Exception:
pass
self.server.log('error', 'Error on request:\n%s',
traceback.plaintext)

References

Flask信号

Flask signals 默认没有自己实现signal,而是使用blink进行信号的定义,连接,分发。

Flask singal

flask未实现自己信号处理,而是使用blink.

1
2
3
4
5
6
7
8
signals_available = False
try:
from blinker import Namespace
signals_available = True
except ImportError:
class Namespace(object):
def signal(self, name, doc=None):
return _FakeSignal(name, doc)

flask 支持的信号

1
2
3
4
5
6
7
8
9
10
11
12
13
14
_signals = Namespace()

# Core signals. For usage examples grep the source code or consult
# the API documentation in docs/api.rst as well as docs/signals.rst
template_rendered = _signals.signal('template-rendered')
before_render_template = _signals.signal('before-render-template')
request_started = _signals.signal('request-started')
request_finished = _signals.signal('request-finished')
request_tearing_down = _signals.signal('request-tearing-down')
got_request_exception = _signals.signal('got-request-exception')
appcontext_tearing_down = _signals.signal('appcontext-tearing-down')
appcontext_pushed = _signals.signal('appcontext-pushed')
appcontext_popped = _signals.signal('appcontext-popped')
message_flashed = _signals.signal('message-flashed')

Blinker signal

blinker支持的特性:

  • a global registry of named signals
  • anonymous signals
  • custom name registries
  • permanently or temporarily connected receivers
  • automically disconnected receivers via weak referencing
  • sending arbirary data payloads
  • collecting return values from signal receivers
  • thread safety

Blinker signal sample

TODO

Blinker signal realization

TODO

注册

弱引用

线程安全

Problems

connecter weak reference

1
2
3
4
5
6
7
8
def register_signal_handlers(app):
import logging
from flask.signals import request_finished
_logger = logging.getLogger('api.debug')

def log_response(sender, response, **options):
print(response)
request_finished.connect(log_response, app)

以上代码,不能按原意正确的运行。因为log_response是局部作用域函数,在函数调用完成后,该作用域会消失,因此不能正确的调用log_response函数。

References